From f1441f46fb5e8ad82edce4ad8b23e999c302adb4 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 11 Aug 2026 08:54:21 -0500 Subject: [PATCH] seal: build the footer incrementally so rotation stops stalling ingest ~50s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit sealing decompressed and re-walked the whole ~1.3GB segment under the archive mutex to build the block index and blooms; every append (and so every live subscriber) went quiet for the duration — measured 44-52s per seal, matching a 52s delivery hole observed live. blocks now feed the footer builder at flush time, commitPrepared patches offset/frame size, and recovery pays its one walk at startup via rebuildIndex. the full walk remains only as a fallback when the index is out of lockstep. Co-Authored-By: Claude Opus 5 (1M context) --- src/internal/storage/archive.zig | 1 + src/internal/storage/segment_writer.zig | 156 ++++++++++++++++++++++++ 2 files changed, 157 insertions(+) diff --git a/src/internal/storage/archive.zig b/src/internal/storage/archive.zig index 6bffd9c..4856ea7 100644 --- a/src/internal/storage/archive.zig +++ b/src/internal/storage/archive.zig @@ -383,6 +383,7 @@ pub const Archive = struct { try writer.buf.appendSlice(self.allocator, bytes[0..valid_end]); writer.event_count = event_count; writer.block_count = block_count; + try writer.rebuildIndex(); var file = try self.dir.openFile(self.io, name, .{ .mode = .read_write }); errdefer file.close(self.io); diff --git a/src/internal/storage/segment_writer.zig b/src/internal/storage/segment_writer.zig index 2583ae2..df98af0 100644 --- a/src/internal/storage/segment_writer.zig +++ b/src/internal/storage/segment_writer.zig @@ -121,11 +121,28 @@ pub const ActiveWriter = struct { max_events_per_block: usize = default_max_events_per_block, event_count: u64 = 0, block_count: u32 = 0, + /// Footer state accumulated as blocks flush, so seal() is O(footer) + /// instead of decompressing and re-walking the whole segment while the + /// archive mutex starves ingest (and with it every live subscriber). + /// Heap-allocated so the arena's inner pointer survives the writer + /// being moved by value (recover() constructs then assigns). + index_arena: *std.heap.ArenaAllocator, + builder: footer_mod.Builder, + env: footer_mod.Envelope = .{}, + /// blocks represented in `builder`; seal() takes the incremental path + /// only when this matches block_count, else it falls back to the walk + indexed_blocks: u32 = 0, pub fn init(allocator: Allocator) Error!ActiveWriter { + const index_arena = try allocator.create(std.heap.ArenaAllocator); + errdefer allocator.destroy(index_arena); + index_arena.* = std.heap.ArenaAllocator.init(allocator); + errdefer index_arena.deinit(); var w: ActiveWriter = .{ .allocator = allocator, .pending_arena = std.heap.ArenaAllocator.init(allocator), + .index_arena = index_arena, + .builder = footer_mod.Builder.init(index_arena.allocator()), }; try w.buf.appendSlice(allocator, segment.magic); try w.buf.appendNTimes(allocator, 0, segment.header_size - segment.magic.len); @@ -136,6 +153,8 @@ pub const ActiveWriter = struct { self.buf.deinit(self.allocator); self.pending.deinit(self.allocator); self.pending_arena.deinit(); + self.index_arena.deinit(); + self.allocator.destroy(self.index_arena); } /// stage one event; flushes a block when the threshold is reached. @@ -175,13 +194,59 @@ pub const ActiveWriter = struct { if (self.pending.items.len == 0) return null; const max_seq = self.pending.items[self.pending.items.len - 1].seq; const block = try encodeBlock(self.allocator, self.pending.items); + try self.noteBlock(.{ + .offset = 0, // patched with the file position in commitPrepared + .compressed_size = 0, // ditto, once the frame size is known + .uncompressed_size = @intCast(block.len), + .event_count = 0, + .min_seq = 0, + .max_seq = 0, + .min_witnessed_at = 0, + .max_witnessed_at = 0, + }, self.pending.items); self.pending.clearRetainingCapacity(); _ = self.pending_arena.reset(.retain_capacity); return .{ .bytes = block, .max_seq = max_seq }; } + /// Record one block into the incremental footer state. `info` needs only + /// sizes/offset populated; envelope fields are derived from `events`. + fn noteBlock(self: *ActiveWriter, info_in: segment.BlockIndexEntry, events: []const segment.Event) Error!void { + var info = info_in; + info.event_count = @intCast(events.len); + info.min_seq = events[0].seq; + info.max_seq = events[0].seq; + info.min_witnessed_at = events[0].witnessed_at; + info.max_witnessed_at = events[0].witnessed_at; + for (events) |ev| { + info.min_seq = @min(info.min_seq, ev.seq); + info.max_seq = @max(info.max_seq, ev.seq); + info.min_witnessed_at = @min(info.min_witnessed_at, ev.witnessed_at); + info.max_witnessed_at = @max(info.max_witnessed_at, ev.witnessed_at); + } + if (self.env.event_count == 0) { + self.env.min_seq = info.min_seq; + self.env.max_seq = info.max_seq; + self.env.min_witnessed_at = info.min_witnessed_at; + self.env.max_witnessed_at = info.max_witnessed_at; + } else { + self.env.min_seq = @min(self.env.min_seq, info.min_seq); + self.env.max_seq = @max(self.env.max_seq, info.max_seq); + self.env.min_witnessed_at = @min(self.env.min_witnessed_at, info.min_witnessed_at); + self.env.max_witnessed_at = @max(self.env.max_witnessed_at, info.max_witnessed_at); + } + self.env.event_count += events.len; + try self.builder.addBlock(info, events); + self.indexed_blocks += 1; + } + /// Commit a previously compressed block in preparation order. pub fn commitPrepared(self: *ActiveWriter, frame: []const u8) Error!void { + if (self.block_count < self.builder.infos.items.len) { + const info = &self.builder.infos.items[self.block_count]; + info.offset = self.buf.items.len; + info.compressed_size = @intCast(frame.len); + } var len_prefix: [8]u8 = undefined; std.mem.writeInt(u64, &len_prefix, frame.len, .little); try self.buf.appendSlice(self.allocator, &len_prefix); @@ -209,6 +274,16 @@ pub const ActiveWriter = struct { pub fn seal(self: *ActiveWriter) !void { try self.flush(); + // Incremental path: every block was indexed as it flushed (or by + // rebuildIndex after recovery), so the footer needs no re-walk. The + // walk below survives only as the fallback for a writer whose index + // fell out of lockstep — it decompresses the entire segment and is + // what used to stall ingest ~50s per seal under the archive mutex. + if (self.indexed_blocks == self.block_count) { + try self.builder.finish(self.allocator, &self.buf, self.env); + return; + } + // Footer metadata must live until finish(), but decoded block payloads // must not. A single arena for both retained every uncompressed block // in the segment and made recovery of a 268 MiB compressed segment @@ -265,6 +340,30 @@ pub const ActiveWriter = struct { try builder.finish(self.allocator, &self.buf, env); } + + /// Re-derive the incremental footer index from a resumed file image + /// (recovery), paying the one full decompression walk at startup so the + /// eventual seal stays off the ingest hot path. + pub fn rebuildIndex(self: *ActiveWriter) !void { + var it: ActiveIterator = .{ .bytes = self.buf.items }; + while (true) { + var block_arena = std.heap.ArenaAllocator.init(self.allocator); + defer block_arena.deinit(); + const before = it.offset; + const block = (try it.next(block_arena.allocator())) orelse break; + if (block.events.len == 0) break; + try self.noteBlock(.{ + .offset = before, + .compressed_size = @intCast(it.offset - before - 8), + .uncompressed_size = @intCast(block.buffer.len), + .event_count = 0, + .min_seq = 0, + .max_seq = 0, + .min_witnessed_at = 0, + .max_witnessed_at = 0, + }, block.events); + } + } }; pub fn compressPrepared(allocator: Allocator, prepared: *const PreparedBlock) Error![]u8 { @@ -340,6 +439,63 @@ test "validateEvent enforces column widths" { try testing.expectError(error.InvalidEvent, validateEvent(ev)); } +test "seal: incremental footer is byte-identical to the walk fallback" { + // varied dids/collections/kinds so the blooms, interning table, and + // $-sentinels all carry real content into the footer + var dids: [13][32]u8 = undefined; + var cols: [5][32]u8 = undefined; + for (&dids, 0..) |*d, i| _ = std.fmt.bufPrint(d, "did:plc:writer{d:0>17}", .{i}) catch unreachable; + for (&cols, 0..) |*c, i| _ = std.fmt.bufPrint(c, "app.test.collection{d:0>3}", .{i}) catch unreachable; + + var a = try ActiveWriter.init(testing.allocator); + defer a.deinit(); + a.max_events_per_block = 100; + var seq: u64 = 1; + while (seq <= 250) : (seq += 1) { + var ev = sampleEvent(seq); + ev.did = dids[seq % 13][0..24]; + ev.collection = cols[seq % 5][0..22]; + if (seq % 11 == 0) { + ev.kind = .account; + ev.collection = ""; + ev.rkey = ""; + } + try a.append(ev); + } + try a.flush(); + try testing.expectEqual(a.block_count, a.indexed_blocks); + + // resumed cold with no index: seal must take the walk fallback + var b = try ActiveWriter.init(testing.allocator); + defer b.deinit(); + b.buf.clearRetainingCapacity(); + try b.buf.appendSlice(testing.allocator, a.bytes()); + b.event_count = a.event_count; + b.block_count = a.block_count; + + // resumed the way recovery does: rebuildIndex restores the fast path + var c = try ActiveWriter.init(testing.allocator); + defer c.deinit(); + c.buf.clearRetainingCapacity(); + try c.buf.appendSlice(testing.allocator, a.bytes()); + c.event_count = a.event_count; + c.block_count = a.block_count; + try c.rebuildIndex(); + try testing.expectEqual(c.block_count, c.indexed_blocks); + + try testing.expectEqual(@as(u32, 0), b.indexed_blocks); // proves b walks + try a.seal(); + try b.seal(); + try c.seal(); + try testing.expectEqualSlices(u8, b.bytes(), a.bytes()); + try testing.expectEqualSlices(u8, b.bytes(), c.bytes()); + + // and the sealed image parses + var sealed = try segment.Sealed.parse(testing.allocator, a.bytes()); + defer sealed.deinit(testing.allocator); + try testing.expectEqual(@as(u64, 250), sealed.header.event_count); +} + test "active file round-trips through our reader's block walk" { var w = try ActiveWriter.init(testing.allocator); defer w.deinit(); -- 2.51.2