zzstoatzz.io
atproto relay implementation in zig
zlay.waow.tech
fix: restore resolve queue dedup to prevent unbounded memory growth
the queued_set dedup was accidentally removed in b96335c (LRU refactor).
without it, every cache miss for the same DID appends a fresh dupe to the
queue — 18K entries within 6 minutes. adds regression test.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+25
src/validator.zig
+25
src/validator.zig
···
99
99
self.allocator.free(did);
100
100
}
101
101
self.queue.deinit(self.allocator);
102
+
self.queued_set.deinit(self.allocator);
102
103
103
104
// free migration queue
104
105
for (self.migration_queue.items) |mc| {
···
392
393
393
394
self.queue_mutex.lock();
394
395
defer self.queue_mutex.unlock();
396
+
397
+
// skip if already queued (prevents unbounded queue growth)
398
+
if (self.queued_set.contains(duped)) {
399
+
self.allocator.free(duped);
400
+
return;
401
+
}
402
+
395
403
self.queue.append(self.allocator, duped) catch {
396
404
self.allocator.free(duped);
397
405
return;
398
406
};
407
+
self.queued_set.put(self.allocator, duped, {}) catch {};
399
408
self.queue_cond.signal();
400
409
}
401
410
···
414
423
}
415
424
if (self.queue.items.len > 0) {
416
425
did = self.queue.orderedRemove(0);
426
+
_ = self.queued_set.remove(did.?);
417
427
} else if (self.migration_queue.items.len > 0) {
418
428
migration = self.migration_queue.orderedRemove(0);
419
429
}
···
902
912
try std.testing.expect(result.valid);
903
913
try std.testing.expect(result.skipped);
904
914
}
915
+
916
+
test "queueResolve deduplicates repeated DIDs" {
917
+
var stats = broadcaster.Stats{};
918
+
var v = Validator.init(std.testing.allocator, &stats);
919
+
defer v.deinit();
920
+
921
+
// queue the same DID 100 times
922
+
for (0..100) |_| {
923
+
v.queueResolve("did:plc:duplicate");
924
+
}
925
+
926
+
// should have exactly 1 entry, not 100
927
+
try std.testing.expectEqual(@as(usize, 1), v.queue.items.len);
928
+
try std.testing.expectEqual(@as(u32, 1), v.queued_set.count());
929
+
}