From cf461ef2cf20f13744be3d8b91d166408f36b3e6 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 6 Aug 2026 22:55:24 -0500 Subject: [PATCH] tail: durable watermark pins eviction only, never reader visibility Upstream ReadableLog.ReadFrom serves entries to the appended tip; the durable watermark exists to pin undurable rows against eviction (they are the only copy until the archive fsyncs). Stream's tail additionally hid rows from readers until publishDurable, quantizing live delivery to the block-seal cadence (~13s bursts measured on /subscribe, vs the official instance's per-second flow). Remove the visibility gate: append wakes readers unconditionally, readBatch/next/indexForSeq serve resident rows, publishedTip/NextSeq/FloorSeq describe the resident log. Pinning and eviction semantics are unchanged. --- src/internal/ingest/ingest.zig | 4 +- src/internal/serve/tail.zig | 78 ++++++++++++---------------------- 2 files changed, 29 insertions(+), 53 deletions(-) diff --git a/src/internal/ingest/ingest.zig b/src/internal/ingest/ingest.zig index 5aee49a..c5cc276 100644 --- a/src/internal/ingest/ingest.zig +++ b/src/internal/ingest/ingest.zig @@ -504,8 +504,8 @@ pub const Consumer = struct { } try self.commitDurableMetadata(durable_seq, next_seq, upstream_seq); if (consumed > 0) self.pending_cursors.replaceRange(self.allocator, 0, consumed, &.{}) catch unreachable; - // Segment bytes and their cursor metadata are now durable. Only now - // may downstream subscribers observe the corresponding tail prefix. + // Segment bytes and their cursor metadata are now durable; the tail + // may release the corresponding pinned prefix to eviction. self.tail.publishDurable(durable_seq); } diff --git a/src/internal/serve/tail.zig b/src/internal/serve/tail.zig index 7a67d9d..a31a759 100644 --- a/src/internal/serve/tail.zig +++ b/src/internal/serve/tail.zig @@ -76,8 +76,10 @@ pub const Tail = struct { pinned_bytes: usize = 0, max_bytes: usize, closed: bool = false, - /// When enabled, archived entries remain private until their segment - /// bytes and matching metadata have crossed the durability hook. + /// When enabled, rows newer than the durable watermark are pinned against + /// eviction (upstream ReadableLog: the hot log is the only copy until the + /// archive fsyncs them). Visibility is not affected: readers always see + /// every appended row, matching upstream ReadFrom. durability_gated: bool = false, published_seq: u64 = 0, @@ -98,7 +100,9 @@ pub const Tail = struct { self.evictLocked(); } - /// Publish frames only after Archive's fsync + metadata transaction. + /// Advance the durable watermark after Archive's fsync + metadata + /// transaction. Unpins the newly durable rows so eviction may reclaim + /// them; does not change reader visibility (upstream advanceDurable). pub fn publishDurable(self: *Tail, durable_seq: u64) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); @@ -118,7 +122,6 @@ pub const Tail = struct { self.published_seq = durable_seq; } self.evictLocked(); - self.cond.broadcast(self.io); } pub fn deinit(self: *Tail) void { @@ -146,8 +149,7 @@ pub const Tail = struct { if (self.durability_gated and entry.seq != 0 and entry.seq > self.published_seq) self.pinned_bytes += entry_bytes; self.evictLocked(); - if (!self.durability_gated or entry.seq == 0 or entry.seq <= self.published_seq) - self.cond.broadcast(self.io); + self.cond.broadcast(self.io); } fn evictLocked(self: *Tail) void { @@ -220,46 +222,36 @@ pub const Tail = struct { return self.base + self.liveItems().len; } - /// Next index after the publicly visible prefix. A new live subscriber - /// starts here so rows appended but not durable at handshake time are - /// delivered once their durability boundary is published. + /// Next index after the retained log. A new live subscriber starts here, + /// matching upstream ReadFrom's at-tip semantics. pub fn publishedTip(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); - const live = self.liveItems(); - if (!self.durability_gated) return self.base + live.len; - var visible: usize = 0; - for (live) |entry| { - if (entry.seq != 0 and entry.seq > self.published_seq) break; - visible += 1; - } - return self.base + visible; + return self.base + self.liveItems().len; } - /// Next local archive seq after the visible hot prefix. Zero means no - /// seq-bearing row is visible yet, matching upstream's unwarmed writer + /// Next local archive seq after the resident hot rows. Zero means no + /// seq-bearing row is resident yet, matching upstream's unwarmed writer /// sentinel used by cursor resolution. pub fn publishedNextSeq(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); var next_seq: u64 = 0; for (self.liveItems()) |entry| { - if (self.durability_gated and entry.seq != 0 and entry.seq > self.published_seq) break; if (entry.seq != 0) next_seq = entry.seq +| 1; } return next_seq; } - /// First durable sequence still resident in the hot log. Cold readers - /// stop immediately before this sequence, then resolve it back to a hot + /// First sequence still resident in the hot log. Cold readers stop + /// immediately before this sequence, then resolve it back to a hot /// index. A moving floor may force another cold batch; it must never turn /// into a skipped range. pub fn publishedFloorSeq(self: *Tail) ?u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); for (self.liveItems()) |entry| { - if (entry.seq != 0 and (!self.durability_gated or entry.seq <= self.published_seq)) - return entry.seq; + if (entry.seq != 0) return entry.seq; } return null; } @@ -310,10 +302,6 @@ pub const Tail = struct { continue; } const e = live[rel]; - if (self.durability_gated and e.seq != 0 and e.seq > self.published_seq) { - self.cond.wait(self.io, &self.mutex) catch return null; - continue; - } idx.* += 1; if (!v2 and e.skip_v1 != .none) { _ = pred(ctx, e.kind, e.did, e.collection, e.skip_v1); @@ -350,16 +338,7 @@ pub const Tail = struct { continue; } - var visible: usize = 0; - while (rel + visible < live.len and visible < limit) : (visible += 1) { - const entry = live[rel + visible]; - if (self.durability_gated and entry.seq != 0 and entry.seq > self.published_seq) break; - } - if (visible == 0) { - self.cond.wait(self.io, &self.mutex) catch return .closed; - continue; - } - + const visible = @min(live.len - rel, limit); const result = try allocator.alloc(BatchEntry, visible); var initialized: usize = 0; errdefer { @@ -393,7 +372,6 @@ pub const Tail = struct { if (live.len == 0) return null; var saw_seq = false; for (live, 0..) |e, i| { - if (self.durability_gated and e.seq != 0 and e.seq > self.published_seq) break; // Non-archived diagnostic/control rows do not define the retained // sequence floor. A leading seq=0 row must not make an actually // too-old cursor look resident in the hot tail. @@ -410,13 +388,7 @@ pub const Tail = struct { // The caller must consult the archive rather than treating its tip as // a successful hot-cursor resolution. if (!saw_seq) return null; - if (!self.durability_gated) return self.base + live.len; - var visible: usize = 0; - for (live) |entry| { - if (entry.seq != 0 and entry.seq > self.published_seq) break; - visible += 1; - } - return self.base + visible; + return self.base + live.len; } /// wake all blocked readers without closing (e.g. so a reader can @@ -518,25 +490,29 @@ test "large eviction burst retires a prefix without per-entry shifts" { try testing.expect(t.entries.items.len < 20); } -test "durability gate hides appended frames until publication" { +test "appended frames are readable before durability publication" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit(); t.enableDurabilityGate(0); + // Upstream ReadFrom serves to the appended tip; durability only pins + // rows against eviction. A live reader must see the row immediately. try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none); try testing.expectEqual(@as(u64, 1), t.tip()); - try testing.expectEqual(@as(u64, 0), t.publishedTip()); - try testing.expectEqual(@as(?u64, null), t.indexForSeq(1)); - - t.publishDurable(1); try testing.expectEqual(@as(u64, 1), t.publishedTip()); + try testing.expectEqual(@as(?u64, 0), t.indexForSeq(1)); + var idx: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); const json = (try t.next(testing.allocator, &idx, &stop, false, {}, acceptAll)).?; defer testing.allocator.free(json); try testing.expectEqualStrings("{\"n\":1}", json); + try testing.expectEqual(t.gauges().bytes, t.gauges().pinned_bytes); + + t.publishDurable(1); + try testing.expectEqual(@as(usize, 0), t.gauges().pinned_bytes); } test "seq cursor below the hot floor ignores leading unarchived rows" { -- 2.51.2