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" {