diff --git a/src/broadcaster.zig b/src/broadcaster.zig index 6924129..d77db66 100644 --- a/src/broadcaster.zig +++ b/src/broadcaster.zig @@ -34,6 +34,7 @@ pub const Stats = struct { cache_misses: std.atomic.Value(u64) = .{ .raw = 0 }, slow_consumers: std.atomic.Value(u64) = .{ .raw = 0 }, connected_inbound: std.atomic.Value(u64) = .{ .raw = 0 }, + cache_evictions: std.atomic.Value(u64) = .{ .raw = 0 }, start_time: i64 = 0, }; @@ -593,6 +594,9 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, migrat \\# TYPE relay_validator_migration_queue gauge \\relay_validator_migration_queue {d} \\ + \\# TYPE relay_validator_cache_evictions_total counter + \\relay_validator_cache_evictions_total {d} + \\ , .{ stats.frames_in.load(.acquire), stats.frames_out.load(.acquire), @@ -611,6 +615,7 @@ pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, migrat uptime, cache_entries, migration_queue_len, + stats.cache_evictions.load(.acquire), }) catch return fbs.getWritten(); // linux-only process metrics from /proc @@ -789,6 +794,7 @@ test "formatPrometheusMetrics produces valid output" { try std.testing.expect(std.mem.indexOf(u8, output, "# TYPE relay_uptime_seconds gauge") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_entries 42") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_migration_queue 3") != null); + try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_evictions_total 0") != null); } test "formatStatsResponse produces valid JSON" { diff --git a/src/validator.zig b/src/validator.zig index 56ac09a..a53163b 100644 --- a/src/validator.zig +++ b/src/validator.zig @@ -61,6 +61,7 @@ pub const Validator = struct { queue_cond: std.Thread.Condition = .{}, resolver_threads: [max_resolver_threads]?std.Thread = .{null} ** max_resolver_threads, alive: std.atomic.Value(bool) = .{ .raw = true }, + max_cache_size: u32 = 500_000, const max_resolver_threads = 8; const default_resolver_threads = 4; @@ -109,6 +110,7 @@ pub const Validator = struct { /// start background resolver threads pub fn start(self: *Validator) !void { + self.max_cache_size = parseEnvInt(u32, "VALIDATOR_CACHE_SIZE", self.max_cache_size); const n = parseEnvInt(u8, "RESOLVER_THREADS", default_resolver_threads); const count = @min(n, max_resolver_threads); for (self.resolver_threads[0..count]) |*t| { @@ -445,6 +447,16 @@ pub const Validator = struct { }; @memcpy(cached.raw[0..public_key.raw.len], public_key.raw); + // evict oldest entries if cache is at capacity + const needs_eviction = blk: { + self.cache_mutex.lock(); + defer self.cache_mutex.unlock(); + break :blk self.cache.count() >= self.max_cache_size; + }; + if (needs_eviction) { + self.evictOldest(); + } + const did_duped = self.allocator.dupe(u8, d) catch continue; self.cache_mutex.lock(); @@ -545,6 +557,45 @@ pub const Validator = struct { defer self.queue_mutex.unlock(); return self.migration_queue.items.len; } + + /// evict the oldest 10% of cache entries by resolve_time. + /// called from resolveLoop when cache is at capacity. + fn evictOldest(self: *Validator) void { + self.cache_mutex.lock(); + defer self.cache_mutex.unlock(); + + const count = self.cache.count(); + if (count == 0) return; + const evict_count = @max(1, count / 10); + + // collect all entries with their resolve_times + const Entry = struct { key: []const u8, resolve_time: i64 }; + const entries = self.allocator.alloc(Entry, count) catch return; + defer self.allocator.free(entries); + + var i: usize = 0; + var it = self.cache.iterator(); + while (it.next()) |entry| { + entries[i] = .{ .key = entry.key_ptr.*, .resolve_time = entry.value_ptr.resolve_time }; + i += 1; + } + + // sort by resolve_time ascending (oldest first) + std.mem.sort(Entry, entries[0..i], {}, struct { + fn lessThan(_: void, a: Entry, b: Entry) bool { + return a.resolve_time < b.resolve_time; + } + }.lessThan); + + // evict the oldest entries + for (entries[0..evict_count]) |entry| { + if (self.cache.fetchRemove(entry.key)) |removed| { + self.allocator.free(removed.key); + } + } + + _ = self.stats.cache_evictions.fetchAdd(evict_count, .monotonic); + } }; /// extract hostname from a URL like "https://pds.example.com" or "https://pds.example.com:443/path" @@ -700,6 +751,40 @@ test "validateSync skips when no did field" { try std.testing.expectEqual(@as(u64, 1), stats.skipped.load(.acquire)); } +test "evictOldest removes oldest entries by resolve_time" { + var stats = broadcaster.Stats{}; + var v = Validator.init(std.testing.allocator, &stats); + v.max_cache_size = 5; + defer v.deinit(); + + // insert 5 entries with staggered resolve_times + const dids = [_][]const u8{ "did:plc:aaa", "did:plc:bbb", "did:plc:ccc", "did:plc:ddd", "did:plc:eee" }; + for (dids, 0..) |did, i| { + const key = try std.testing.allocator.dupe(u8, did); + v.cache.put(std.testing.allocator, key, .{ + .key_type = .p256, + .raw = .{0} ** 33, + .len = 33, + .resolve_time = @intCast(100 + i), // 100, 101, 102, 103, 104 + }) catch { + std.testing.allocator.free(key); + return error.TestFailed; + }; + } + try std.testing.expectEqual(@as(u32, 5), v.cache.count()); + + // evict — should remove oldest 10% = 1 entry (the one with resolve_time=100) + v.evictOldest(); + + try std.testing.expectEqual(@as(u32, 4), v.cache.count()); + // oldest entry ("did:plc:aaa" with resolve_time=100) should be gone + try std.testing.expect(v.cache.get("did:plc:aaa") == null); + // newest entries should still be present + try std.testing.expect(v.cache.get("did:plc:eee") != null); + try std.testing.expect(v.cache.get("did:plc:ddd") != null); + try std.testing.expectEqual(@as(u64, 1), stats.cache_evictions.load(.acquire)); +} + test "checkCommitStructure rejects too many ops" { var stats = broadcaster.Stats{}; var v = Validator.initWithConfig(std.testing.allocator, &stats, .{ .max_ops = 2 });