From c571d69146f31970c904b3b30b7cfd73b024de98 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 31 May 2026 14:56:09 -0500 Subject: [PATCH] seq rewind: gate broadcast on a committed-seq watermark, not in flushLocked MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 579807a fixed the rewind by broadcasting from flushLocked after the durable write, but that put the CBOR resequence under dp.mutex — every flush held the persist lock for ~400 frame-resequences, serializing all ~2670 producers. prod regressed: fr_out -25%, resolver queue 0->1280, live heap +3.6x (queued backlog), OOMs. operator rolled back to e77cb89. this keeps e77cb89's proven hot path (producers resequence + push in parallel, off dp.mutex, ordered by persist_order) and gates EMISSION on durability: - DiskPersist.committed_seq: atomic, advanced only by flushLocked after the write, to the highest seq just flushed. init seeds it to the recovered seq. - broadcaster fiber holds any queued frame whose seq > committed_seq (peekSeq + watermark compare), emitting only once a flush makes it durable. same invariant (last broadcast seq <= committed <= last durable), but the expensive resequence stays parallel and off the hot lock, so throughput equals e77cb89. queued frames now sit <=1 flush interval (<=100ms); steady-state broadcast_queue_depth rises to ~tens-hundreds (expected, not backpressure). regression test: "committed_seq advances only on durable flush". Co-Authored-By: Claude Opus 4.8 --- docs/incident-2026-05-31-seq-rewind.md | 46 +++++++++------- src/broadcaster.zig | 39 +++++++------- src/event_log.zig | 75 ++++++++++---------------- src/frame_worker.zig | 26 +++++++-- src/host_ops.zig | 17 +++++- src/main.zig | 6 --- src/subscriber.zig | 26 +++++++-- 7 files changed, 135 insertions(+), 100 deletions(-) diff --git a/docs/incident-2026-05-31-seq-rewind.md b/docs/incident-2026-05-31-seq-rewind.md index 00ecc94..e629161 100644 --- a/docs/incident-2026-05-31-seq-rewind.md +++ b/docs/incident-2026-05-31-seq-rewind.md @@ -46,24 +46,34 @@ zlay had relocated the broadcast into the persist callers (`frame_worker`, `subs queue *before* `flushLocked` wrote to disk. `flushLocked` did not broadcast at all. That single relocation is the entire defect. -## the fix (implemented) - -Restored indigo's contract: - -- `DiskPersist` gained `setBroadcaster(ctx, fn)` and broadcasts each event **from - `flushLocked`, after the durable write**, in seq order (`evtbuf` is append-ordered). -- The three producers no longer broadcast or resequence; they just `persist()`. seq order is - preserved by `persist()`'s own mutex, so the per-producer `persist_order` spinlock that - guarded broadcast ordering is no longer needed in the hot path (field kept for now; see - design notes). -- `broadcaster.broadcastPersisted` is the registered callback: it resequences to `relay_seq` - and enqueues for fan-out, invoked only post-flush. - -Regression test: `event_log.zig` "broadcast is gated on durable flush" — persists below the -flush threshold, asserts **zero** broadcasts, flushes, then asserts all events broadcast once -in seq order. (Also fixed the DB test harness to give each test an isolated database, since the -suite's runner executes tests concurrently against one server — they were contaminating each -other through the shared relay tables.) +## the fix (implemented) — a committed-seq watermark + +First attempt broadcast directly from `flushLocked` after the write. It was correct but +**regressed throughput**: the resequence (a frame-proportional CBOR decode+re-encode) ran under +`dp.mutex`, so every flush held the persist mutex for ~400 frame-resequences and serialized all +~2,670 producers behind it. In prod that dropped fr_out ~25%, backed the resolver queue up to +1280, ballooned live heap ~3.6× (queued backlog, not fragmentation), and OOMed. Rolled back. + +The shipped fix keeps e77cb89's proven-fast hot path and gates *emission* on durability: + +- producers resequence and `push` to the broadcast queue **in parallel, off `dp.mutex`** (exactly + as e77cb89 did — no throughput change). `persist_order` still serializes the push so queue + order == seq order. +- `flushLocked` does one cheap thing after the durable write: advance an atomic + `committed_seq` to the highest seq just written. No copy, no CBOR, negligible lock time. +- the broadcaster fiber **holds back any queued frame whose seq > `committed_seq`**, emitting it + only once a flush makes it durable (`peekSeq` + watermark compare in `runBroadcastLoop`). + +So a frame still can't reach consumers before it's on disk (last broadcast seq ≤ committed_seq ≤ +last durable seq), but the expensive resequence stays parallel and off the hot lock. Queued +frames now sit in the broadcast queue for up to one flush interval (≤100 ms), so steady-state +`relay_broadcast_queue_depth` rises to ~tens–hundreds — expected, not backpressure. + +Regression test: `event_log.zig` "committed_seq advances only on durable flush" — persists +below the flush threshold, asserts `committed_seq` stays 0 (frames held), flushes, asserts it +jumps to the highest durable seq. (Also fixed the DB test harness to give each test an isolated +database — the suite's runner executes tests concurrently against one server, so they were +contaminating each other through the shared relay tables.) ## still to harden: why the gap exceeded the flush bound diff --git a/src/broadcaster.zig b/src/broadcaster.zig index 1ba6bf1..707c6ce 100644 --- a/src/broadcaster.zig +++ b/src/broadcaster.zig @@ -280,22 +280,6 @@ pub fn resequenceFrame(allocator: Allocator, data: []const u8, relay_seq: u64) ? return result; } -/// callback registered with DiskPersist.setBroadcaster. resequences a persisted -/// frame to its relay seq and enqueues it for fan-out. invoked from flushLocked, -/// i.e. ONLY after the event is durably on disk — so an enqueued seq can never -/// exceed the on-disk seq. `payload` is owned by the persister and freed right -/// after this returns, so anything enqueued must be a fresh allocation. -pub fn broadcastPersisted(ctx: *anyopaque, seq: u64, payload: []const u8) void { - const self: *Broadcaster = @ptrCast(@alignCast(ctx)); - if (resequenceFrame(self.allocator, payload, seq)) |reseq| { - self.broadcast_queue.push(seq, reseq, &self.stats); - } else { - // decode failed — broadcast the raw frame verbatim (must be owned). - const owned = self.allocator.dupe(u8, payload) catch return; - self.broadcast_queue.push(seq, owned, &self.stats); - } -} - // --- broadcast queue (worker → fiber handoff) --- /// item produced by frame workers, consumed by the broadcaster fiber. @@ -371,6 +355,14 @@ pub const BroadcastQueue = struct { } } + /// peek the seq at the head without removing it (broadcaster fiber only). + /// used to gate emission on the durable watermark. null if empty. + pub fn peekSeq(self: *BroadcastQueue) ?u64 { + const head = self.head.load(.monotonic); + if (head == self.tail.load(.acquire)) return null; // empty + return self.items[head].seq; + } + /// pop an item (called by broadcaster fiber only). returns null if empty. pub fn pop(self: *BroadcastQueue) ?BroadcastItem { const head = self.head.load(.monotonic); @@ -761,14 +753,25 @@ pub const Broadcaster = struct { pub fn runBroadcastLoop(self: *Broadcaster) void { log.info("broadcaster fiber started", .{}); while (!self.shutdown.load(.acquire)) { + // only emit frames that are durably on disk: hold any whose seq is + // past the persister's committed watermark until the next flush + // advances it. without a persister (degenerate no-persist mode), + // nothing is gated. see docs/incident-2026-05-31-seq-rewind.md. + const committed: u64 = if (self.persist) |dp| + dp.committed_seq.load(.acquire) + else + std.math.maxInt(u64); + var drained: usize = 0; - while (self.broadcast_queue.pop()) |item| { + while (self.broadcast_queue.peekSeq()) |seq| { + if (seq > committed) break; // not durable yet — hold + const item = self.broadcast_queue.pop() orelse break; self.broadcast(item.seq, item.data); self.allocator.free(@constCast(item.data)); drained += 1; } if (drained == 0) { - // no items — yield briefly so other fibers run + // nothing emittable — yield briefly so other fibers run self.io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return; } } diff --git a/src/event_log.zig b/src/event_log.zig index c55c613..494f343 100644 --- a/src/event_log.zig +++ b/src/event_log.zig @@ -234,12 +234,12 @@ pub const DiskPersist = struct { flush_future: ?Io.Future(void) = null, alive: std.atomic.Value(bool) = .{ .raw = true }, - // live broadcaster, invoked from flushLocked AFTER the durable write (indigo's - // SetEventBroadcaster contract). null until wired in main. broadcasting only - // post-flush guarantees a broadcast seq never exceeds the on-disk seq, so the - // seq recovered on restart can't trail one already sent to consumers. - broadcast_ctx: ?*anyopaque = null, - broadcast_fn: ?*const fn (ctx: *anyopaque, seq: u64, payload: []const u8) void = null, + // highest seq that is durably on disk. advanced only by flushLocked, after the + // write. the broadcaster fiber refuses to emit a queued frame until its seq is + // <= committed_seq, so a frame never reaches consumers before it is durable — + // which means the seq recovered on restart can't trail an already-broadcast + // seq (the post-crash rewind). see docs/incident-2026-05-31-seq-rewind.md. + committed_seq: std.atomic.Value(u64) = .{ .raw = 0 }, io: Io, @@ -281,14 +281,6 @@ pub const DiskPersist = struct { return self.outbuf.capacity; } - /// register the live broadcaster. mirrors indigo's SetEventBroadcaster: - /// events are handed to it ONLY from flushLocked, after the durable write. - /// must be called before any producers start (no events in flight). - pub fn setBroadcaster(self: *DiskPersist, ctx: *anyopaque, f: *const fn (ctx: *anyopaque, seq: u64, payload: []const u8) void) void { - self.broadcast_ctx = ctx; - self.broadcast_fn = f; - } - pub fn init(allocator: Allocator, dir_path: []const u8, database_url: []const u8, db_pool_size: u16, io: Io) !DiskPersist { // ensure directory exists try Io.Dir.cwd().createDirPath(io, dir_path); @@ -398,6 +390,9 @@ pub const DiskPersist = struct { // recover from existing log files try self.resumeLog(); + // everything already on disk is committed; cur_seq points one past it. + self.committed_seq.store(self.cur_seq -| 1, .monotonic); + return self; } @@ -1229,15 +1224,15 @@ pub const DiskPersist = struct { // clear write buffer (bytes are now durable) self.outbuf.clearRetainingCapacity(); - // broadcast each event AFTER its durable write, in seq order (evtbuf is - // append-ordered = seq-ordered). this is the invariant that prevents the - // post-crash seq rewind: consumers never see a seq that isn't on disk. - // payload is the original frame (data after the fixed header); the - // broadcaster resequences it to job.seq. + // these events are now durable — publish the watermark so the broadcaster + // fiber may emit them. evtbuf is append-ordered, so the last entry holds + // the highest seq just written. this is the only place committed_seq moves. + if (self.evtbuf.items.len > 0) { + self.committed_seq.store(self.evtbuf.items[self.evtbuf.items.len - 1].seq, .release); + } + + // free job data for (self.evtbuf.items) |job| { - if (self.broadcast_fn) |bf| { - bf(self.broadcast_ctx.?, job.seq, job.data[header_size..]); - } self.allocator.free(job.data); } self.evtbuf.clearRetainingCapacity(); @@ -1542,11 +1537,12 @@ test "persist and playback" { try std.testing.expectEqualStrings("payload-three", entries.items[2].data); } -test "broadcast is gated on durable flush" { +test "committed_seq advances only on durable flush" { // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md): - // an event must NOT reach the broadcaster until it is durably written, so the - // seq recovered on restart can never trail an already-broadcast seq. matches - // indigo, which broadcasts only from flushLog. + // the broadcaster fiber only emits a frame once its seq <= committed_seq, and + // committed_seq advances ONLY here, after the durable write. so a frame can + // never reach consumers before it is on disk, and the seq recovered on restart + // can't trail an already-broadcast seq. const base_url = try requireDatabaseUrl(); var tdb = try TestDb.create(base_url); defer tdb.destroy(); @@ -1561,36 +1557,23 @@ test "broadcast is gated on durable flush" { var dp = try DiskPersist.init(std.testing.allocator, dir_path, database_url, 5, std.testing.io); defer dp.deinit(); - const Recorder = struct { - seqs: std.ArrayListUnmanaged(u64) = .empty, - fn cb(ctx: *anyopaque, seq: u64, payload: []const u8) void { - _ = payload; - const self: *@This() = @ptrCast(@alignCast(ctx)); - self.seqs.append(std.testing.allocator, seq) catch {}; - } - }; - var rec = Recorder{}; - defer rec.seqs.deinit(std.testing.allocator); - dp.setBroadcaster(&rec, Recorder.cb); - // unwire before rec is destroyed so deinit's final flush can't call back - defer dp.broadcast_fn = null; + // fresh log: nothing durable yet + try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire)); - // persist below the flush threshold — nothing broadcast yet + // persist below the flush threshold — assigned seqs but NOT yet committed, + // so the broadcaster would hold all three back. _ = try dp.persist(.commit, 1, "a"); _ = try dp.persist(.commit, 2, "b"); _ = try dp.persist(.commit, 3, "c"); - try std.testing.expectEqual(@as(usize, 0), rec.seqs.items.len); + try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire)); - // flush → all three broadcast exactly once, in seq order + // flush → committed_seq jumps to the highest durable seq, releasing all three { dp.mutex.lockUncancelable(dp.io); defer dp.mutex.unlock(dp.io); try dp.flushLocked(); } - try std.testing.expectEqual(@as(usize, 3), rec.seqs.items.len); - try std.testing.expectEqual(@as(u64, 1), rec.seqs.items[0]); - try std.testing.expectEqual(@as(u64, 2), rec.seqs.items[1]); - try std.testing.expectEqual(@as(u64, 3), rec.seqs.items[2]); + try std.testing.expectEqual(@as(u64, 3), dp.committed_seq.load(.acquire)); } test "playback with cursor" { diff --git a/src/frame_worker.zig b/src/frame_worker.zig index ee1d328..76f44a1 100644 --- a/src/frame_worker.zig +++ b/src/frame_worker.zig @@ -274,18 +274,34 @@ pub fn processFrame(work: *FrameWork) void { else .identity; - // persist the event. seq assignment and the live broadcast both happen inside - // DiskPersist now: persist() assigns seq under its mutex (so order is kept), and - // the broadcast fires from flushLocked AFTER the durable write — so a broadcast - // seq can never outrun the on-disk seq. see docs/incident-2026-05-31-seq-rewind.md. + // persist + resequence + enqueue under ordering lock to guarantee + // broadcast_queue insertion order matches seq assignment order. if (work.persist) |dp| { + var spins: u64 = 0; + while (work.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { + spins += 1; + std.atomic.spinLoopHint(); + } + if (spins > 0) { + _ = work.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic); + } + const relay_seq = dp.persist(kind, uid, data) catch |err| { + work.bc.persist_order.store(0, .release); log.warn("persist failed: {s}", .{@errorName(err)}); return; }; work.bc.stats.relay_seq.store(relay_seq, .release); - // update per-DID state (Postgres round-trip) + const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data; + const owned = work.allocator.dupe(u8, broadcast_data) catch { + work.bc.persist_order.store(0, .release); + return; + }; + work.bc.broadcast_queue.push(relay_seq, owned, &work.bc.stats); + work.bc.persist_order.store(0, .release); + + // update per-DID state outside the ordering lock (Postgres round-trip) if ((is_commit or is_sync) and uid > 0) { if (commit_rev) |rev| { const cid_str: []const u8 = if (commit_data_cid) |cid_raw| diff --git a/src/host_ops.zig b/src/host_ops.zig index f2ef796..0dd1448 100644 --- a/src/host_ops.zig +++ b/src/host_ops.zig @@ -313,12 +313,25 @@ pub const HostOpsQueue = struct { const frame_bytes = td.frameSlice(); const bc = self.bc orelse return; - // persist the #account event. broadcast fires from flushLocked after the - // durable write (see docs/incident-2026-05-31-seq-rewind.md). + // persist the #account event under ordering lock + while (bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { + std.atomic.spinLoopHint(); + } + if (self.persist.persist(.account, td.uid, frame_bytes)) |relay_seq| { bc.stats.relay_seq.store(relay_seq, .release); + + const broadcast_data = broadcaster.resequenceFrame(self.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes; + const owned = self.persist.allocator.dupe(u8, broadcast_data) catch { + bc.persist_order.store(0, .release); + log.warn("host_ops: failed to alloc broadcast data for takedown uid={d}", .{td.uid}); + return; + }; + bc.broadcast_queue.push(relay_seq, owned, &bc.stats); + bc.persist_order.store(0, .release); log.info("host_ops: emitted #account takedown for uid={d} (seq={d})", .{ td.uid, relay_seq }); } else |err| { + bc.persist_order.store(0, .release); log.warn("host_ops: persist #account takedown failed: {s}", .{@errorName(err)}); } } diff --git a/src/main.zig b/src/main.zig index a3f0436..b971aac 100644 --- a/src/main.zig +++ b/src/main.zig @@ -259,12 +259,6 @@ pub fn main() !void { bc.db_queue = &db_queue; val.persist = &dp; - // wire the live broadcaster into the persister: events are broadcast from - // flushLocked AFTER their durable write (indigo's contract), not by the - // producers pre-flush. set before any producer starts. see - // docs/incident-2026-05-31-seq-rewind.md. - dp.setBroadcaster(&bc, broadcaster.broadcastPersisted); - // init collection index (RocksDB — inspired by lightrail/microcosm.blue) const ci_dir = getenv("COLLECTION_INDEX_DIR") orelse "data/collection-index"; var ci = collection_index_mod.CollectionIndex.open(allocator, ci_dir) catch |err| { diff --git a/src/subscriber.zig b/src/subscriber.zig index af9fa8f..36b7a7b 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -738,18 +738,34 @@ const FrameHandler = struct { else // is_identity (unknown types already filtered above) .identity; - // persist the event. seq assignment + live broadcast both happen inside - // DiskPersist: persist() assigns seq under its mutex (order preserved), the - // broadcast fires from flushLocked AFTER the durable write — so a broadcast - // seq can never outrun the on-disk seq. see docs/incident-2026-05-31-seq-rewind.md. + // persist + resequence + enqueue under ordering lock to guarantee + // broadcast_queue insertion order matches seq assignment order. if (sub.persist) |dp| { + var spins: u64 = 0; + while (sub.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { + spins += 1; + std.atomic.spinLoopHint(); + } + if (spins > 0) { + _ = sub.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic); + } + const relay_seq = dp.persist(kind, uid, data) catch |err| { + sub.bc.persist_order.store(0, .release); log.warn("persist failed: {s}", .{@errorName(err)}); return; }; sub.bc.stats.relay_seq.store(relay_seq, .release); - // update per-DID state (Postgres round-trip) + const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data; + const owned = sub.allocator.dupe(u8, broadcast_data) catch { + sub.bc.persist_order.store(0, .release); + return; + }; + sub.bc.broadcast_queue.push(relay_seq, owned, &sub.bc.stats); + sub.bc.persist_order.store(0, .release); + + // update per-DID state outside the ordering lock (Postgres round-trip) if ((is_commit or is_sync) and uid > 0) { if (commit_rev) |rev| { const cid_str: []const u8 = if (commit_data_cid) |cid_raw| -- 2.51.2