//! sealed-segment rewrite — the physical side of compaction
//!
//! upstream segment/rewrite.go: read a sealed jss file, apply a keep
//! predicate per event, and produce a new sealed file. block topology is
//! PRESERVED — a fully-dropped block remains as an event_count=0 block,
//! and every block keeps its HISTORICAL seq/witnessed-at envelope (a
//! superset of the surviving rows), so block numbering, cursor
//! translation, and seq-bound pruning stay stable across compaction
//! generations. rewritten files get a new checksum.
//!
//! crash safety: write sibling `<name>.tmp` → fsync → rename over the
//! original → fsync the parent dir. failure at or before the rename
//! leaves the original untouched; stale tmps are reclaimed at boot and
//! pass start (cleanStaleTmp). when no row drops the file is untouched
//! ("clean" — rewritten=false).

const std = @import("std");
const crashpoint = @import("../runtime/crashpoint.zig");
const footer_mod = @import("segment_footer.zig");
const gloom = @import("gloom.zig");
const segment = @import("segment.zig");
const writer_mod = @import("segment_writer.zig");
const segment_io = @import("segment_io.zig");
const zstd = @import("zstd.zig");

const Io = std.Io;
const Allocator = std.mem.Allocator;
const log = std.log.scoped(.stream);

pub const Result = struct {
    rewritten: bool = false,
    kept: u64 = 0,
    dropped: u64 = 0,
    bytes_rewritten: u64 = 0,
    /// Blocks that actually lost a row and so had to be re-encoded and
    /// recompressed. Blocks that dropped nothing reuse their source frame, so
    /// this stays far below block_count on a typical compaction rewrite.
    blocks_recompressed: u64 = 0,
};

pub const Options = struct {
    /// null means full scan; a non-null slice enables the segment-level
    /// candidate-DID bloom prefilter used by upstream compaction.
    candidate_dids: ?[]const []const u8 = null,
};

/// rewrite sealed bytes in memory. returns the new sealed file bytes
/// (caller frees) when at least one row drops, null when the segment is
/// clean. `keep(ctx, ev)` returns true to retain the row.
pub fn rewrite(
    allocator: Allocator,
    old_bytes: []const u8,
    result: *Result,
    opts: Options,
    ctx: anytype,
    comptime keep: fn (@TypeOf(ctx), segment.Event) bool,
) !?[]u8 {
    var sealed = try segment.Sealed.parse(allocator, old_bytes);
    defer sealed.deinit(allocator);

    if (opts.candidate_dids) |candidates| {
        if (sealed.header.block_did_bloom_offset < sealed.header.did_bloom_offset)
            return error.BadFooter;
        var filter = try gloom.unmarshal(allocator, old_bytes[sealed.header.did_bloom_offset..sealed.header.block_did_bloom_offset]);
        defer filter.deinit(allocator);
        var possible = false;
        for (candidates) |did| {
            if (filter.contains(did)) {
                possible = true;
                break;
            }
        }
        if (!possible) return null;
    }

    var arena = std.heap.ArenaAllocator.init(allocator);
    defer arena.deinit();
    const scratch = arena.allocator();

    var out: std.ArrayList(u8) = .empty;
    defer out.deinit(allocator);
    try out.appendSlice(allocator, segment.magic);
    try out.appendNTimes(allocator, 0, segment.header_size - segment.magic.len);

    // the footer builder sizes per-block blooms from the surviving rows'
    // actual unique-DID cardinality (tighter than the source segment's
    // event-count bound it previously carried forward)
    var builder = footer_mod.Builder.init(scratch);

    var kept_events: std.ArrayList(segment.Event) = .empty;
    defer kept_events.deinit(allocator);

    for (sealed.block_index, 0..) |old_entry, i| {
        var block = try sealed.readBlock(allocator, i);
        defer block.deinit(allocator);

        kept_events.clearRetainingCapacity();
        var dropped_in_block = false;
        for (block.events) |ev| {
            if (keep(ctx, ev)) {
                try kept_events.append(allocator, ev);
            } else {
                result.dropped += 1;
                dropped_in_block = true;
            }
        }
        result.kept += kept_events.items.len;

        // A block that lost nothing re-encodes to the same rows, so reuse its
        // original compressed frame verbatim rather than paying zstd to
        // reproduce bytes we already have (upstream segment/rewrite.go does the
        // same). Compaction rewrites the whole archive to apply the tombstones
        // of a much smaller window, so most blocks in most segments drop
        // nothing; recompressing them unconditionally was the bulk of the work,
        // and on a clean segment every byte of it is discarded below.
        var encoded: ?[]u8 = null;
        defer if (encoded) |e| allocator.free(e);
        var compressed: ?[]u8 = null;
        defer if (compressed) |c| allocator.free(c);

        var frame: []const u8 = undefined;
        var uncompressed_size: u32 = old_entry.uncompressed_size;
        if (dropped_in_block) {
            result.blocks_recompressed += 1;
            const e = try writer_mod.encodeBlock(allocator, kept_events.items);
            encoded = e;
            const c = zstd.compress(allocator, e) catch |err| switch (err) {
                error.OutOfMemory => return error.OutOfMemory,
                else => return error.CompressFailed,
            };
            compressed = c;
            frame = c;
            uncompressed_size = @intCast(e.len);
        } else {
            frame = old_bytes[old_entry.offset + 8 ..][0..old_entry.compressed_size];
        }

        var info = old_entry; // historical envelopes preserved (supersets)
        info.offset = out.items.len;
        info.compressed_size = @intCast(frame.len);
        info.uncompressed_size = uncompressed_size;
        info.event_count = @intCast(kept_events.items.len);
        try builder.addBlock(info, kept_events.items);

        var len_prefix: [8]u8 = undefined;
        std.mem.writeInt(u64, &len_prefix, frame.len, .little);
        try out.appendSlice(allocator, &len_prefix);
        try out.appendSlice(allocator, frame);
    }

    if (result.dropped == 0) return null;
    result.rewritten = true;

    // header envelopes stay historical too; only the event count shrinks
    try builder.finish(allocator, &out, .{
        .min_seq = sealed.header.min_seq,
        .max_seq = sealed.header.max_seq,
        .min_witnessed_at = sealed.header.min_witnessed_at,
        .max_witnessed_at = sealed.header.max_witnessed_at,
        .event_count = result.kept,
    });
    const owned = try out.toOwnedSlice(allocator);
    result.bytes_rewritten = owned.len;
    return owned;
}

/// rewrite one sealed segment file in place (tmp + fsync + rename +
/// dir fsync). clean segments are untouched. errors are the caller's to
/// surface loudly — a failed rewrite must fail the pass, never advance
/// the watermark.
pub fn rewriteFile(
    allocator: Allocator,
    io: Io,
    dir: Io.Dir,
    name: []const u8,
    opts: Options,
    ctx: anytype,
    comptime keep: fn (@TypeOf(ctx), segment.Event) bool,
) !Result {
    const old_bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31));
    defer allocator.free(old_bytes);

    var result: Result = .{};
    const new_bytes = (try rewrite(allocator, old_bytes, &result, opts, ctx, keep)) orelse return result;
    defer allocator.free(new_bytes);

    try swapFile(io, dir, name, new_bytes);
    return result;
}

/// crash-safe replacement of a sealed file: sibling tmp → fsync → rename
/// over the original → fsync the parent dir. failure at or before the
/// rename leaves the original untouched.
fn swapFile(io: Io, dir: Io.Dir, name: []const u8, new_bytes: []const u8) !void {
    var tmp_buf: [128]u8 = undefined;
    const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name});
    var f = try dir.createFile(io, tmp_name, .{ .truncate = true });
    var file_open = true;
    errdefer {
        if (file_open) f.close(io);
        dir.deleteFile(io, tmp_name) catch {};
    }
    try segment_io.writeStreamingAll(&f, io, new_bytes);
    crashpoint.hit(.after_segment_rewrite_temp_written);
    try segment_io.sync(&f, io);
    crashpoint.hit(.after_segment_rewrite_temp_synced);
    f.close(io);
    file_open = false;
    try segment_io.rename(dir, tmp_name, dir, name, io);
    crashpoint.hit(.after_segment_rewrite_renamed);
    // make the rename durable (portable directory fsync; see segment_io.syncDir)
    try segment_io.syncDir(dir, io);
    crashpoint.hit(.after_segment_rewrite_dir_synced);
}

/// Rebuild only the footer with right-sized per-block DID blooms,
/// reusing the original compressed frames verbatim (the data region is
/// byte-identical; blocks are decompressed solely to read their DID
/// sets). Returns null when the segment's bloom region is already at
/// the right-sized width. Used by the legacy-archive rebloom sweep:
/// segments sealed before the cardinality sizing fix carry blooms sized
/// for the 4096-event block cap, ~1000x over the measured per-block
/// unique-DID median, and resident bloom regions dominate server heap.
pub fn rebloom(allocator: Allocator, old_bytes: []const u8, result: *Result) !?[]u8 {
    var sealed = try segment.Sealed.parse(allocator, old_bytes);
    defer sealed.deinit(allocator);

    var arena = std.heap.ArenaAllocator.init(allocator);
    defer arena.deinit();
    const scratch = arena.allocator();

    var builder = footer_mod.Builder.init(scratch);
    for (sealed.block_index, 0..) |old_entry, i| {
        var block = try sealed.readBlock(allocator, i);
        defer block.deinit(allocator);
        result.kept += block.events.len;
        // offsets/sizes unchanged: the data region is copied verbatim
        try builder.addBlock(old_entry, block.events);
    }

    var out: std.ArrayList(u8) = .empty;
    defer out.deinit(allocator);
    // header placeholder + the original data region, byte for byte
    try out.appendSlice(allocator, old_bytes[0..sealed.header.block_index_offset]);
    @memset(out.items[segment.magic.len..segment.header_size], 0);

    try builder.finish(allocator, &out, .{
        .min_seq = sealed.header.min_seq,
        .max_seq = sealed.header.max_seq,
        .min_witnessed_at = sealed.header.min_witnessed_at,
        .max_witnessed_at = sealed.header.max_witnessed_at,
        .event_count = sealed.header.event_count,
    });

    if (out.items.len >= old_bytes.len) return null; // already right-sized
    result.rewritten = true;
    const owned = try out.toOwnedSlice(allocator);
    result.bytes_rewritten = owned.len;
    return owned;
}

/// rebloom one sealed segment file in place with rewriteFile's exact
/// crash-safety dance. Returns rewritten=false when already right-sized.
pub fn rebloomFile(allocator: Allocator, io: Io, dir: Io.Dir, name: []const u8) !Result {
    const old_bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31));
    defer allocator.free(old_bytes);

    var result: Result = .{};
    const new_bytes = (try rebloom(allocator, old_bytes, &result)) orelse return result;
    defer allocator.free(new_bytes);

    try swapFile(io, dir, name, new_bytes);
    return result;
}

/// reclaim tmps left by a crashed rewrite (boot and pass start; the
/// single-compactor invariant means any tmp seen here is stale)
pub fn cleanStaleTmp(allocator: Allocator, io: Io, dir: Io.Dir) !void {
    var stale: std.ArrayList([]const u8) = .empty;
    defer {
        for (stale.items) |n| allocator.free(n);
        stale.deinit(allocator);
    }
    var it = dir.iterate();
    while (try it.next(io)) |entry| {
        if (std.mem.endsWith(u8, entry.name, ".jss.tmp"))
            try stale.append(allocator, try allocator.dupe(u8, entry.name));
    }
    for (stale.items) |n| {
        dir.deleteFile(io, n) catch |err| {
            log.warn("compaction: could not remove stale {s}: {s}", .{ n, @errorName(err) });
            continue;
        };
        log.info("compaction: reclaimed stale {s}", .{n});
    }
}

// === tests ===

const testing = std.testing;

fn sampleEvent(seq: u64, kind: segment.Kind, rkey: []const u8) segment.Event {
    return .{
        .seq = seq,
        .witnessed_at = @intCast(1_700_000_000_000_000 + seq),
        .indexed_at = 0,
        .kind = kind,
        .did = "did:plc:rewritetestaccount",
        .collection = "app.bsky.feed.post",
        .rkey = rkey,
        .rev = "3l3qo2vutsw2b",
        .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69",
    };
}

fn buildSealed(allocator: Allocator) ![]u8 {
    var w = try writer_mod.ActiveWriter.init(allocator);
    defer w.deinit();
    w.max_events_per_block = 10;
    var seq: u64 = 1;
    while (seq <= 30) : (seq += 1) try w.append(sampleEvent(seq, .create, "3l3qo2vuowo2b"));
    try w.seal();
    return allocator.dupe(u8, w.bytes());
}

const DropBelow = struct { cutoff: u64 };
fn keepAbove(ctx: DropBelow, ev: segment.Event) bool {
    return ev.seq >= ctx.cutoff;
}

fn unreachableKeep(_: void, _: segment.Event) bool {
    @panic("segment bloom false negative: predicate must not run");
}

test "rewrite: clean segment returns null, file layout preserved on drops" {
    const old = try buildSealed(testing.allocator);
    defer testing.allocator.free(old);

    // clean: nothing drops
    var clean_res: Result = .{};
    try testing.expectEqual(@as(?[]u8, null), try rewrite(testing.allocator, old, &clean_res, .{}, DropBelow{ .cutoff = 0 }, keepAbove));
    try testing.expect(!clean_res.rewritten);
    try testing.expectEqual(@as(u64, 30), clean_res.kept);

    // drop the whole first block (seqs 1..10) plus half the second
    var res: Result = .{};
    const new_bytes = (try rewrite(testing.allocator, old, &res, .{}, DropBelow{ .cutoff = 16 }, keepAbove)).?;
    defer testing.allocator.free(new_bytes);
    try testing.expect(res.rewritten);
    try testing.expectEqual(@as(u64, 15), res.dropped);
    try testing.expectEqual(@as(u64, 15), res.kept);

    var sealed = try segment.Sealed.parse(testing.allocator, new_bytes);
    defer sealed.deinit(testing.allocator);
    // block topology preserved: 3 blocks, first now empty
    try testing.expectEqual(@as(u32, 3), sealed.header.block_count);
    try testing.expectEqual(@as(u32, 15), sealed.header.event_count);
    // historical envelopes survive as supersets
    try testing.expectEqual(@as(u64, 1), sealed.header.min_seq);
    try testing.expectEqual(@as(u64, 30), sealed.header.max_seq);
    try testing.expectEqual(@as(u32, 0), sealed.block_index[0].event_count);
    try testing.expectEqual(@as(u64, 1), sealed.block_index[0].min_seq);
    try testing.expectEqual(@as(u64, 10), sealed.block_index[0].max_seq);
    try testing.expectEqual(@as(u32, 5), sealed.block_index[1].event_count);
    try testing.expectEqual(@as(u64, 11), sealed.block_index[1].min_seq);

    var total: usize = 0;
    var expect_seq: u64 = 16;
    for (0..sealed.block_index.len) |i| {
        var block = try sealed.readBlock(testing.allocator, i);
        defer block.deinit(testing.allocator);
        for (block.events) |ev| {
            try testing.expectEqual(expect_seq, ev.seq);
            expect_seq += 1;
        }
        total += block.events.len;
    }
    try testing.expectEqual(@as(usize, 15), total);
}

test "rewrite: blocks that drop nothing reuse their compressed frame verbatim" {
    const old = try buildSealed(testing.allocator);
    defer testing.allocator.free(old);
    var source = try segment.Sealed.parse(testing.allocator, old);
    defer source.deinit(testing.allocator);

    // drop one row from block 0 only; blocks 1 and 2 are untouched
    var res: Result = .{};
    const new_bytes = (try rewrite(testing.allocator, old, &res, .{}, DropBelow{ .cutoff = 2 }, keepAbove)).?;
    defer testing.allocator.free(new_bytes);
    try testing.expectEqual(@as(u64, 1), res.dropped);

    // The load-bearing assertion. zstd is deterministic, so recompressing an
    // untouched block reproduces its bytes exactly -- comparing frames alone
    // would pass whether or not the reuse path exists. Counting the
    // recompressions is what actually distinguishes the two.
    try testing.expectEqual(@as(u64, 1), res.blocks_recompressed);

    var sealed = try segment.Sealed.parse(testing.allocator, new_bytes);
    defer sealed.deinit(testing.allocator);
    try testing.expectEqual(@as(u32, 3), sealed.header.block_count);

    // untouched blocks carry the ORIGINAL compressed bytes, not a
    // re-compression that merely happens to decode the same
    for (1..3) |i| {
        const src_entry = source.block_index[i];
        const new_entry = sealed.block_index[i];
        const src_frame = old[src_entry.offset + 8 ..][0..src_entry.compressed_size];
        const new_frame = new_bytes[new_entry.offset + 8 ..][0..new_entry.compressed_size];
        try testing.expectEqualSlices(u8, src_frame, new_frame);
        try testing.expectEqual(src_entry.uncompressed_size, new_entry.uncompressed_size);
    }
    // the touched block genuinely shrank
    try testing.expectEqual(@as(u32, 9), sealed.block_index[0].event_count);

    // and every surviving row still decodes in order
    var expect_seq: u64 = 2;
    for (0..sealed.block_index.len) |i| {
        var block = try sealed.readBlock(testing.allocator, i);
        defer block.deinit(testing.allocator);
        for (block.events) |ev| {
            try testing.expectEqual(expect_seq, ev.seq);
            expect_seq += 1;
        }
    }
    try testing.expectEqual(@as(u64, 31), expect_seq);
}

test "rewrite file exhaustively propagates segment I/O faults without silent corruption" {
    segment_io.disarm();
    defer segment_io.disarm();
    const old = try buildSealed(testing.allocator);
    defer testing.allocator.free(old);

    var reference_tmp = testing.tmpDir(.{});
    defer reference_tmp.cleanup();
    {
        var file = try reference_tmp.dir.createFile(testing.io, "seg.jss", .{});
        defer file.close(testing.io);
        try file.writeStreamingAll(testing.io, old);
    }
    _ = try rewriteFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove);
    const committed = try reference_tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
    defer testing.allocator.free(committed);
    try testing.expect(!std.mem.eql(u8, old, committed));

    const cases = [_]struct { spec: []const u8, committed_after_error: bool }{
        .{ .spec = "write:1:enospc", .committed_after_error = false },
        .{ .spec = "write:1:shortwrite", .committed_after_error = false },
        .{ .spec = "sync:1:eio", .committed_after_error = false },
        .{ .spec = "rename:1:eio", .committed_after_error = false },
        .{ .spec = "sync:2:eio", .committed_after_error = true },
    };
    for (cases) |case| {
        var tmp = testing.tmpDir(.{});
        defer tmp.cleanup();
        {
            var file = try tmp.dir.createFile(testing.io, "seg.jss", .{});
            defer file.close(testing.io);
            try file.writeStreamingAll(testing.io, old);
        }
        try testing.expect(segment_io.armSpec(case.spec));
        try testing.expectError(switch (case.spec[0]) {
            'w' => if (std.mem.endsWith(u8, case.spec, "enospc")) error.NoSpaceLeft else error.ShortWrite,
            else => error.InputOutput,
        }, rewriteFile(testing.allocator, testing.io, tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove));
        segment_io.disarm();
        try testing.expectError(error.FileNotFound, tmp.dir.openFile(testing.io, "seg.jss.tmp", .{}));
        const got = try tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
        defer testing.allocator.free(got);
        try testing.expectEqualSlices(u8, if (case.committed_after_error) committed else old, got);
    }
}

test "rewrite: unrelated candidate DIDs skip the real segment decode" {
    const old = try buildSealed(testing.allocator);
    defer testing.allocator.free(old);
    var result: Result = .{};
    const candidates = [_][]const u8{"did:plc:not-in-this-segment"};
    try testing.expectEqual(@as(?[]u8, null), try rewrite(
        testing.allocator,
        old,
        &result,
        .{ .candidate_dids = &candidates },
        {},
        unreachableKeep,
    ));
    try testing.expect(!result.rewritten);
}

test "rewriteFile: atomic in-place rewrite; stale tmp reclaim" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } });
    defer dir.close(io);

    const old = try buildSealed(testing.allocator);
    defer testing.allocator.free(old);
    {
        var f = try dir.createFile(io, "seg_0000000000.jss", .{});
        defer f.close(io);
        try f.writeStreamingAll(io, old);
    }
    // a crashed rewrite left a stale tmp
    {
        var f = try dir.createFile(io, "seg_0000000000.jss.tmp", .{});
        f.close(io);
    }
    try cleanStaleTmp(testing.allocator, io, dir);
    try testing.expectError(error.FileNotFound, dir.access(io, "seg_0000000000.jss.tmp", .{}));

    const res = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove);
    try testing.expect(res.rewritten);
    try testing.expectEqual(@as(u64, 20), res.dropped);

    const bytes = try dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20));
    defer testing.allocator.free(bytes);
    var sealed = try segment.Sealed.parse(testing.allocator, bytes);
    defer sealed.deinit(testing.allocator);
    try testing.expectEqual(@as(u32, 10), sealed.header.event_count);
    try testing.expectEqual(@as(u32, 3), sealed.header.block_count);

    // second pass over the compacted file is clean
    const res2 = try rewriteFile(testing.allocator, io, dir, "seg_0000000000.jss", .{}, DropBelow{ .cutoff = 21 }, keepAbove);
    try testing.expect(!res2.rewritten);
}

test "rebloom shrinks a pre-fix upstream fixture and is idempotent" {
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    // committed upstream-produced segment sealed before the bloom
    // cardinality-sizing fix (2026-07-08 < upstream f02919c): its per-block
    // blooms carry the fixed 4096-event capacity. skipped when absent.
    const old_bytes = Io.Dir.cwd().readFileAlloc(
        io,
        "tests/fixtures/seg_0000000000.jss",
        testing.allocator,
        .limited(4 << 20),
    ) catch return;
    defer testing.allocator.free(old_bytes);

    var result: Result = .{};
    const new_bytes = (try rebloom(testing.allocator, old_bytes, &result)) orelse
        return error.FixtureAlreadyRightSized;
    defer testing.allocator.free(new_bytes);
    try testing.expect(result.rewritten);
    try testing.expect(new_bytes.len < old_bytes.len);

    // the rewritten file must parse, checksum, and serve identical rows
    var old_sealed = try segment.Sealed.parse(testing.allocator, old_bytes);
    defer old_sealed.deinit(testing.allocator);
    var new_sealed = try segment.Sealed.parse(testing.allocator, new_bytes);
    defer new_sealed.deinit(testing.allocator);
    try testing.expect(segment.checksumValid(new_bytes, new_sealed.header));
    try testing.expectEqual(old_sealed.header.event_count, new_sealed.header.event_count);
    try testing.expectEqual(old_sealed.block_index.len, new_sealed.block_index.len);
    for (0..new_sealed.block_index.len) |i| {
        var old_block = try old_sealed.readBlock(testing.allocator, i);
        defer old_block.deinit(testing.allocator);
        var new_block = try new_sealed.readBlock(testing.allocator, i);
        defer new_block.deinit(testing.allocator);
        try testing.expectEqual(old_block.events.len, new_block.events.len);
        for (old_block.events, new_block.events) |a, b| {
            try testing.expectEqual(a.seq, b.seq);
            try testing.expectEqualStrings(a.did, b.did);
            try testing.expectEqualStrings(a.payload, b.payload);
        }
    }

    // right-sized blooms still answer membership for every resident DID
    const blobs = try new_sealed.blockBloomBlobs(testing.allocator);
    defer testing.allocator.free(blobs);
    for (0..new_sealed.block_index.len) |i| {
        var block = try new_sealed.readBlock(testing.allocator, i);
        defer block.deinit(testing.allocator);
        for (block.events) |ev| {
            if (ev.did.len > 0) try testing.expect(try gloom.containsMarshaled(blobs[i], ev.did));
        }
    }

    // second pass: nothing left to shrink
    var again: Result = .{};
    try testing.expectEqual(@as(?[]u8, null), try rebloom(testing.allocator, new_bytes, &again));
}