diff --git a/src/internal/storage/cold.zig b/src/internal/storage/cold.zig index a5f7dbb..51002a7 100644 --- a/src/internal/storage/cold.zig +++ b/src/internal/storage/cold.zig @@ -381,9 +381,16 @@ pub fn resolveTimeToSeq( break; } } - if (candidate == null) return .{ - .seq = metas[metas.len - 1].header.max_seq +| 1, - }; + if (candidate == null) { + // The time falls in the not-yet-sealed window. Upstream anchors at + // sealed_max+1 and its replay walks the active segment efficiently; + // our cold walk re-reads the active region on every pass, so a + // subscriber anchored up to a whole open segment back lands on a + // treadmill that cannot outrun live ingest (2026-08-08 boot-window + // consumers). Resolve INTO the active region's fsynced blocks: one + // sequential walk, at connect time only. + return resolveTimeInActive(allocator, io, archive, time_us, metas[metas.len - 1].header.max_seq +| 1); + } const meta = candidate.?; var name_buffer: [64]u8 = undefined; @@ -412,6 +419,56 @@ pub fn resolveTimeToSeq( return .{ .seq = @max(entry.max_seq, 1) }; } +/// First fsynced active-region sequence witnessed at/after `time_us`. +/// Falls back to `sealed_fallback` when the active region cannot answer +/// (no active file, no fsynced frames); when every fsynced row is older +/// than the requested time, the answer is the next not-yet-durable seq — +/// the true near-now anchor. +fn resolveTimeInActive( + allocator: Allocator, + io: Io, + archive: *archive_mod.Archive, + time_us: i64, + sealed_fallback: u64, +) !ResolvedCursor { + var snapshot = try archive.coldSnapshot(allocator); + defer snapshot.deinit(); + var name_buf: [64]u8 = undefined; + const name = archive_mod.formatSegmentName(&name_buf, snapshot.active_index); + var file = archive.dir.openFile(io, name, .{}) catch return .{ .seq = sealed_fallback }; + defer file.close(io); + + var next_after: u64 = sealed_fallback; + var offset: usize = segment.header_size; + while (offset < snapshot.active_length) { + var block_arena = std.heap.ArenaAllocator.init(allocator); + defer block_arena.deinit(); + if (offset + 8 > snapshot.active_length) break; + var frame_len_bytes: [8]u8 = undefined; + _ = try file.readPositionalAll(io, &frame_len_bytes, offset); + const frame_len = std.mem.readInt(u64, &frame_len_bytes, .little); + if (frame_len > snapshot.active_length - offset - 8) break; + const compressed = try block_arena.allocator().alloc(u8, @intCast(frame_len)); + _ = try file.readPositionalAll(io, compressed, offset + 8); + offset += 8 + @as(usize, @intCast(frame_len)); + var frame_reader: std.Io.Reader = .fixed(compressed); + var zstd_stream: std.compress.zstd.Decompress = .init(&frame_reader, &.{}, .{}); + var decoded: std.Io.Writer.Allocating = .init(block_arena.allocator()); + _ = zstd_stream.reader.streamRemaining(&decoded.writer) catch break; + const block = try segment.decodeBlock(block_arena.allocator(), try decoded.toOwnedSlice()); + if (block.events.len == 0) continue; + const last = block.events[block.events.len - 1]; + if (last.witnessed_at < time_us) { + next_after = last.seq +| 1; + continue; + } + for (block.events) |event| { + if (event.witnessed_at >= time_us) return .{ .seq = @max(event.seq, 1) }; + } + } + return .{ .seq = @max(next_after, 1) }; +} + /// stream all events with witnessed_at in [from_us, until_us) matching /// `filter` to `sink` (fn (ctx, json: []const u8) !void). returns the /// number of frames written. @@ -1331,6 +1388,64 @@ test "timestamp cursor resolves exact seq from real sealed blocks" { try testing.expect(!newer.clamped); } +test "timestamp cursor resolves into the active region, not the sealed boundary" { + // the 2026-08-08 treadmill: a near-now cursor resolved to sealed_max+1 — + // up to a whole open segment back — and the cold walk could not outrun + // live ingest. times inside the fsynced active region must resolve there. + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buf); + var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); + defer archive.deinit(); + archive.max_events_per_block = 2; + archive.writer.max_events_per_block = 2; + + const base: i64 = 1_700_000_000_000_000; + for (0..4) |i| { + _ = try archive.append(.{ + .seq = 0, + .witnessed_at = base + 1_000 + @as(i64, @intCast(i)) * 1_000, + .indexed_at = 0, + .kind = .delete, + .did = "did:plc:activefixture", + .collection = "app.bsky.feed.post", + .rkey = "3k2abcdefghij", + .rev = "3k2abcdefghij", + .payload = "", + }, base + 1_000); + } + try archive.rotate(); + // two rows into the fresh active segment (one full flushed block) + for (0..2) |i| { + _ = try archive.append(.{ + .seq = 0, + .witnessed_at = base + 5_000 + @as(i64, @intCast(i)) * 1_000, + .indexed_at = 0, + .kind = .delete, + .did = "did:plc:activefixture", + .collection = "app.bsky.feed.post", + .rkey = "3k2abcdefghij", + .rev = "3k2abcdefghij", + .payload = "", + }, base + 5_000); + } + + // between sealed max and the first active row: first active row + const boundary = try resolveTimeToSeq(testing.allocator, io, &archive, base + 4_500); + try testing.expectEqual(@as(u64, 5), boundary.seq); + // inside the active region: the exact row, not sealed_max+1 + const inside = try resolveTimeToSeq(testing.allocator, io, &archive, base + 5_800); + try testing.expectEqual(@as(u64, 6), inside.seq); + // newer than every fsynced row: the next not-yet-durable seq + const ahead = try resolveTimeToSeq(testing.allocator, io, &archive, base + 9_000); + try testing.expectEqual(@as(u64, 7), ahead.seq); +} + test "timestamp cursor surfaces corrupt selected block" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{});