atproto relay implementation in zig
zlay.waow.tech
event log gc: add size-based pruning to prevent disk exhaustion
time-based GC (72h retention) can't protect against replay storms
that generate 150+ GB in hours. add a max directory size cap
(RELAY_MAX_EVENTS_GB, default 100) that deletes oldest files first
when the limit is exceeded.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+81
-1
+78
src/event_log.zig
+78
src/event_log.zig
···
91
91
// config
92
92
events_per_file: u32 = default_events_per_file,
93
93
retention_hours: u64 = 72, // 3 days
94
+
max_dir_bytes: u64 = 100 * 1024 * 1024 * 1024, // 100 GB
94
95
95
96
// DID → UID cache (matches indigo's bidirectional ARC cache)
96
97
did_cache: lru.LruCache(u64),
···
794
795
if (expired.items.len > 0) {
795
796
log.info("gc: removed {d} expired log files", .{expired.items.len});
796
797
}
798
+
799
+
// size-based GC: delete oldest files until total size is under max_dir_bytes
800
+
if (self.max_dir_bytes > 0) {
801
+
try self.gcBySize();
802
+
}
803
+
}
804
+
805
+
/// delete oldest event files until total directory size is under max_dir_bytes
806
+
fn gcBySize(self: *DiskPersist) !void {
807
+
// query all refs ordered by creation time (oldest first)
808
+
var refs: std.ArrayListUnmanaged(GcSizeRef) = .{};
809
+
defer {
810
+
for (refs.items) |e| self.allocator.free(e.path);
811
+
refs.deinit(self.allocator);
812
+
}
813
+
814
+
var total_bytes: u64 = 0;
815
+
{
816
+
var result = try self.db.query(
817
+
"SELECT id, path FROM log_file_refs ORDER BY id ASC",
818
+
.{},
819
+
);
820
+
defer result.deinit();
821
+
while (result.nextUnsafe() catch null) |r| {
822
+
const path = r.get([]const u8, 1);
823
+
const stat = self.dir.statFile(path) catch continue;
824
+
total_bytes += stat.size;
825
+
try refs.append(self.allocator, .{
826
+
.id = r.get(i64, 0),
827
+
.path = try self.allocator.dupe(u8, path),
828
+
.size = stat.size,
829
+
});
830
+
}
831
+
}
832
+
833
+
if (total_bytes <= self.max_dir_bytes) return;
834
+
835
+
log.info("gc: event dir size {d} MB exceeds limit {d} MB, pruning oldest files", .{
836
+
total_bytes / (1024 * 1024),
837
+
self.max_dir_bytes / (1024 * 1024),
838
+
});
839
+
840
+
var removed: u64 = 0;
841
+
for (refs.items) |ref| {
842
+
if (total_bytes <= self.max_dir_bytes) break;
843
+
844
+
// never delete the current file
845
+
if (self.current_file_path) |cur| {
846
+
if (std.mem.eql(u8, ref.path, cur)) continue;
847
+
}
848
+
849
+
_ = self.db.exec("DELETE FROM log_file_refs WHERE id = $1", .{ref.id}) catch |err| {
850
+
log.warn("gc: failed to delete db record {d}: {s}", .{ ref.id, @errorName(err) });
851
+
continue;
852
+
};
853
+
854
+
self.dir.deleteFile(ref.path) catch |err| {
855
+
log.warn("gc: failed to delete {s}: {s}", .{ ref.path, @errorName(err) });
856
+
continue; // don't subtract size if file delete failed
857
+
};
858
+
859
+
total_bytes -= @min(total_bytes, ref.size);
860
+
removed += 1;
861
+
}
862
+
863
+
if (removed > 0) {
864
+
log.info("gc: size-based pruning removed {d} files, dir now {d} MB", .{
865
+
removed,
866
+
total_bytes / (1024 * 1024),
867
+
});
868
+
}
797
869
}
798
870
799
871
/// take down all events for a user (set flag + zero payload)
···
949
1021
const GcRef = struct {
950
1022
id: i64,
951
1023
path: []const u8,
1024
+
};
1025
+
1026
+
const GcSizeRef = struct {
1027
+
id: i64,
1028
+
path: []const u8,
1029
+
size: u64,
952
1030
};
953
1031
954
1032
// --- file-level operations ---
+3
-1
src/main.zig
+3
-1
src/main.zig
···
153
153
const upstream = normalizeSeedHost(std.posix.getenv("RELAY_UPSTREAM") orelse "bsky.network");
154
154
const data_dir = std.posix.getenv("RELAY_DATA_DIR") orelse "data/events";
155
155
const retention_hours = parseEnvInt(u64, "RELAY_RETENTION_HOURS", 72);
156
+
const max_events_gb = parseEnvInt(u64, "RELAY_MAX_EVENTS_GB", 100);
156
157
const frame_workers = parseEnvInt(u16, "FRAME_WORKERS", 16);
157
158
const frame_queue_capacity = parseEnvInt(u16, "FRAME_QUEUE_CAPACITY", 4096);
158
159
···
175
176
};
176
177
defer dp.deinit();
177
178
dp.retention_hours = retention_hours;
179
+
dp.max_dir_bytes = max_events_gb * 1024 * 1024 * 1024;
178
180
179
181
if (dp.lastSeq()) |last| {
180
182
log.info("event log recovered: last_seq={d}", .{last});
···
264
266
// start downstream WebSocket server (also serves HTTP API via httpFallback)
265
267
log.info("relay listening on :{d} (ws+http), :{d} (metrics)", .{ port, metrics_port });
266
268
log.info("seed host: {s}", .{upstream});
267
-
log.info("data dir: {s} (retention: {d}h)", .{ data_dir, retention_hours });
269
+
log.info("data dir: {s} (retention: {d}h, max: {d} GB)", .{ data_dir, retention_hours, max_events_gb });
268
270
269
271
var server = try websocket.Server(broadcaster.Handler).init(allocator, .{
270
272
.port = port,