milestone 25: client names learned over reverse dns

src/server/clients.zig

@@ -21,6 +21,7 @@
 const std = @import("std");
 
 const address = @import("../platform/address.zig");
+const client_names = @import("client_names.zig");
 const clients_repo = @import("../storage/repositories/clients_repo.zig");
 const db = @import("../storage/db.zig");
 const disk_monitor = @import("../storage/disk_monitor.zig");
@@ -135,6 +136,7 @@ pub const Tracker = struct {
         io: std.Io,
         database: *db.Db,
         monitor: ?*disk_monitor.Monitor,
+        names: ?*client_names.Resolver,
     ) std.Io.Cancelable!void {
         const interval: std.Io.Clock.Duration = .{
             .raw = .fromSeconds(flush_interval_s),
@@ -143,12 +145,18 @@ pub const Tracker = struct {
         while (true) {
             try interval.sleep(io);
             const writes_allowed = if (monitor) |m| m.writesAllowed() else true;
-            self.flushOnce(io, database, writes_allowed);
+            self.flushOnce(io, database, writes_allowed, names);
         }
     }
 
-    /// One pass: drain the table, write a row per client, and prune on every
-    /// `prune_every_passes`-th pass.
+    /// One pass: drain the table, write a row per client, prune on every
+    /// `prune_every_passes`-th pass, and then learn names for the rows that
+    /// have none (milestone-25 ruling 1).
+    ///
+    /// The order of the three steps is fixed and one `now_s` serves all three.
+    /// Resolving before pruning would spend PTR queries on rows the same pass
+    /// deletes; resolving before the drain would make a slow resolver delay the
+    /// writes the pass exists for.
     ///
     /// A gated pass does nothing at all, not even count: the work it skipped is
     /// still owed, and the pending addresses it leaves behind are re-tracked by
@@ -161,9 +169,17 @@ pub const Tracker = struct {
     /// Only `run` may call this concurrently with itself — the drain buffer is
     /// this call's stack, but the pass counter and the prune schedule assume a
     /// single caller.
-    pub fn flushOnce(self: *Tracker, io: std.Io, database: *db.Db, writes_allowed: bool) void {
+    pub fn flushOnce(
+        self: *Tracker,
+        io: std.Io,
+        database: *db.Db,
+        writes_allowed: bool,
+        names: ?*client_names.Resolver,
+    ) void {
+        // A gated pass skips naming too: naming writes.
         if (!writes_allowed) return;
 
+        const now_s = std.Io.Clock.real.now(io).toSeconds();
         var drained: [max_pending]Pending = undefined;
         const batch = self.drain(io, &drained);
 
@@ -193,18 +209,21 @@ pub const Tracker = struct {
         const due = self.passes % prune_every_passes == 0;
         self.mutex.unlock(io);
 
-        if (!due) return;
-        const cutoff = std.Io.Clock.real.now(io).toSeconds() - @as(i64, self.retention_days) * 86_400;
-        if (clients_repo.pruneStale(database, cutoff)) |deleted| {
-            self.mutex.lockUncancelable(io);
-            self.stats.pruned += deleted;
-            self.mutex.unlock(io);
-        } else |err| {
-            log.warn("pruning clients before {d} failed: {s}", .{ cutoff, @errorName(err) });
-            self.mutex.lockUncancelable(io);
-            self.stats.flush_failures += 1;
-            self.mutex.unlock(io);
+        if (due) {
+            const cutoff = now_s - @as(i64, self.retention_days) * 86_400;
+            if (clients_repo.pruneStale(database, cutoff)) |deleted| {
+                self.mutex.lockUncancelable(io);
+                self.stats.pruned += deleted;
+                self.mutex.unlock(io);
+            } else |err| {
+                log.warn("pruning clients before {d} failed: {s}", .{ cutoff, @errorName(err) });
+                self.mutex.lockUncancelable(io);
+                self.stats.flush_failures += 1;
+                self.mutex.unlock(io);
+            }
         }
+
+        if (names) |resolver| resolver.runPass(io, database, now_s);
     }
 
     pub fn snapshotStats(self: *Tracker, io: std.Io) Stats {
@@ -277,7 +296,7 @@ test "a client tracked twice before a flush yields one row at the later time" {
     try testing.expectEqual(@as(u64, 0), tracker.trackAt(io, parsed("192.168.1.10"), 1700000030));
     try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
 
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(i64, 1700000030), try lastSeen(&database, "192.168.1.10"));
@@ -305,7 +324,7 @@ test "distinct clients each get a row and ipv6 text is canonical" {
     _ = tracker.trackAt(io, address.NetAddress.fromIp(try std.Io.net.IpAddress.parse("::ffff:192.168.1.10", 53)), 1700000003);
     try testing.expectEqual(@as(u32, 3), tracker.pendingClients(io));
 
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 3), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(i64, 1700000003), try lastSeen(&database, "192.168.1.10"));
@@ -343,7 +362,7 @@ test "a full table drops further clients and counts them" {
     // A tracked client still refreshes while the table is full, and the flush
     // makes room for the next newcomer.
     _ = tracker.trackAt(io, .{ .ip4 = .{ 0, 0, 0, 0 } }, 1700000060);
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
     try testing.expectEqual(@as(i64, Tracker.max_pending), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(i64, 1700000060), try lastSeen(&database, "0.0.0.0"));
 
@@ -366,7 +385,7 @@ test "a flush touches a hand-edited row without changing what the operator set"
 
     var tracker: Tracker = .init(30);
     _ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(i64, 1700000000), try lastSeen(&database, "192.168.1.10"));
@@ -394,7 +413,7 @@ test "a gated pass writes nothing and keeps the pending clients" {
 
     var tracker: Tracker = .init(30);
     _ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
-    tracker.flushOnce(io, &database, monitor.writesAllowed());
+    tracker.flushOnce(io, &database, monitor.writesAllowed(), null);
 
     try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
@@ -403,7 +422,7 @@ test "a gated pass writes nothing and keeps the pending clients" {
 
     // Free space recovers and the same pending client lands.
     monitor.state_raw.store(@intFromEnum(disk_monitor.State.warn), .monotonic);
-    tracker.flushOnce(io, &database, monitor.writesAllowed());
+    tracker.flushOnce(io, &database, monitor.writesAllowed(), null);
     try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(u64, 1), tracker.passes);
 }
@@ -423,7 +442,7 @@ test "a failing upsert counts and leaves the client to be tracked again" {
     var tracker: Tracker = .init(30);
     _ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
     _ = tracker.trackAt(io, parsed("192.168.1.11"), 1700000000);
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
     const stats = tracker.snapshotStats(io);
@@ -433,7 +452,7 @@ test "a failing upsert counts and leaves the client to be tracked again" {
 
     try database.exec("DROP TRIGGER refuse_insert;");
     _ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000060);
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
     try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
 }
 
@@ -453,12 +472,12 @@ test "the pass that comes due prunes the clients that went quiet" {
     var tracker: Tracker = .init(30);
     // Every pass before the due one leaves both rows alone.
     for (0..Tracker.prune_every_passes - 1) |_| {
-        tracker.flushOnce(io, &database, true);
+        tracker.flushOnce(io, &database, true, null);
     }
     try testing.expectEqual(@as(i64, 2), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(u64, 0), tracker.snapshotStats(io).pruned);
 
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(i64, now - 29 * day), try lastSeen(&database, "10.0.0.2"));
@@ -479,7 +498,7 @@ test "a shorter retention prunes what the default keeps" {
 
     var tracker: Tracker = .init(1);
     tracker.passes = Tracker.prune_every_passes - 1;
-    tracker.flushOnce(io, &database, true);
+    tracker.flushOnce(io, &database, true, null);
 
     try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(u64, 1), tracker.snapshotStats(io).pruned);
@@ -501,6 +520,7 @@ test "the run loop flushes on its interval and returns on cancel" {
         io,
         &database,
         @as(?*disk_monitor.Monitor, null),
+        @as(?*client_names.Resolver, null),
     });
     // The first flush is one interval away, so cancelling immediately proves the
     // loop starts by sleeping rather than by writing.
@@ -508,3 +528,124 @@ test "the run loop flushes on its interval and returns on cancel" {
     try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
     try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
 }
+
+// --- the naming step's place in the pass (milestone-25 ruling 1) ------------
+
+const forward_zones = @import("../local/forward_zones.zig");
+const local_tables = @import("local_tables.zig");
+const validate = @import("../config/validate.zig");
+
+/// Counts exchanges and times out on every one, and records how many client
+/// rows existed when the first exchange was attempted.
+const CountingExchange = struct {
+    var calls: usize = 0;
+    var database: ?*db.Db = null;
+    var rows_at_first_call: i64 = -1;
+
+    fn reset(target: *db.Db) void {
+        calls = 0;
+        database = target;
+        rows_at_first_call = -1;
+    }
+
+    fn exchange(
+        _: std.Io,
+        _: validate.Resolver,
+        _: []const u8,
+        _: []u8,
+    ) @import("../upstream/transport.zig").ExchangeError![]u8 {
+        if (calls == 0) {
+            rows_at_first_call = clients_repo.countClients(database.?) catch -1;
+        }
+        calls += 1;
+        return error.Timeout;
+    }
+};
+
+fn namingTables(io: std.Io, tables: *local_tables.LocalTables) !void {
+    tables.swap(io, testing.allocator, .empty, try forward_zones.Zones.build(
+        testing.allocator,
+        &.{.{ .zone = "168.192.in-addr.arpa", .resolver = "udp://192.168.1.1:53" }},
+    ));
+}
+
+test "the drain lands before any exchange, and attempts stop at the cap" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openMigrated();
+    defer database.close();
+    var tables: local_tables.LocalTables = .empty;
+    defer tables.deinit(testing.allocator);
+    try namingTables(io, &tables);
+
+    var names: client_names.Resolver = .init(&tables);
+    names.exchange_fn = CountingExchange.exchange;
+    CountingExchange.reset(&database);
+
+    var tracker: Tracker = .init(30);
+    const pending = clients_repo.max_per_pass + 4;
+    for (0..pending) |i| {
+        _ = tracker.trackAt(io, .{ .ip4 = .{ 192, 168, 2, @intCast(i) } }, 1700000000);
+    }
+
+    tracker.flushOnce(io, &database, true, &names);
+
+    // Every pending row was written before the first exchange went out, which
+    // is what keeps a slow resolver off the drain.
+    try testing.expectEqual(@as(i64, @intCast(pending)), CountingExchange.rows_at_first_call);
+    try testing.expectEqual(clients_repo.max_per_pass, CountingExchange.calls);
+    try testing.expectEqual(@as(u64, @intCast(pending)), tracker.snapshotStats(io).flushed);
+    try testing.expectEqual(@as(u64, clients_repo.max_per_pass), names.snapshotStats(io).failed);
+}
+
+test "a row the due pass prunes is never asked about" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openMigrated();
+    defer database.close();
+    var tables: local_tables.LocalTables = .empty;
+    defer tables.deinit(testing.allocator);
+    try namingTables(io, &tables);
+
+    var names: client_names.Resolver = .init(&tables);
+    names.exchange_fn = CountingExchange.exchange;
+    CountingExchange.reset(&database);
+
+    const now = std.Io.Clock.real.now(io).toSeconds();
+    try clients_repo.upsertSeen(&database, "192.168.1.10", now - 40 * 86_400);
+
+    var tracker: Tracker = .init(30);
+    tracker.passes = Tracker.prune_every_passes - 1;
+    tracker.flushOnce(io, &database, true, &names);
+
+    try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
+    try testing.expectEqual(@as(usize, 0), CountingExchange.calls);
+    try testing.expectEqual(@as(u64, 0), names.snapshotStats(io).attempted);
+}
+
+test "a gated pass attempts no naming either" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var database = try openMigrated();
+    defer database.close();
+    var tables: local_tables.LocalTables = .empty;
+    defer tables.deinit(testing.allocator);
+    try namingTables(io, &tables);
+
+    var names: client_names.Resolver = .init(&tables);
+    names.exchange_fn = CountingExchange.exchange;
+    CountingExchange.reset(&database);
+
+    try clients_repo.upsertSeen(&database, "192.168.1.10", 1700000000);
+    var tracker: Tracker = .init(30);
+    tracker.flushOnce(io, &database, false, &names);
+
+    try testing.expectEqual(@as(usize, 0), CountingExchange.calls);
+    try testing.expectEqual(@as(u64, 0), names.snapshotStats(io).attempted);
+}