milestone 19: hygiene sweep - dead ecs surface, single-source constants, tls classification, frontend state hazards, docker smoke network fix
This commit is contained in:
+35
-27
@@ -24,13 +24,10 @@ const address = @import("../platform/address.zig");
|
||||
const clients_repo = @import("../storage/repositories/clients_repo.zig");
|
||||
const db = @import("../storage/db.zig");
|
||||
const disk_monitor = @import("../storage/disk_monitor.zig");
|
||||
const logger = @import("../storage/logger.zig");
|
||||
|
||||
const log = std.log.scoped(.clients);
|
||||
|
||||
/// RFC 5952 text of any IPv6 address. `format` never writes more than this, so
|
||||
/// the formatting in `flushOnce` cannot fail.
|
||||
const max_ip_text = 45;
|
||||
|
||||
/// One address waiting for its row, with the wall-clock second of its most
|
||||
/// recent query.
|
||||
const Pending = struct {
|
||||
@@ -83,16 +80,21 @@ pub const Tracker = struct {
|
||||
/// Records `addr` as seen now. Called from the query path, so it writes no
|
||||
/// database and returns no error: a full table drops the address.
|
||||
///
|
||||
/// Returns `dropped_full` as it stands after this call. The query path
|
||||
/// mirrors that counter into its own atomic, and reporting it from here
|
||||
/// costs the caller nothing — the mutex is already held — where a second
|
||||
/// `snapshotStats` call would take it again on every query.
|
||||
///
|
||||
/// `lockUncancelable` rather than `lock`: the caller is `Handler.handle`,
|
||||
/// which has no error union to carry `error.Canceled` out of. The critical
|
||||
/// section is a scan of at most `max_pending` addresses and holds no I/O.
|
||||
pub fn track(self: *Tracker, io: std.Io, addr: address.NetAddress) void {
|
||||
self.trackAt(io, addr, std.Io.Clock.real.now(io).toSeconds());
|
||||
pub fn track(self: *Tracker, io: std.Io, addr: address.NetAddress) u64 {
|
||||
return self.trackAt(io, addr, std.Io.Clock.real.now(io).toSeconds());
|
||||
}
|
||||
|
||||
/// `track` with the timestamp supplied, so a test does not depend on the
|
||||
/// wall clock.
|
||||
pub fn trackAt(self: *Tracker, io: std.Io, addr: address.NetAddress, now_s: i64) void {
|
||||
pub fn trackAt(self: *Tracker, io: std.Io, addr: address.NetAddress, now_s: i64) u64 {
|
||||
self.mutex.lockUncancelable(io);
|
||||
defer self.mutex.unlock(io);
|
||||
|
||||
@@ -100,15 +102,16 @@ pub const Tracker = struct {
|
||||
if (!entry.addr.eql(addr)) continue;
|
||||
entry.last_seen = now_s;
|
||||
self.stats.tracked += 1;
|
||||
return;
|
||||
return self.stats.dropped_full;
|
||||
}
|
||||
if (self.count == max_pending) {
|
||||
self.stats.dropped_full += 1;
|
||||
return;
|
||||
return self.stats.dropped_full;
|
||||
}
|
||||
self.pending[self.count] = .{ .addr = addr, .last_seen = now_s };
|
||||
self.count += 1;
|
||||
self.stats.tracked += 1;
|
||||
return self.stats.dropped_full;
|
||||
}
|
||||
|
||||
/// Flush loop, first flush one interval in: an empty table at startup has
|
||||
@@ -167,7 +170,9 @@ pub const Tracker = struct {
|
||||
var flushed: u64 = 0;
|
||||
var failures: u64 = 0;
|
||||
for (batch) |entry| {
|
||||
var buf: [max_ip_text]u8 = undefined;
|
||||
// `logger.max_client_len` is the RFC 5952 bound every address text
|
||||
// in this program is sized by, so `format` cannot fail here.
|
||||
var buf: [logger.max_client_len]u8 = undefined;
|
||||
var w: std.Io.Writer = .fixed(&buf);
|
||||
entry.addr.format(&w) catch unreachable;
|
||||
|
||||
@@ -267,8 +272,9 @@ test "a client tracked twice before a flush yields one row at the later time" {
|
||||
defer database.close();
|
||||
|
||||
var tracker: Tracker = .init(30);
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000030);
|
||||
// A table with room reports no drops.
|
||||
try testing.expectEqual(@as(u64, 0), tracker.trackAt(io, parsed("192.168.1.10"), 1700000000));
|
||||
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);
|
||||
@@ -292,11 +298,11 @@ test "distinct clients each get a row and ipv6 text is canonical" {
|
||||
defer database.close();
|
||||
|
||||
var tracker: Tracker = .init(30);
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
tracker.trackAt(io, parsed("192.168.1.11"), 1700000001);
|
||||
tracker.trackAt(io, parsed("fd00:0:0:0:0:0:0:1"), 1700000002);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.11"), 1700000001);
|
||||
_ = tracker.trackAt(io, parsed("fd00:0:0:0:0:0:0:1"), 1700000002);
|
||||
// An IPv4-mapped literal is the same client as its plain form.
|
||||
tracker.trackAt(io, address.NetAddress.fromIp(try std.Io.net.IpAddress.parse("::ffff:192.168.1.10", 53)), 1700000003);
|
||||
_ = 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);
|
||||
@@ -319,13 +325,15 @@ test "a full table drops further clients and counts them" {
|
||||
for (0..Tracker.max_pending) |i| {
|
||||
var octets: [4]u8 = undefined;
|
||||
std.mem.writeInt(u32, &octets, @intCast(i), .big);
|
||||
tracker.trackAt(io, .{ .ip4 = octets }, 1700000000);
|
||||
_ = tracker.trackAt(io, .{ .ip4 = octets }, 1700000000);
|
||||
}
|
||||
try testing.expectEqual(@as(u32, Tracker.max_pending), tracker.pendingClients(io));
|
||||
try testing.expectEqual(@as(u64, 0), tracker.snapshotStats(io).dropped_full);
|
||||
|
||||
tracker.trackAt(io, parsed("203.0.113.7"), 1700000000);
|
||||
tracker.trackAt(io, parsed("203.0.113.8"), 1700000000);
|
||||
// The return value is what the handler mirrors, so it must agree with
|
||||
// `snapshotStats` without a second lock acquisition.
|
||||
try testing.expectEqual(@as(u64, 1), tracker.trackAt(io, parsed("203.0.113.7"), 1700000000));
|
||||
try testing.expectEqual(@as(u64, 2), tracker.trackAt(io, parsed("203.0.113.8"), 1700000000));
|
||||
try testing.expectEqual(@as(u32, Tracker.max_pending), tracker.pendingClients(io));
|
||||
|
||||
const stats = tracker.snapshotStats(io);
|
||||
@@ -334,12 +342,12 @@ 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.trackAt(io, .{ .ip4 = .{ 0, 0, 0, 0 } }, 1700000060);
|
||||
tracker.flushOnce(io, &database, true);
|
||||
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"));
|
||||
|
||||
tracker.trackAt(io, parsed("203.0.113.7"), 1700000060);
|
||||
_ = tracker.trackAt(io, parsed("203.0.113.7"), 1700000060);
|
||||
try testing.expectEqual(@as(u32, 1), tracker.pendingClients(io));
|
||||
}
|
||||
|
||||
@@ -357,7 +365,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.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
tracker.flushOnce(io, &database, true);
|
||||
|
||||
try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
|
||||
@@ -385,7 +393,7 @@ test "a gated pass writes nothing and keeps the pending clients" {
|
||||
try testing.expect(!monitor.writesAllowed());
|
||||
|
||||
var tracker: Tracker = .init(30);
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
tracker.flushOnce(io, &database, monitor.writesAllowed());
|
||||
|
||||
try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
|
||||
@@ -413,8 +421,8 @@ 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.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.11"), 1700000000);
|
||||
tracker.flushOnce(io, &database, true);
|
||||
|
||||
try testing.expectEqual(@as(i64, 0), try clients_repo.countClients(&database));
|
||||
@@ -424,7 +432,7 @@ test "a failing upsert counts and leaves the client to be tracked again" {
|
||||
try testing.expectEqual(@as(u32, 0), tracker.pendingClients(io));
|
||||
|
||||
try database.exec("DROP TRIGGER refuse_insert;");
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000060);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000060);
|
||||
tracker.flushOnce(io, &database, true);
|
||||
try testing.expectEqual(@as(i64, 1), try clients_repo.countClients(&database));
|
||||
}
|
||||
@@ -486,7 +494,7 @@ test "the run loop flushes on its interval and returns on cancel" {
|
||||
defer database.close();
|
||||
|
||||
var tracker: Tracker = .init(30);
|
||||
tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
_ = tracker.trackAt(io, parsed("192.168.1.10"), 1700000000);
|
||||
|
||||
var future = try io.concurrent(Tracker.run, .{
|
||||
&tracker,
|
||||
|
||||
Reference in New Issue
Block a user