From c8efaa16a782c575f2f8c8881182220e4bb3ee8e Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sat, 4 Apr 2026 14:09:23 -0500 Subject: [PATCH] narrow persist_order to cover only dp.persist() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit move resequenceFrame, heap dupe, and broadcast_queue.push() outside the ordering lock. persist_order now covers only the DB persist call and seq store — the minimum needed for monotonic sequence assignment. this eliminates the cascade where producers spin on persist_order while another producer is blocked in a full broadcast_queue.push(). slight out-of-order in the ring is acceptable — seq is embedded in frame data and consumers/history track by seq. metrics showed persist_order_spins_total dominating at ~1,100 hosts (548M spins) while push_lock_spins was zero — confirming the critical section width was the bottleneck. Co-Authored-By: Claude Opus 4.6 --- src/api/admin.zig | 8 +++----- src/frame_worker.zig | 18 +++++++----------- src/subscriber.zig | 15 ++++++--------- 3 files changed, 16 insertions(+), 25 deletions(-) diff --git a/src/api/admin.zig b/src/api/admin.zig index 415f8a0..cffaedf 100644 --- a/src/api/admin.zig +++ b/src/api/admin.zig @@ -74,9 +74,7 @@ pub fn handleBan(conn: *h.Conn, body: []const u8, headers: *const websocket.Hand log.debug("collection removeAll after ban failed: {s}", .{@errorName(err)}); }; - // emit #account event — same ordered publication path as workers: - // persist_order spinlock → dp.persist → resequence → queue.push. - // one publication path for all relay-sequenced events. + // emit #account event — persist under narrow ordering lock, resequence + enqueue outside. if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| { while (ctx.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { std.atomic.spinLoopHint(); @@ -84,15 +82,15 @@ pub fn handleBan(conn: *h.Conn, body: []const u8, headers: *const websocket.Hand if (ctx.persist.persist(.account, uid, frame_bytes)) |relay_seq| { ctx.bc.stats.relay_seq.store(relay_seq, .release); + ctx.bc.persist_order.store(0, .release); + const broadcast_data = broadcaster.resequenceFrame(ctx.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes; const owned = ctx.persist.allocator.dupe(u8, broadcast_data) catch { - ctx.bc.persist_order.store(0, .release); log.warn("admin: failed to alloc broadcast data for {s}", .{did}); h.respondJson(conn, .ok, "{\"success\":true}"); return; }; ctx.bc.broadcast_queue.push(relay_seq, owned, &ctx.bc.stats); - ctx.bc.persist_order.store(0, .release); log.info("admin: emitted #account takedown event for {s} (seq={d})", .{ did, relay_seq }); } else |err| { ctx.bc.persist_order.store(0, .release); diff --git a/src/frame_worker.zig b/src/frame_worker.zig index 79d369c..7ae227e 100644 --- a/src/frame_worker.zig +++ b/src/frame_worker.zig @@ -273,9 +273,9 @@ pub fn processFrame(work: *FrameWork) void { else .identity; - // persist under ordering lock, then push to broadcast queue. - // the broadcaster fiber (Evented) drains the queue and does fan-out — - // worker threads never touch consumer state directly. + // persist under narrow ordering lock (seq assignment only), then + // resequence + enqueue outside the lock. slight out-of-order in the + // ring is fine — seq is embedded in frame data and consumers track by seq. if (work.persist) |dp| { const relay_seq = blk: { var spins: u64 = 0; @@ -293,17 +293,13 @@ pub fn processFrame(work: *FrameWork) void { return; }; work.bc.stats.relay_seq.store(seq, .release); - const broadcast_data = broadcaster.resequenceFrame(alloc, data, seq) orelse data; - // dupe for the broadcast queue — arena will free broadcast_data - const owned = work.allocator.dupe(u8, broadcast_data) catch { - work.bc.persist_order.store(0, .release); - return; - }; - work.bc.broadcast_queue.push(seq, owned, &work.bc.stats); work.bc.persist_order.store(0, .release); break :blk seq; }; - _ = relay_seq; + + const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data; + const owned = work.allocator.dupe(u8, broadcast_data) catch return; + work.bc.broadcast_queue.push(relay_seq, owned, &work.bc.stats); // update per-DID state outside the ordering lock (Postgres round-trip) if ((is_commit or is_sync) and uid > 0) { diff --git a/src/subscriber.zig b/src/subscriber.zig index b2a0194..70f5cb4 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -709,8 +709,8 @@ const FrameHandler = struct { else // is_identity (unknown types already filtered above) .identity; - // persist and push to broadcast queue (broadcaster fiber handles fan-out). - // ordering spinlock ensures frames are persisted + enqueued in seq order. + // persist under narrow ordering lock (seq assignment only), then + // resequence + enqueue outside the lock. if (sub.persist) |dp| { const relay_seq = blk: { var spins: u64 = 0; @@ -728,16 +728,13 @@ const FrameHandler = struct { return; }; sub.bc.stats.relay_seq.store(seq, .release); - const broadcast_data = broadcaster.resequenceFrame(alloc, data, 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(seq, owned, &sub.bc.stats); sub.bc.persist_order.store(0, .release); break :blk seq; }; - _ = relay_seq; + + const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data; + const owned = sub.allocator.dupe(u8, broadcast_data) catch return; + sub.bc.broadcast_queue.push(relay_seq, owned, &sub.bc.stats); // update per-DID state outside the ordering lock (Postgres round-trip) if ((is_commit or is_sync) and uid > 0) { -- 2.51.2