diff --git a/build.zig b/build.zig index f04bada197..b1252cb381 100644 --- a/build.zig +++ b/build.zig @@ -317,53 +317,69 @@ const Sig = struct { }); unit_tests.add("lib", lib); - // Components: one per subdir of v2/components/. Each gets lib, tracy, and build-options + // Components: one per subdir of v2/components/. Each gets lib, tracy, and build-options. + // + // A component's impl (component.zig + siblings) may opt into a sibling + // component's api via `extra_api_deps`. The `_api` module itself always + // stays isolated to lib/tracy/build-options so the public surface is + // guaranteed cycle-free. + const extra_api_deps = std.StaticStringMap([]const []const u8).initComptime(&.{ + .{ "replay", &.{"shred_api"} }, + }); + + const component_base_imports: []const Build.Module.Import = &.{ + .{ .name = "lib", .module = lib }, + .{ .name = "tracy", .module = deps.tracy }, + .{ .name = "build-options", .module = build_options_mod }, + }; + + const Pending = struct { name: []const u8, comp_dir: []const u8, api: *Build.Module }; + var pending: std.ArrayList(Pending) = .empty; var components: std.StringHashMapUnmanaged(Component) = .empty; var api_import_list: std.ArrayList(Build.Module.Import) = .empty; + + // Pass 1: build every api so pass 2's sibling lookups can't miss. { - var dir = try b.build_root.handle.openDir( - "v2/components", - .{ .iterate = true }, - ); + var dir = try b.build_root.handle.openDir("v2/components", .{ .iterate = true }); defer dir.close(); var it = dir.iterate(); while (try it.next()) |entry| { if (entry.kind != .directory) continue; if (std.mem.eql(u8, entry.name, "runtime")) continue; const name = b.dupe(entry.name); - const comp_dir = b.fmt("v2/components/{s}", .{name}); - const imports: []const Build.Module.Import = &.{ - .{ .name = "lib", .module = lib }, - .{ .name = "tracy", .module = deps.tracy }, - .{ .name = "build-options", .module = build_options_mod }, - }; - const api_name = b.fmt("{s}_api", .{name}); const api = b.addModule(api_name, .{ .root_source_file = b.path(b.fmt("{s}/api.zig", .{comp_dir})), .target = config.target, .optimize = config.optimize, - .imports = imports, + .imports = component_base_imports, }); unit_tests.add(api_name, api); - - const component = b.addModule(name, .{ - .root_source_file = b.path(b.fmt("{s}/component.zig", .{comp_dir})), - .target = config.target, - .optimize = config.optimize, - .imports = concatImports(b, &.{ - imports, - &.{.{ .name = "api", .module = api }}, - }), - }); - unit_tests.add(name, component); - - try components.put(b.allocator, name, .{ .api = api, .component = component }); try api_import_list.append(b.allocator, .{ .name = api_name, .module = api }); + try pending.append(b.allocator, .{ .name = name, .comp_dir = comp_dir, .api = api }); } } + // Pass 2: build each component impl module. + for (pending.items) |p| { + var imports: std.ArrayList(Build.Module.Import) = .empty; + try imports.appendSlice(b.allocator, component_base_imports); + try imports.append(b.allocator, .{ .name = "api", .module = p.api }); + for (extra_api_deps.get(p.name) orelse &.{}) |extra| { + try imports.append(b.allocator, findApi(api_import_list.items, extra)); + } + + const component = b.addModule(p.name, .{ + .root_source_file = b.path(b.fmt("{s}/component.zig", .{p.comp_dir})), + .target = config.target, + .optimize = config.optimize, + .imports = imports.items, + }); + unit_tests.add(p.name, component); + try components.put(b.allocator, p.name, .{ .api = p.api, .component = component }); + } + // runtime is special cased because it needs codegen and a bunch of extra deps that // no other component uses. const runtime = addRuntime(b, config, deps, unit_tests, lib, features_zon, feature_set_id); @@ -594,6 +610,13 @@ fn concatImports( return out; } +fn findApi(imports: []const Build.Module.Import, name: []const u8) Build.Module.Import { + for (imports) |imp| { + if (std.mem.eql(u8, imp.name, name)) return imp; + } + std.debug.panic("extra_api_deps references unknown api '{s}'", .{name}); +} + /// Everything other than Sig itself: developer tools, ci scripts, docs, /// integration tests, etc. const Tools = struct { diff --git a/conformance/build.zig b/conformance/build.zig index aea5f781cf..d9b84c9853 100644 --- a/conformance/build.zig +++ b/conformance/build.zig @@ -93,6 +93,8 @@ pub fn build(b: *Build) void { const sig_v2_mod = sig_v2_dep.module("lib"); const shred_api_mod = sig_v2_dep.module("shred_api"); const shred_mod = sig_v2_dep.module("shred"); + const replay_api_mod = sig_v2_dep.module("replay_api"); + const replay_mod = sig_v2_dep.module("replay"); const pb_dep = b.dependency("pb", .{ .target = target, @@ -105,6 +107,8 @@ pub fn build(b: *Build) void { .{ .name = "sig_v2", .module = sig_v2_mod }, .{ .name = "shred_api", .module = shred_api_mod }, .{ .name = "shred", .module = shred_mod }, + .{ .name = "replay_api", .module = replay_api_mod }, + .{ .name = "replay", .module = replay_mod }, .{ .name = "protobuf", .module = pb_mod }, .{ .name = "build-options", .module = build_options.createModule() }, }; diff --git a/conformance/src/shred_parse.zig b/conformance/src/shred_parse.zig index 63a90ef133..fb7adf2d66 100644 --- a/conformance/src/shred_parse.zig +++ b/conformance/src/shred_parse.zig @@ -27,6 +27,8 @@ const Allocator = std.mem.Allocator; const shred_api = @import("shred_api"); const shred_impl = @import("shred"); +const replay_api = @import("replay_api"); +const replay_impl = @import("replay"); const Shred = shred_api.Shred; const DeshredRing = shred_api.DeshredRing; @@ -39,6 +41,9 @@ const Hash = sig_v2.solana.Hash; const Slot = sig_v2.solana.Slot; const LeaderSchedule = sig_v2.solana.LeaderSchedule; +const MerkleForest = replay_impl.MerkleForest; +const BlockPool = replay_api.BlockPool; + // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- @@ -51,18 +56,14 @@ const HASHES_PER_TICK: u64 = 62_500; // mainnet constant const IN_PROGRESS_CAPACITY: u32 = 64; const DONE_CAPACITY: u32 = 256; -/// Upper bound on shred.slot accepted by the parse pipeline: -/// `root + max(500, 2 * slots_in_epoch(epoch(root)))` against the default -/// (warmup=true) epoch schedule. Shreds past this bound are "too far in the -/// future" and must be discarded before they reach FEC assembly; the agave -/// reference harness derives the same bound via `ShredFilterContext` against -/// a bank built with `EpochSchedule::default()` -/// (agave/ledger/src/shred/filter.rs). +/// Upper bound on shred.slot: `root + max(500, slots_in_epoch(epoch(root)) / 2)` +/// against the default (warmup=true) epoch schedule. +/// [agave] https://github.com/anza-xyz/agave/blob/v4.1.0-rc.1/ledger/src/shred/filter.rs#L405-L414 const MAX_SHRED_DISTANCE_MINIMUM: Slot = 500; fn maxShredSlot(root: Slot) Slot { const schedule = sig_v2.solana.EpochSchedule.INIT; const slots_in_epoch = schedule.getSlotsInEpoch(schedule.getEpoch(root)); - const distance = @max(MAX_SHRED_DISTANCE_MINIMUM, 2 *| slots_in_epoch); + const distance = @max(MAX_SHRED_DISTANCE_MINIMUM, slots_in_epoch / 2); return root +| distance; } @@ -137,10 +138,9 @@ const DataShredCapture = struct { const FECSetParseResult = struct { merkle_root: Hash, chained_merkle_root: Hash, - /// Full concatenation of every data shred's data region across all 32 - /// slots (mirrors agave's fixture, not sig's ring which truncates at - /// the first data_complete). Populated from `DataShredCapture` entries - /// matching `(slot, fec_set_idx)` at proto-encode time. + /// Concatenation of the data regions of all 32 data shreds in this FEC + /// set. Filled at proto-encode time by matching `DataShredCapture` + /// entries on `(slot, fec_set_idx)`. payload: []const u8, slot: Slot, fec_set_index: u32, @@ -174,6 +174,15 @@ const HarnessState = struct { deshred_reader: DeshredRing.Iterator(.reader), leader_schedule: *LeaderSchedule, + /// Replay-side view of the completed FEC sets, driven by + /// `drainCompletions`. Walked in `deriveBlockParseResult` for + /// cross-set invariants. Reset between fixtures. + forest: MerkleForest, + /// Backing storage for `block_pool`; inline so the pool pointer stays + /// valid for the lifetime of `HarnessState`. + block_pool_buf: [BlockPool.size()]u8 align(@alignOf(BlockPool)) = undefined, + block_pool: *BlockPool, + // Per-input accumulators. Reset in executeShredParse; storage capacity // is retained. shred_parse_results: std.ArrayListUnmanaged(bool) = .empty, @@ -182,6 +191,11 @@ const HarnessState = struct { /// Dedup set for `data_shreds`; a shred can appear in `in_progress` /// across multiple processPacket calls, so we key on (slot, slot_idx). seen_data_shreds: std.AutoHashMapUnmanaged(ShredKey, void) = .empty, + /// Slots where `processPacket` returned a conflict error (equivocation + /// with a mismatched root, variant mismatch, malformed recovery). The + /// rejected shred never reaches `MerkleForest`, so this is the only + /// signal `buildProtoEffects` has for those slots. + dead_slots_from_errors: std.AutoHashMapUnmanaged(Slot, void) = .empty, fn init(allocator: Allocator) !*HarnessState { const self = try allocator.create(HarnessState); @@ -197,7 +211,11 @@ const HarnessState = struct { ls.base_slot = 0; @memset(&ls.leaders, .ZEROES); - const receiver = try Receiver.init(allocator, IN_PROGRESS_CAPACITY, DONE_CAPACITY); + var receiver = try Receiver.init(allocator, IN_PROGRESS_CAPACITY, DONE_CAPACITY); + errdefer receiver.deinit(allocator); + + var forest = try MerkleForest.init(allocator); + errdefer forest.deinit(allocator); self.* = .{ .allocator = allocator, @@ -206,7 +224,10 @@ const HarnessState = struct { .deshred_writer = ring.get(.writer), .deshred_reader = ring.get(.reader), .leader_schedule = ls, + .forest = forest, + .block_pool = @ptrCast(&self.block_pool_buf), }; + self.block_pool.init(); return self; } @@ -215,6 +236,8 @@ const HarnessState = struct { self.receiver.reset(); self.deshred_writer = self.deshred_ring.get(.writer); self.deshred_reader = self.deshred_ring.get(.reader); + self.forest.reset(); + self.block_pool.init(); for (self.fec_set_results.items) |it| self.allocator.free(it.payload); self.fec_set_results.clearRetainingCapacity(); @@ -222,6 +245,7 @@ const HarnessState = struct { self.data_shreds.clearRetainingCapacity(); self.shred_parse_results.clearRetainingCapacity(); self.seen_data_shreds.clearRetainingCapacity(); + self.dead_slots_from_errors.clearRetainingCapacity(); } /// Snapshot a data shred into `data_shreds` if we haven't already. @@ -248,10 +272,12 @@ const HarnessState = struct { } /// Drain any completions the Receiver just wrote to the deshred ring - /// and turn each into a scratch `FECSetParseResult`. The ring provides - /// merkle roots, FEC-set id and the batch/slot flags; per-shred data - /// (payload concat, parent_offset) is filled in later from - /// `data_shreds` because RS-recovered shreds aren't in our capture map. + /// and turn each into a scratch `FECSetParseResult`. Per-shred payload + /// is filled in later from `data_shreds` (RS-recovered shreds aren't + /// in our capture map). Each completion is also inserted into the + /// `MerkleForest` for the cross-set walks in `deriveBlockParseResult`; + /// pool exhaustion is fatal here (both pools are sized for the largest + /// fixture). fn drainCompletions(self: *HarnessState) void { while (self.deshred_reader.next()) |completed| { self.fec_set_results.append(self.allocator, .{ @@ -260,12 +286,22 @@ const HarnessState = struct { .payload = &.{}, // filled at proto-encode .slot = completed.id.slot, .fec_set_index = completed.id.fec_set_idx, - .parent_offset = 0, // filled at proto-encode + .parent_offset = completed.parent_offset, .num_data_shreds = FEC_DATA_SHREDS, .num_coding_shreds = FEC_CODING_SHREDS, .data_complete = completed.data_complete, .slot_complete = completed.slot_complete, }) catch @panic("OutOfMemory"); + + _ = replay_impl.insertFecSet( + sig_v2.telemetry.Logger("main").noop, + completed, + &self.forest, + self.block_pool, + ) catch |err| std.debug.panic( + "insertFecSet failed: {s}", + .{@errorName(err)}, + ); } self.deshred_reader.markUsed(); } @@ -293,17 +329,12 @@ fn executeShredParse( st.receiver.updateSlotRange(ctx.root_slot, maxShredSlot(ctx.root_slot)); st.leader_schedule.base_slot = ctx.root_slot; - // Map proto bool flags into the Receiver's per-feature activation slot. - // Agave's `check_feature_activation` uses an epoch-delayed semantic: - // a feature activated at slot `s` only takes effect for shreds in - // epoch > epoch(s). Agave's harness uses `EpochSchedule::default()`, - // which sets `warmup = true`: epoch 0 is only `MINIMUM_SLOTS_PER_EPOCH` - // (= 32) slots long, and later epochs double in size up to - // `DEFAULT_SLOTS_PER_EPOCH`. A feature activated at slot 0 therefore - // first applies at the start of epoch 1, i.e. slot 32. The Receiver's - // check is a plain `shred.slot >= activation_slot`, so map proto - // `true` to slot 32 so sig's gate mirrors agave's epoch-aware check. - // `false` -> disabled (maxInt). + // Feature-activation slots. Agave activates a feature from the epoch + // after the one containing the activation slot; on the default + // (warmup=true) schedule epoch 0 is 32 slots, so activation-at-0 + // first applies at slot 32. The Receiver's check is a plain + // `shred.slot >= activation_slot`, so proto `true` maps to 32 and + // `false` disables via `maxInt`. const MINIMUM_SLOTS_PER_EPOCH: Slot = 32; const features = ctx.features orelse pb.ShredFeatures{}; st.receiver.features = .{ @@ -345,16 +376,29 @@ fn executeShredParse( &st.deshred_writer, .noop, ) catch |err| switch (err) { + // Conflict / equivocation errors: the rejected shred leaves + // no trace in `MerkleForest`, so record its slot here for + // `buildProtoEffects` to fold into the dead-slots set. + error.MerkleRootConflict, + error.MismatchedMerkleRoot, + error.MismatchedChainedMerkleRoot, + error.VariantMismatchFromFecSet, + error.RecoveredShredMalformed, + => { + st.dead_slots_from_errors.put( + allocator, + parsed_shred.?.slot, + {}, + ) catch @panic("OutOfMemory"); + continue; + }, else => continue, }; - // Receiver accepted the shred. If it's a data shred, snapshot it - // now — for `.fec_set_finished` the ctx has already been removed - // from `in_progress`, so a post-call `captureInProgress` would - // miss it. Dedup by (slot, slot_idx) so the same shred re-fed as - // `.shred_already_seen` doesn't double-count. RS-recovered data - // shreds still can't be reached without a Receiver hook and are - // the known fidelity gap under option B. + // Snapshot data shreds the Receiver accepted. Do it now — for + // `.fec_set_finished` the ctx has already been retired, so a + // post-call walk would miss it. `captureShredIfNew` dedups by + // (slot, slot_idx). RS-recovered shreds are a known fidelity gap. switch (result) { .unfinished_fec_set, .fec_set_finished => { const shred = parsed_shred.?; @@ -402,10 +446,80 @@ fn buildProtoEffects( ); out.shred_results.appendSliceAssumeCapacity(st.shred_parse_results.items); - // Fill in per-FEC-set payload + parent_offset from the data-shred - // captures. Every captured shred sits in exactly one FEC set; iterate - // captures once per FEC set (bounded to 32 shreds per set, ~64 - // in-progress sets, so linear scan is fine). + // The runtime keeps no per-slot dead flag, so derive one from + // `MerkleForest` + `receiver.in_progress` for the block verdict. + var dead_slots: DeadSlots = .empty; + defer dead_slots.deinit(allocator); + try deriveBlockParseResult(&st.receiver, &st.forest, allocator, &dead_slots); + // Errors captured in the per-shred loop are the only signal for + // conflicting variants that never landed in `in_progress` or + // `MerkleForest`. + var err_it = st.dead_slots_from_errors.keyIterator(); + while (err_it.next()) |slot| try dead_slots.put(allocator, slot.*, {}); + if (dead_slots.count() > 0) out.block_parse_result = .REJECTED_INVALID_HEADER; + + // Reimplement agave's `check_insert_data_shred` per-slot admission + // rules, in order: + // 1. Parent-offset pin: first-seen `parent_offset` wins; mismatch + // marks the slot dead and skips the `received` bump. + // 2. `last_in_slot` with `slot_idx < received`: reject, mark dead. + // 3. `slot_idx >= last_index` (once `last_in_slot` pins it): reject. + // 4. On admission: `received = max(received, slot_idx + 1)`. + // Sig's runtime carries none of this per-slot state, so we run it + // here over captured admissions to identify FEC sets agave would + // have failed to complete. + var first_seen_parent: std.AutoHashMapUnmanaged(Slot, u16) = .empty; + defer first_seen_parent.deinit(allocator); + var suppressed_fec_sets: std.AutoHashMapUnmanaged(shred_api.FecSetId, void) = .empty; + defer suppressed_fec_sets.deinit(allocator); + var received: std.AutoHashMapUnmanaged(Slot, u32) = .empty; + defer received.deinit(allocator); + var slot_last_index: std.AutoHashMapUnmanaged(Slot, u32) = .empty; + defer slot_last_index.deinit(allocator); + for (st.data_shreds.items) |ds| { + // (1) Parent-offset pin. Shreds disagreeing with the pin never + // reach the `received` bump in agave — skip them here too so + // their slot_idx doesn't spuriously advance `received`. + const parent_gop = try first_seen_parent.getOrPut(allocator, ds.slot); + if (!parent_gop.found_existing) { + parent_gop.value_ptr.* = ds.parent_offset; + } else if (parent_gop.value_ptr.* != ds.parent_offset) { + try dead_slots.put(allocator, ds.slot, {}); + continue; + } + + // (2) + (3) Order-dependent per-slot rejects. + const rejected = blk: { + if (slot_last_index.get(ds.slot)) |li| if (ds.slot_idx >= li) break :blk true; + if (ds.last_shred_in_slot) { + const r = received.get(ds.slot) orelse 0; + if (ds.slot_idx < r) break :blk true; + } + break :blk false; + }; + if (rejected) { + try suppressed_fec_sets.put(allocator, .{ + .slot = ds.slot, + .fec_set_idx = ds.fec_set_idx, + }, {}); + try dead_slots.put(allocator, ds.slot, {}); + continue; + } + + // (4) Admitted -> update per-slot cursors. + if (ds.last_shred_in_slot) { + try slot_last_index.put(allocator, ds.slot, ds.slot_idx); + } + const gop = try received.getOrPut(allocator, ds.slot); + if (!gop.found_existing) gop.value_ptr.* = 0; + gop.value_ptr.* = @max(gop.value_ptr.*, ds.slot_idx + 1); + } + if (dead_slots.count() > 0) out.block_parse_result = .REJECTED_INVALID_HEADER; + + // Fill in per-FEC-set payload from the data-shred captures. Every + // captured shred sits in exactly one FEC set; iterate captures once + // per FEC set (bounded to 32 shreds per set, ~64 in-progress sets, + // so linear scan is fine). std.sort.heap(DataShredCapture, st.data_shreds.items, {}, dataShredOrder); for (st.fec_set_results.items) |*fec_res| { var payload: std.ArrayListUnmanaged(u8) = .empty; @@ -413,18 +527,16 @@ fn buildProtoEffects( for (st.data_shreds.items) |*ds| { if (ds.slot != fec_res.slot or ds.fec_set_idx != fec_res.fec_set_index) continue; payload.appendSlice(st.allocator, ds.payload) catch @panic("OutOfMemory"); - fec_res.parent_offset = ds.parent_offset; } fec_res.payload = payload.toOwnedSlice(st.allocator) catch @panic("OutOfMemory"); } - // Sort by (slot, fec_set_index) and chain-validate. + // Emit in (slot, fec_set_index) order, dropping any FEC set that + // admission checks (above) or `deriveBlockParseResult` marked bad. std.sort.heap(FECSetParseResult, st.fec_set_results.items, {}, fecOrder); - // Cross-FEC chained_merkle_root check (not enforced by the Receiver). var i: usize = 0; while (i < st.fec_set_results.items.len) { - // Contiguous run of FEC sets for one slot. const slot = st.fec_set_results.items[i].slot; var j = i + 1; while (j < st.fec_set_results.items.len and @@ -432,20 +544,31 @@ fn buildProtoEffects( {} var expected_idx: u32 = 0; - var prev_merkle: ?Hash = null; for (st.fec_set_results.items[i..j]) |*r| { // Gap or out-of-order -> remaining sets in this slot are dropped. if (r.fec_set_index != expected_idx) { out.block_parse_result = .REJECTED_INVALID_HEADER; break; } - if (prev_merkle) |pm| { - if (!pm.eql(&r.chained_merkle_root)) { - out.block_parse_result = .REJECTED_INVALID_HEADER; - break; + + // Suppress FEC sets whose parent_offset disagrees with the + // per-slot pin (agave rejects these at insertion). + if (first_seen_parent.get(r.slot)) |pinned| { + if (pinned != r.parent_offset) { + expected_idx += FEC_DATA_SHREDS; + continue; } } + // Suppress FEC sets marked by the last-index walk above. + if (suppressed_fec_sets.contains(.{ + .slot = r.slot, + .fec_set_idx = r.fec_set_index, + })) { + expected_idx += FEC_DATA_SHREDS; + continue; + } + try out.fec_set_results.append(allocator, .{ .completed = true, .merkle_root = try allocator.dupe(u8, &r.merkle_root.data), @@ -459,7 +582,6 @@ fn buildProtoEffects( .num_coding_shreds = r.num_coding_shreds, }); - prev_merkle = r.merkle_root; expected_idx += FEC_DATA_SHREDS; } @@ -499,6 +621,30 @@ fn buildProtoEffects( const TickVerifyOutcome = enum { ok, rejected }; +/// Validates the fixed-shape header of an agave `VersionedBlockMarker`. +/// Consumes 5 or 6 bytes on success: +/// +/// - `VersionedBlockMarker`: u16 tag, only `1` (V1) is valid. +/// - `BlockMarkerV1`: u8 tag in +/// `{0: BlockFooter, 1: BlockHeader, 2: UpdateParent, 3: GenesisCert}`. +/// - `LengthPrefixed`: u16 length prefix (not enforced by agave). +/// - For markers 0..2: inner `Versioned` +/// u8 tag, only `1` (V1) is valid. Marker 3 has no version tag byte. +/// +/// Shape-only; deeper inner fields (BLS sigs, certs) are unmodelled. +fn validateBlockMarkerHeader(reader: *std.Io.Reader) bool { + const outer_tag = reader.takeInt(u16, .little) catch return false; + if (outer_tag != 1) return false; + const inner_tag = reader.takeByte() catch return false; + if (inner_tag > 3) return false; + _ = reader.takeInt(u16, .little) catch return false; + if (inner_tag <= 2) { + const versioned_tag = reader.takeByte() catch return false; + if (versioned_tag != 1) return false; + } + return true; +} + /// Per-slot tick verification driven by accepted data shreds. Walks the /// contiguous prefix `slot_idx = 0, 1, 2, ...`; each DATA_COMPLETE-bounded /// run concatenates to one bincode `Vec` record (one shredder @@ -521,9 +667,10 @@ fn verifyTicksFromDataShreds( defer allocator.free(fba_buf); var fba = std.heap.FixedBufferAllocator.init(fba_buf); - // slot_is_full := any captured shred carried LAST_SHRED_IN_SLOT. + // A LAST_SHRED_IN_SLOT flag only marks the slot full if it lies + // within the contiguous prefix from index 0. Set inside the + // gap-bounded walk so shreds past a gap don't flip it. var slot_is_full = false; - for (shreds) |s| slot_is_full = slot_is_full or s.last_shred_in_slot; var all_entries: std.ArrayListUnmanaged(sig_v2.solana.transaction.Entry) = .empty; // No deinit: storage is in `fba_buf`, freed via defer above. @@ -534,19 +681,21 @@ fn verifyTicksFromDataShreds( // Gap in the contiguous-from-zero prefix -> stop. Anything past a // gap is unreachable from `get_slot_entries_with_shred_info`. if (s.slot_idx != expected_idx) break; + if (s.last_shred_in_slot) slot_is_full = true; batch_buf.appendSlice(fba.allocator(), s.payload) catch return .rejected; if (s.data_complete) { - // A data-complete batch is one shredder batch, encoded as a single - // bincode `Vec` record. `get_slot_entries_with_shred_info` - // deserializes each completed set and surfaces any decode failure as - // a rejected block. A zero-byte batch is not valid bincode (the u64 - // length prefix needs 8 bytes) and rejects here too. + // Each data-complete batch is one bincode `BlockComponent`: + // a `Vec` with a u64 length prefix, followed by a + // `VersionedBlockMarker` when the length is 0. var reader: std.Io.Reader = .fixed(batch_buf.items); const entries = sig_v2.solana.bincode.read( &fba, &reader, sig_v2.solana.bincode.Vec(sig_v2.solana.transaction.Entry), ) catch return .rejected; + if (entries.items.len == 0) { + if (!validateBlockMarkerHeader(&reader)) return .rejected; + } for (entries.items) |e| { all_entries.append(fba.allocator(), e) catch return .rejected; } @@ -595,6 +744,102 @@ fn verifyTicksFromDataShreds( return .ok; } +// --------------------------------------------------------------------------- +// deriveBlockParseResult: block-parse verdict from replay + receiver state +// --------------------------------------------------------------------------- + +const DeadSlots = std.AutoHashMapUnmanaged(Slot, void); + +/// Cross-FEC-set invariant checks that mark slots dead, reading both +/// `MerkleForest` (completed sets) and `receiver.in_progress` (partial +/// sets): +/// - Equivocation: two nodes in `forest.map` sharing a `FecSetId` +/// (defensive — normally caught at admission). +/// - `parent_offset` divergence across in-progress ctxs of the same slot. +/// - SIMD-0340 chain: `chained_merkle_root` of `(slot, k+32)` must equal +/// `merkle_root` of `(slot, k)`. +/// +/// Iterates `id_map` because signature-collision under fuzz can hide the +/// older ctx from `signature_map`. +fn deriveBlockParseResult( + receiver: *const Receiver, + forest: *const MerkleForest, + scratch_allocator: Allocator, + dead_slots: *DeadSlots, +) !void { + // Equivocation: any two nodes in `forest.map` sharing a `FecSetId`. + var seen_ids: std.AutoHashMapUnmanaged(shred_api.FecSetId, void) = .empty; + defer seen_ids.deinit(scratch_allocator); + for (forest.map.values()) |node| { + const gop_id = try seen_ids.getOrPut(scratch_allocator, node.id); + if (gop_id.found_existing) + try dead_slots.put(scratch_allocator, node.id.slot, {}); + } + + // Per-slot `parent_offset` consistency across in-progress ctxs. + var slot_parents: std.AutoHashMapUnmanaged(Slot, u16) = .empty; + defer slot_parents.deinit(scratch_allocator); + var ctx_it = receiver.in_progress.id_map.valueIterator(); + while (ctx_it.next()) |ctx_ptr| { + const ctx = ctx_ptr.*; + if (ctx.data_variant.isData()) { + var i: u32 = 0; + while (i < FEC_DATA_SHREDS) : (i += 1) { + if (!ctx.data_shreds_received.isSet(i)) continue; + const shred: *const Shred = + .fromBufferUnchecked(&ctx.data_shreds_buf[i]); + const gop = try slot_parents.getOrPut(scratch_allocator, shred.slot); + if (gop.found_existing) { + if (gop.value_ptr.* != shred.code_or_data.data.parent_offset) + try dead_slots.put(scratch_allocator, shred.slot, {}); + } else { + gop.value_ptr.* = shred.code_or_data.data.parent_offset; + } + } + } + } + + // SIMD-0340 cross-FEC chain check. Reconstruct a + // `(slot, fec_set_idx) -> (merkle_root, chained_merkle_root)` map + // from `in_progress` + `forest.map` — together they cover every FEC + // set that saw a shred. + var pins: std.AutoHashMapUnmanaged(shred_api.FecSetId, struct { + merkle_root: Hash, + chained_merkle_root: Hash, + }) = .empty; + defer pins.deinit(scratch_allocator); + var pin_ctx_it = receiver.in_progress.id_map.valueIterator(); + while (pin_ctx_it.next()) |ctx_ptr| { + const ctx = ctx_ptr.*; + const id = receiver.in_progress.fecSetIdOf(ctx); + const gop = try pins.getOrPut(scratch_allocator, id); + if (!gop.found_existing) gop.value_ptr.* = .{ + .merkle_root = ctx.merkle_root, + .chained_merkle_root = ctx.chained_merkle_root, + }; + } + for (forest.map.values()) |node| { + const gop = try pins.getOrPut(scratch_allocator, node.id); + if (!gop.found_existing) gop.value_ptr.* = .{ + .merkle_root = node.merkle_root, + .chained_merkle_root = node.chained_merkle_root, + }; + } + var pin_it = pins.iterator(); + while (pin_it.next()) |entry| { + const key = entry.key_ptr.*; + const roots = entry.value_ptr.*; + const next_id: shred_api.FecSetId = .{ + .slot = key.slot, + .fec_set_idx = key.fec_set_idx + FEC_DATA_SHREDS, + }; + if (pins.get(next_id)) |next_roots| { + if (!next_roots.chained_merkle_root.eql(&roots.merkle_root)) + try dead_slots.put(scratch_allocator, key.slot, {}); + } + } +} + // --------------------------------------------------------------------------- // Self-test // --------------------------------------------------------------------------- diff --git a/v2/components/replay/component.zig b/v2/components/replay/component.zig index ce26961e27..0ee2a82095 100644 --- a/v2/components/replay/component.zig +++ b/v2/components/replay/component.zig @@ -4,3 +4,15 @@ //! deserialisation state, account-fetching cache, and exec scheduling loop. //! Those implementation details should eventually live in this component, //! leaving the service file as orchestration around shared-memory regions. + +comptime { + if (@import("builtin").is_test) { + _ = @import("forest.zig"); + } +} + +pub const api = @import("api"); + +pub const MerkleNode = @import("forest.zig").MerkleNode; +pub const MerkleForest = @import("forest.zig").MerkleForest; +pub const insertFecSet = @import("forest.zig").insertFecSet; diff --git a/v2/components/replay/forest.zig b/v2/components/replay/forest.zig new file mode 100644 index 0000000000..236d2b24e5 --- /dev/null +++ b/v2/components/replay/forest.zig @@ -0,0 +1,580 @@ +const std = @import("std"); +const tracy = @import("tracy"); +const lib = @import("lib"); +const api = @import("api"); +const shred = @import("shred_api"); + +const Hash = lib.solana.Hash; +const Pool = lib.collections.Pool; +const telemetry = lib.telemetry; + +const Shred = shred.Shred; +const FecSetId = shred.FecSetId; + +const BlockPool = api.BlockPool; +const BlockRef = api.BlockRef; + +/// Represents a deshredded FEC set. +/// +/// Used as a hashmap value, and a tree node (these are the same memory) +/// This node is also used for the keys of hashmaps. When doing so, be careful of which adapted +/// context you use. +/// +/// NOTE: When used inside the Pool, these may be items in a free list. However such nodes should +/// not be in either map or the tree. +pub const MerkleNode = extern struct { + parent: MerkleForest.NodePool.ItemId.Optional = .null, + child: MerkleForest.NodePool.ItemId.Optional = .null, + sibling: MerkleForest.NodePool.ItemId.Optional = .null, + + merkle_root: Hash, + chained_merkle_root: Hash, + id: FecSetId, + parent_offset: u16, + data_complete: bool, + slot_complete: bool, + + // allocated upon insertion of 1st fec set, copied down through children + // TODO: eviction + block_ref: BlockRef.Optional, + + payload_len: u16, + + // TODO: pool the payload buffer out-of-line; MerkleNode-in-map benefits + // from cache locality. + payload_buf: [32 * Shred.data_payload_max]u8, + + pub fn payload(node: *const MerkleNode) []const u8 { + return node.payload_buf[0..node.payload_len]; + } + + pub fn format(node: *const MerkleNode, writer: *std.io.Writer) !void { + try writer.print( + \\ {{ + \\ id: {}, slot_complete: {} + \\ parent: {}, child: {}, sibling: {} + \\ root: {f}, chained_root: {f} + \\ data_complete: {}, slot_complete: {} + \\ block_ref: {} + \\ }} + \\ + , .{ + node.id, + node.slot_complete, + node.parent, + node.child, + node.sibling, + node.merkle_root, + node.chained_merkle_root, + node.data_complete, + node.slot_complete, + node.block_ref, + }); + } +}; + +// TODO: handle eviction +/// A tree of FEC sets, which are also keyed by their merkle (and chained) merkle roots. +pub const MerkleForest = struct { + // owns all of the memory of nodes used in the map/tree nodes + pool: NodePool, + + // Nodes are inserted, keyed by their merkle root. + // New nodes can look for their parent using this map. + // + // merkle-hash -> node + map: MerkleMap, + + // Nodes are inserted, keyed by their *chained* merkle root. + // New nodes can look for their child using this map. + // + // chained-merkle-hash -> node + orphan_map: OrphanMap, + + pub const capacity = 4096; + + // keyed by merkle root + pub const OrphanMap = std.ArrayHashMapUnmanaged(void, *MerkleNode, OrphanContext, true); + + // keyed by chained merkle root + pub const MerkleMap = std.ArrayHashMapUnmanaged(void, *MerkleNode, MerkleContext, true); + + pub const NodePool = Pool(MerkleNode, u32); + + pub const MerkleContext = struct { + map: *const MerkleMap, + + pub fn hash(ctx: MerkleContext, key: *const Hash) u32 { + _ = ctx; + return @bitCast(key.data[0..4].*); + } + pub fn eql(ctx: MerkleContext, a: *const Hash, _: void, key_idx: usize) bool { + const b: *const Hash = &ctx.map.values()[key_idx].merkle_root; + return a.eql(b); + } + }; + + pub const OrphanContext = struct { + map: *const OrphanMap, + + pub fn hash(ctx: OrphanContext, key: *const Hash) u32 { + _ = ctx; + return @bitCast(key.data[0..4].*); + } + pub fn eql(ctx: OrphanContext, a: *const Hash, _: void, key_idx: usize) bool { + const b: *const Hash = &ctx.map.values()[key_idx].chained_merkle_root; + return a.eql(b); + } + }; + + pub fn init(allocator: std.mem.Allocator) !MerkleForest { + const pool_buf = try allocator.alloc(MerkleNode, capacity); + errdefer allocator.free(pool_buf); + + var map: MerkleMap = .empty; + errdefer map.deinit(allocator); + try map.ensureTotalCapacity(allocator, capacity); + + var orphan_map: OrphanMap = .empty; + errdefer orphan_map.deinit(allocator); + try orphan_map.ensureTotalCapacity(allocator, capacity); + + return .{ + // NOTE: the pool and the tree share the exact same buffer - this is intentional + .pool = .init(pool_buf[0..capacity]), + .map = map, + .orphan_map = orphan_map, + }; + } + + pub fn deinit(self: *MerkleForest, allocator: std.mem.Allocator) void { + allocator.free(self.pool.buf[0..self.pool.len]); + self.map.deinit(allocator); + self.orphan_map.deinit(allocator); + } + + pub fn reset(self: *MerkleForest) void { + self.pool.reset(); + self.map.clearRetainingCapacity(); + self.orphan_map.clearRetainingCapacity(); + } + + fn assertCounts(self: *const MerkleForest) void { + std.debug.assert(self.orphan_map.count() <= self.map.count()); + tracy.plot(u32, "Merkle forest fec sets", @intCast(self.map.count())); + tracy.plot(u32, "Merkle forest fec sets (orphaned)", @intCast(self.orphan_map.count())); + } +}; + +/// Finds a node's parent, and attaches the new node to it. +/// +/// In the case of a missing parent, also adds the current node (keyed by the parent's merkle root) +/// into the orphan map. +/// +/// NOTE: when removing nodes from the orphan map, make sure to handle all sibling nodes. +fn attachParent( + logger: telemetry.Logger("main"), + node: *MerkleNode, + forest: *MerkleForest, +) ?*MerkleNode { + const zone = tracy.Zone.init(@src(), .{ .name = "attachParent" }); + defer zone.deinit(); + + std.debug.assert(node.parent == .null); + const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; + const orphan_map_ctx: MerkleForest.OrphanContext = .{ .map = &forest.orphan_map }; + + const parent = forest.map.getAdapted(&node.chained_merkle_root, map_ctx) orelse { + zone.text("parent not found"); + + // No parent found, insert current node into orphan map (keyed by its parent's merkle-root) + // NOTE: it's probably a good idea to preemptively send repair requests for the parent here. + + const orphan_map_result = forest.orphan_map.getOrPutAssumeCapacityAdapted( + &node.chained_merkle_root, + orphan_map_ctx, + ); + + logger.info().logf( + "inserting ({}:{}) into orphans, already other orphans with same parent?: {}", + .{ node.id.slot, node.id.fec_set_idx, orphan_map_result.found_existing }, + ); + + if (orphan_map_result.found_existing) { + @branchHint(.unlikely); + // this could happen under equivocation or forking, should be unlikely to hit this. + // i.e. there's multiple fec sets currently missing the same parent + + // NOTE: the code that finds this orphan entry later *must* attach all of the orphan's + // siblings + + // insert at tail of the existing node's siblings + var orphan_sibling_tail: ?*MerkleNode = orphan_map_result.value_ptr.*; + while (orphan_sibling_tail) |tail_node| { + const next_sibling = (tail_node.sibling.opt() orelse break).ptr(&forest.pool); + // Adjacent orphans may share a FecSetId under leader + // equivocation; they'll have distinct merkle_roots and thus + // distinct entries in `forest.map`. + orphan_sibling_tail = next_sibling; + } + std.debug.assert(orphan_sibling_tail.?.sibling == .null); + orphan_sibling_tail.?.sibling = .init(forest.pool.ptrToIndex(node)); + } else { + orphan_map_result.value_ptr.* = node; // TODO: double check this + } + + return null; // no parent found + }; + + if (!parent.id.mayFollowWith(&node.id)) { + @branchHint(.cold); // this would be malicious behaviour, chaining in an invalid order + zone.text("mayFollowWith check failed"); + + return null; // parent found, but fec set ids don't align + } + + // insert node into tree of parent + { + std.debug.assert(node.parent == .null); + node.parent = .init(forest.pool.ptrToIndex(parent)); + + if (parent.child == .null) { + @branchHint(.likely); // no equivocation or forking + + parent.child = .init(forest.pool.ptrToIndex(node)); + return parent; + } + + std.debug.assert(parent.child != .null); + var last_child_of_parent: *MerkleNode = parent.child.opt().?.ptr(&forest.pool); + while (true) { + // Children may share a FecSetId under leader equivocation; + // distinct merkle_roots keep them as distinct forest entries. + const next = (last_child_of_parent.sibling.opt() orelse break).ptr(&forest.pool); + last_child_of_parent = next; + } + + last_child_of_parent.sibling = .init(forest.pool.ptrToIndex(node)); + return parent; + } +} + +fn attachChildren(node: *MerkleNode, forest: *MerkleForest) void { + std.debug.assert(node.child == .null); + + const zone = tracy.Zone.init(@src(), .{ .name = "attachChildren" }); + defer zone.deinit(); + + const orphan_map_ctx: MerkleForest.OrphanContext = .{ .map = &forest.orphan_map }; + const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; + + var children_head: ?*MerkleNode = forest.orphan_map.getAdapted( + &node.merkle_root, + orphan_map_ctx, + ) orelse return; + + // Iterate over all child nodes, setting their parent node, and deleting any bad child nodes + { + var maybe_child_node: ?*MerkleNode = children_head; + while (maybe_child_node) |child_node| { + std.debug.assert(child_node.parent == .null); + maybe_child_node = if (child_node.sibling.opt()) |s| s.ptr(&forest.pool) else null; + + if (node.id.mayFollowWith(&child_node.id)) { + @branchHint(.likely); + child_node.parent = .init(forest.pool.ptrToIndex(node)); + + continue; + } + + // Invalid child chaining implies leader malice; unlink the + // orphan sibling and drop it. + const next_node = if (child_node.sibling.opt()) |s| s.ptr(&forest.pool) else null; + const prev_node: ?*MerkleNode = node: { + if (child_node == children_head) break :node null; + + var prev_node: ?*MerkleNode = children_head; + break :node while (prev_node) |n| { + const next = if (n.sibling.opt()) |s| s.ptr(&forest.pool) else null; + if (next == child_node) break n; + prev_node = next.?; + } else unreachable; // either we're the head node, or we have a prev + }; + + if (prev_node) |prev| { + prev.sibling = if (next_node) |next| + .init(forest.pool.ptrToIndex(next)) + else + .null; + } else { + // head of list is invalid, let's move it forward + children_head = next_node; + } + + if (child_node.child != .null) + // Remove the node from the map + delete it + // if this orphan has invalid chaining, this means *all* of its children are also invalid + @panic("TODO: handle recursive removal from invalid orphan chaining"); + const removed = forest.map.swapRemoveAdapted(&child_node.merkle_root, map_ctx); + std.debug.assert(removed); + forest.pool.destroy(child_node); + } + } + + if (children_head) |c_h| node.child = .init(forest.pool.ptrToIndex(c_h)); + + // NOTE: this 2nd map lookup could be removed (we looked up this entry earlier) + const removed = forest.orphan_map.swapRemoveAdapted(&node.merkle_root, orphan_map_ctx); + std.debug.assert(removed); +} + +// Either: +// a) does nothing (parent has no BlockRef) +// b) allocates a new BlockRef due to reaching the slot boundary +// c) allocates a new BlockRef due to the parent already having a child (forking/equivocation) +// d) carries the parent's BlockRef forward (same slot + no forking/equivocation) +fn setChildBlockRef( + parent: *const MerkleNode, + child: *MerkleNode, + forest_pool: *MerkleForest.NodePool, + block_pool: *BlockPool, +) !void { + std.debug.assert(child.block_ref == .null); + const parent_block_ref = parent.block_ref.opt() orelse return; // a) + + // optionally allocate a new BlockRef + child.block_ref = if (parent.id.slot != child.id.slot) ref: { + // new slot, let's create a new BlockRef + const new_block = try block_pool.create(); + new_block.* = .{ + .parent = .init(parent_block_ref), + .slot = .init(child.id.slot), + }; + + break :ref .init(block_pool.ptrToIndex(new_block)); // b) + } else ref: { + // treat the first child as the canonical path + if (parent.child.opt()) |child_id| if (child_id == forest_pool.ptrToIndex(child)) { + break :ref .init(parent_block_ref); // d) + }; + + // forking/equivocation + const new_block = try block_pool.create(); + new_block.* = .{ + .parent = .init(parent_block_ref), + .slot = .init(child.id.slot), + }; + + break :ref .init(block_pool.ptrToIndex(new_block)); // c) + + }; +} + +fn setChildTreeBlockRefs( + parent: *MerkleNode, + child: *MerkleNode, + forest_pool: *MerkleForest.NodePool, + block_pool: *BlockPool, +) !void { + const zone = tracy.Zone.init(@src(), .{ .name = "setChildTreeBlockRefs" }); + defer zone.deinit(); + + // If we have a BlockRef we must have a parent (except in the case of an evicted parent, which + // doesn't apply here) + std.debug.assert(child.parent != .null); + + try setChildBlockRef(parent, child, forest_pool, block_pool); + + // recursively apply BlockRefs to reachable merkle nodes + // NOTE: it is possible to do this without recursion *or* a stack, as a non-null block_ref can + // be used to mark a node as visited. + var maybe_child = if (child.child.opt()) |id| id.ptr(forest_pool) else null; + while (maybe_child) |child_node| { + try setChildTreeBlockRefs(child, child_node, forest_pool, block_pool); + maybe_child = if (child_node.sibling.opt()) |id| id.ptr(forest_pool) else null; + } +} + +pub fn insertFecSet( + logger: telemetry.Logger("main"), + // to be transformed and inserted into the forest + deshredded_node: *const shred.DeshreddedFecSet, + forest: *MerkleForest, + // block associated parameters + // additional blocks may be allocated when inserting a fec set + block_pool: *BlockPool, +) error{OutOfSpace}!?*MerkleNode { + const zone = tracy.Zone.init(@src(), .{ .name = "insertFecSet" }); + defer zone.deinit(); + + forest.assertCounts(); + defer forest.assertCounts(); + + const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; + + const node: *MerkleNode = newly_inserted: { + const map_result = forest.map.getOrPutAssumeCapacityAdapted( + &deshredded_node.merkle_root, + map_ctx, + ); + if (map_result.found_existing) return null; // node already known + + const node = try forest.pool.create(); + map_result.value_ptr.* = node; + + node.* = .{ + .merkle_root = deshredded_node.merkle_root, + .chained_merkle_root = deshredded_node.chained_merkle_root, + .id = deshredded_node.id, + .parent_offset = deshredded_node.parent_offset, + .data_complete = deshredded_node.data_complete, + .slot_complete = deshredded_node.slot_complete, + + .block_ref = .null, + + .payload_len = deshredded_node.payload_len, + .payload_buf = deshredded_node.payload_buf, + }; + + break :newly_inserted node; + }; + + const maybe_parent = attachParent(logger, node, forest); + attachChildren(node, forest); + + // "propagate" BlockRefs (see `setChildBlockRef` for details) + if (maybe_parent) |parent| { + try setChildTreeBlockRefs(parent, node, &forest.pool, block_pool); + } + + return node; +} + +test "MerkleForest tree put" { + var tree: MerkleForest = try .init(std.testing.allocator); + defer tree.deinit(std.testing.allocator); + + const a_hash: Hash = .parse("ByzshhkRgXWnTkHjapkkqaKgEFnsg8ceY3bw4MWBzFE"); + const b_hash: Hash = .parse("BMHr4knWhDp8JhqCYhA2K5DUYQsYUVXdy2zWahzt5jLd"); + const c_hash: Hash = .parse("2GyMeUytf6fcsfNP2QQ6F5e5qwAUoMtKUbnH6QU6bTNm"); + const d_hash: Hash = .parse("4UahX8LzYC7xnubvP9QzRHmPPYovtcNYo7rBXKpp3ADM"); + const e_hash: Hash = .parse("An7mDXKMpRninZw6rvqc4wnQ6ukqd3ARko6QmPitjx8B"); + + const a: shred.DeshreddedFecSet = .{ + .chained_merkle_root = .parse("DWCWjQciWoWDzJKwqUZ1ntKqTyXtLVt4C8aL7biBJZ4z"), // prev slot + .merkle_root = a_hash, + + .id = .{ .slot = 409284941, .fec_set_idx = 0 }, + + .parent_offset = 1, + .data_complete = true, + .slot_complete = false, + + .payload_len = 0, + .payload_buf = undefined, + }; + + const b: shred.DeshreddedFecSet = .{ + .chained_merkle_root = a_hash, + .merkle_root = b_hash, + + .id = .{ .slot = 409284941, .fec_set_idx = 32 }, + + .parent_offset = 1, + .data_complete = true, + .slot_complete = false, + + .payload_len = 0, + .payload_buf = undefined, + }; + + const c: shred.DeshreddedFecSet = .{ + .chained_merkle_root = b_hash, + .merkle_root = c_hash, + + .id = .{ .slot = 409284941, .fec_set_idx = 64 }, + + .parent_offset = 1, + .data_complete = true, + .slot_complete = false, + + .payload_len = 0, + .payload_buf = undefined, + }; + + const d: shred.DeshreddedFecSet = .{ + .chained_merkle_root = c_hash, + .merkle_root = d_hash, + + .id = .{ .slot = 409284941, .fec_set_idx = 96 }, + + .parent_offset = 1, + .data_complete = true, + .slot_complete = true, + + .payload_len = 0, + .payload_buf = undefined, + }; + + // new slot + const e: shred.DeshreddedFecSet = .{ + .chained_merkle_root = d_hash, + .merkle_root = e_hash, + + .id = .{ .slot = 409284942, .fec_set_idx = 0 }, + + .parent_offset = 1, + .data_complete = true, + .slot_complete = true, + + .payload_len = 0, + .payload_buf = undefined, + }; + + var pool_buf: [BlockPool.size()]u8 align(@alignOf(BlockPool)) = undefined; + const pool: *BlockPool = @ptrCast(&pool_buf); + pool.init(); + + const logger = telemetry.Logger("main").noop; + + const a_inserted = (try insertFecSet(logger, &a, &tree, pool)).?; + try std.testing.expect(a_inserted.parent == .null); + try std.testing.expect(a_inserted.child == .null); + try std.testing.expect(a_inserted.block_ref == .null); + // give the ancestor block a BlockRef, so that it may propagate + // NOTE: it is expected that the root-most fec set to be inserted first this way as a special + // case. In a real environment this would be the last fec set in the rooted slot. + a_inserted.block_ref = .init(BlockRef.fromInt(8053)); + + const expected_block_ref: BlockRef.Optional = .init(BlockRef.fromInt(8053)); + + const d_inserted = (try insertFecSet(logger, &d, &tree, pool)).?; + try std.testing.expect(d_inserted.parent == .null); + try std.testing.expect(d_inserted.child == .null); + try std.testing.expect(d_inserted.block_ref == .null); // no path to a => null + + const b_inserted = (try insertFecSet(logger, &b, &tree, pool)).?; + try std.testing.expect(b_inserted.parent != .null); + try std.testing.expect(b_inserted.child == .null); + try std.testing.expect(b_inserted.block_ref == expected_block_ref); + + const c_inserted = (try insertFecSet(logger, &c, &tree, pool)).?; + try std.testing.expect(c_inserted.parent != .null); + try std.testing.expect(c_inserted.child != .null); + try std.testing.expect(c_inserted.block_ref == expected_block_ref); + try std.testing.expect(d_inserted.block_ref == expected_block_ref); + + const e_inserted = (try insertFecSet(logger, &e, &tree, pool)).?; + try std.testing.expect(e_inserted.parent != .null); + try std.testing.expect(e_inserted.child == .null); + // new slot => new BlockRef + try std.testing.expect(e_inserted.block_ref != .null); + try std.testing.expect(e_inserted.block_ref != expected_block_ref); + + // We cannot insert duplicates + try std.testing.expectEqual(null, try insertFecSet(logger, &a, &tree, pool)); + try std.testing.expectEqual(null, try insertFecSet(logger, &b, &tree, pool)); + try std.testing.expectEqual(null, try insertFecSet(logger, &c, &tree, pool)); + try std.testing.expectEqual(null, try insertFecSet(logger, &d, &tree, pool)); + try std.testing.expectEqual(null, try insertFecSet(logger, &e, &tree, pool)); +} diff --git a/v2/components/shred/api.zig b/v2/components/shred/api.zig index 900e0227e6..b0de6af01e 100644 --- a/v2/components/shred/api.zig +++ b/v2/components/shred/api.zig @@ -95,6 +95,9 @@ pub const DeshreddedFecSet = extern struct { chained_merkle_root: Hash, /// set to a meaningless value for the bootstrap root id: FecSetId, + /// `slot - parent_slot`. Identical across a FEC set's data shreds + /// (merkle-hashed DataHeader field). + parent_offset: u16, data_complete: bool, slot_complete: bool, /// empty for the bootstrap root @@ -218,8 +221,10 @@ pub const Shred = extern struct { const min_header_size = @offsetOf(Shred, "code_or_data") + @min(@sizeOf(DataHeader), @sizeOf(CodeHeader)); - const min_size = 1203; - const max_size = 1228; + /// Wire packet length carrying a data shred. + pub const min_size = 1203; + /// Wire packet length carrying a code shred. + pub const max_size = 1228; // This might not be possible? But this definitely always works as an upper bound pub const data_payload_max = min_size - @sizeOf(DataHeader); @@ -250,7 +255,9 @@ pub const Shred = extern struct { const payload_size = shred.code_or_data.data.size - header_size; const effective_size = min_size; - if (effective_size < header_size + payload_size + trailer_size) + const total_size = + @as(u32, header_size) + @as(u32, payload_size) + @as(u32, trailer_size); + if (total_size > effective_size) return error.DataEffectiveSizeTooSmall; break :sizes .{ @@ -267,7 +274,6 @@ pub const Shred = extern struct { return error.PacketSizeUnderExpected3; if (shred.variant.isData()) { - // [firedancer] https://github.com/firedancer-io/firedancer/commit/4936f39676997d95e5d15772d3904e5942fa9864 const parent_offset = shred.code_or_data.data.parent_offset; const slot = shred.slot; const flags = shred.code_or_data.data.flags; @@ -278,12 +284,12 @@ pub const Shred = extern struct { // `discard_unexpected_data_complete_shreds` is enforced in // `Receiver.processPacket` (needs the activation slot). - // TODO: drop shreds with last_shred_in_slot that aren't the last data shred in the set. - if (parent_offset > slot) return error.BadOffset; - if ((slot != 0 and parent_offset == 0) or (slot > 1 and parent_offset == slot)) - return error.BadSlotOrParentOffset; + // `parent_offset == slot` chains to genesis (parent = 0); legal + // at any slot. Only `parent_offset == 0` at slot != 0 is illegal. + // [agave] https://github.com/anza-xyz/agave/blob/v4.1.0-rc.1/ledger/src/blockstore.rs#L6059-L6066 + if (slot != 0 and parent_offset == 0) return error.BadSlotOrParentOffset; if (shred.slot_idx < shred.fec_set_idx) return error.BadSlotIdx; } else { const code_header = shred.code_or_data.code; diff --git a/v2/components/shred/receiver.zig b/v2/components/shred/receiver.zig index dd5dfb6359..a90a614285 100644 --- a/v2/components/shred/receiver.zig +++ b/v2/components/shred/receiver.zig @@ -80,7 +80,7 @@ pub const Receiver = struct { self.root_slot = root_slot; self.max_slot = max_slot; - // TODO: this is where we would add code to prune entries outside of the new range. + // TODO: prune `in_progress` / `done` entries below the new root. } // TODO: report return values to observability @@ -137,6 +137,12 @@ pub const Receiver = struct { { return error.UnexpectedDataCompleteShred; } + // [agave] https://github.com/anza-xyz/agave/blob/v4.1.0-rc.1/ledger/src/shred/filter.rs#L344-L349 + if (shred.code_or_data.data.flags.last_shred_in_slot and + (shred.slot_idx + 1) % FecSetCtx.fec_shred_count != 0) + { + return error.MisalignedLastDataIndex; + } } if (shred.fec_set_idx % FecSetCtx.fec_shred_count != 0) return error.InvalidFecSetIdx; @@ -148,6 +154,12 @@ pub const Receiver = struct { } } + // [agave] https://github.com/anza-xyz/agave/blob/v4.1.0-rc.1/ledger/src/blockstore.rs#L6065 + if (shred.variant.isData()) { + const parent_slot = shred.slot - shred.code_or_data.data.parent_offset; + if (parent_slot < state.root_slot) return error.ShredParentBeforeRoot; + } + const fec_set_id: FecSetId = .{ .fec_set_idx = shred.fec_set_idx, .slot = shred.slot }; var buf: [128]u8 = undefined; @@ -158,11 +170,25 @@ pub const Receiver = struct { ); zone.text(str); - const fec_set_ctx = if (state.in_progress.getFecSetCtx( - &shred.signature, - )) |fec_set_ctx| existing_set: { - // fec set is already being built. This branch will be taken for 31/64 shreds (assuming - // zero packet loss). + // TODO: skip re-computation on the sig-hit fast path once we have + // incremental inclusion proofs against the ctx's pinned merkle root. + var shred_merkle_root: Hash = undefined; + try shred.merkleRoot(&shred_merkle_root); + + // Two-tier lookup: signature fast-path (31/64 shreds hit here), + // then `(slot, fec_set_idx)` fallback for shreds that arrived + // under a different signature but the same erasure set. See + // `InProgressSets.id_map`. + const resolved: ?*FecSetCtx = resolve: { + if (state.in_progress.getFecSetCtx(&shred.signature)) |ctx| { + if (state.in_progress.fecSetIdOf(ctx).eql(&fec_set_id)) + break :resolve ctx; + } + break :resolve state.in_progress.getCtxById(fec_set_id); + }; + + const fec_set_ctx = if (resolved) |fec_set_ctx| existing_set: { + // fec set is already being built. // variant should match that of the first recorded shred in the fec set if ((shred.variant.isData() and !shred.variant.eql(fec_set_ctx.data_variant)) or @@ -171,78 +197,45 @@ pub const Receiver = struct { return error.VariantMismatchFromFecSet; } - // The signature of a shred protects its merkle root. We now have a shred that matches a - // signature that we verified against a merkle root earlier - we just need to check if - // the merkle root is the same. - // - // Checking the signature again requires calculating the merkle root anyway, and is much - // more expensive (37us vs 1us on my CPU, as of writing). - // - // NOTE: firedancer optimises "inserting" shreds into fec sets using - // fd_bmtree_commitp_insert_with_proof, which may be of interest. - var shred_merkle_root: Hash = undefined; - try shred.merkleRoot(&shred_merkle_root); + // Matching root admits into the merged ctx; mismatch rejects. + // In production, a sig-map hit implies a matching root (the + // signature covers it); this branch only rejects via the + // id-map fallback or when signature verification is disabled. if (!shred_merkle_root.eql(&fec_set_ctx.merkle_root)) - // This failing implies that signature verification would fail, i.e. it isn't an - // equivocation problem. return error.MismatchedMerkleRoot; + if (!shred.chainedMerkleRoot().eql(&fec_set_ctx.chained_merkle_root)) + return error.MismatchedChainedMerkleRoot; break :existing_set fec_set_ctx; } else new_set: { - // fec set is not currently being built (likely finished already) + // No in-progress ctx at this signature or fec_set_id. switch (state.done.lookupStatus(fec_set_id, &shred.signature)) { // fec set isn't finished, this is a new set .missing => {}, // fec set was finished already, let's ignore it .matching_signature => return .fec_set_already_finished, - - // NOTE: when we detect equivocation at the shred level, we just drop the incoming - // shred. i.e. the first shred in a fet set "wins", and until it is fully built, - // all other conflicting shreds are dropped until the in-progress fec set is built. - // - // This is intention as it stops the leader from producing many equivocating shreds - // that would a) fill up our in-progress map, and b) starve our CPU from shred - // verification. - // - // TODO: once repair is implemented, repaired shreds should skip these checks to - // allow conflicting fec sets to be inside the in-progress map. We will need to do - // this to reliably repair when equivocation is detected. - .mismatching_signature => return error.EquivocationDifferentHashForSameFecSetId, - } - - // if we have this FecSetId with a different signature, this means - // equivocation has occured - if (state.in_progress.containsId(fec_set_id)) { - // NOTE: see above note. - return error.EquivocationMatchingFecSetWithDifferentSignatureAlreadyInProgress; + // Set completed under a different signature. Admit if + // the merkle roots agree; reject otherwise. + .mismatching_signature => { + const roots = state.done.getRoots(fec_set_id).?; + if (roots.merkle_root.eql(&shred_merkle_root)) + return .fec_set_already_finished; + return error.MerkleRootConflict; + }, } // This is the first shred of a new in-progress fec set. - - // The shred's merkle root must be calculated unconditionally. - const shred_merkle_root: Hash = blk: { - var shred_merkle_root: Hash = undefined; - - if (!build_options.debug_skip_shred_sig_verify) { - const slot_leader = leader_schedule.get(shred.slot) orelse { - logger.warn().logf("slot {} missing?\n", .{shred.slot}); - return error.UnknownLeader; - }; - - try shred.merkleRoot(&shred_merkle_root); - - try shred.signature.verify( - slot_leader, - &shred_merkle_root.data, - ); - } else { - // debug purposes only - try shred.merkleRoot(&shred_merkle_root); - } - - break :blk shred_merkle_root; - }; + if (!build_options.debug_skip_shred_sig_verify) { + const slot_leader = leader_schedule.get(shred.slot) orelse { + logger.warn().logf("slot {} missing?\n", .{shred.slot}); + return error.UnknownLeader; + }; + shred.signature.verify( + slot_leader, + &shred_merkle_root.data, + ) catch return error.SignatureVerificationFailed; + } const fec_set_ctx = try state.in_progress.createFecSetCtx(fec_set_id, &shred.signature); @@ -259,6 +252,7 @@ pub const Receiver = struct { shred.variant.swapType(), .merkle_root = shred_merkle_root, + .chained_merkle_root = shred.chainedMerkleRoot().*, .data_shreds_received = .initEmpty(), .code_shreds_received = .initEmpty(), @@ -270,9 +264,6 @@ pub const Receiver = struct { break :new_set fec_set_ctx; }; - // in the case that we just acquired a fec set, it is critical that we do not leak it - errdefer comptime unreachable; - zone.value(fec_set_ctx.totalShredsReceived()); tracy.plot(u8, "totalShredsReceived", fec_set_ctx.totalShredsReceived()); @@ -312,6 +303,18 @@ pub const Receiver = struct { // starting fec set reconstruction now // NOTE: as an optimisation we should reconstruct directly into the out buffer + // + // RS decode sets `data_shreds_received = .initFull()`; a bare error + // return past this point would leak a ctx whose bitset would trip + // `total <= fec_shred_count` on the next code shred at an unseen + // index. Retire on error so subsequent shreds take the `new_set` + // path against `done`. + errdefer state.in_progress.removeFinishedSet(fec_set_ctx); + + // Snapshot before RS decode overwrites `data_shreds_received`; the + // re-validation loop below iterates only over RS-recovered indices. + const wire_received = fec_set_ctx.data_shreds_received; + { const shreds_bitset, const shreds_reedsol_bufs = fec_set_ctx.erasureEncoded(); @@ -331,6 +334,46 @@ pub const Receiver = struct { std.debug.assert(fec_set_ctx.data_shreds_received.count() == FecSetCtx.data_shreds_max); + // Structural + ctx-field re-validation over the RS-recovered region + // only. Signature and merkle proof live in the trailer, outside RS + // protection; the ctx's pinned merkle root already fixes them. + // [agave] https://github.com/anza-xyz/agave/blob/5efbb99925939a740cf6ff06647257d100e3e286/ledger/src/blockstore.rs#L1628 + for (0..FecSetCtx.data_shreds_max) |idx| { + if (wire_received.isSet(idx)) continue; + var recovered_packet: Packet = .{ + .data = fec_set_ctx.data_shreds_buf[idx], + .len = Shred.min_size, + .addr = std.net.Address.initIp4(.{ 0, 0, 0, 0 }, 0), + }; + const recovered = Shred.fromPacketChecked(&recovered_packet) catch |err| { + logger.warn().logf( + "RS-recovered shred failed structural re-validation: " ++ + "slot={} fec_set_idx={} idx={} err={s}. Dropping FEC set.", + .{ shred.slot, shred.fec_set_idx, idx, @errorName(err) }, + ); + return error.RecoveredShredMalformed; + }; + if (recovered.slot != shred.slot or + recovered.fec_set_idx != shred.fec_set_idx or + !recovered.variant.isData() or + !recovered.variant.eql(fec_set_ctx.data_variant) or + recovered.slot_idx != shred.fec_set_idx + idx) + { + logger.warn().logf( + "RS-recovered shred header disagrees with ctx: slot={} " ++ + "fec_set_idx={} idx={} " ++ + "(recovered slot={} fec_set_idx={} slot_idx={} isData={}). " ++ + "Dropping FEC set.", + .{ + shred.slot, shred.fec_set_idx, idx, + recovered.slot, recovered.fec_set_idx, recovered.slot_idx, + recovered.variant.isData(), + }, + ); + return error.RecoveredShredMalformed; + } + } + // writing out deshredded fec set { const sending_zone = tracy.Zone.init(@src(), .{ .name = "writing deshredded" }); @@ -369,8 +412,13 @@ pub const Receiver = struct { finished.* = .{ .merkle_root = fec_set_ctx.merkle_root, - .chained_merkle_root = shred.chainedMerkleRoot().*, + .chained_merkle_root = fec_set_ctx.chained_merkle_root, .id = fec_set_id, + // All data shreds carry the same parent_offset; index 0 is + // populated by RS recovery above if the wire shred was missing. + .parent_offset = Shred.fromBufferUnchecked( + &fec_set_ctx.data_shreds_buf[0], + ).code_or_data.data.parent_offset, .data_complete = data_complete, .slot_complete = slot_complete, .payload_len = total_payload_len, @@ -379,7 +427,6 @@ pub const Receiver = struct { var bytes_written: u16 = 0; for (&fec_set_ctx.data_shreds_buf) |*buffer| { - // TODO: I think we need to re-validate the data shreds that we recovered const data_shred: *const Shred = Shred.fromBufferUnchecked(buffer); const payload = data_shred.dataPayload(); @memcpy(finished.payload_buf[bytes_written..][0..payload.len], payload); @@ -391,7 +438,12 @@ pub const Receiver = struct { std.debug.assert(bytes_written == total_payload_len); } - state.done.setDone(&shred.signature, fec_set_id); + state.done.setDone( + &shred.signature, + fec_set_id, + &fec_set_ctx.merkle_root, + &fec_set_ctx.chained_merkle_root, + ); state.in_progress.removeFinishedSet(fec_set_ctx); tracy.frameMarkNamed("finished FEC sets"); @@ -410,6 +462,12 @@ pub const Receiver = struct { }; }; +/// The two roots pinned for a completed FEC set. +pub const FecSetRoots = struct { + merkle_root: Hash, + chained_merkle_root: Hash, +}; + /// Represents a FEC (Forward Error Correction) set which has yet to be reconstructed. // TODO: use a separate pool for the packet buffers! We're using at least 2x the memory for these, // and are ruining our cache locality. @@ -428,6 +486,8 @@ pub const FecSetCtx = extern struct { // we store the first seen, and make sure later shreds have the same one merkle_root: Hash, + // Merkle root of the previous FEC set (SIMD-0340). First-seen wins. + chained_merkle_root: Hash, // https://github.com/firedancer-io/firedancer/blob/ecd2d6d8f5b9f926d0b9aa9360efe36ea1550ad6/src/ballet/reedsol/fd_reedsol.h#L23 // https://github.com/solana-foundation/specs/blob/main/p2p/shred.md @@ -516,6 +576,18 @@ const InProgressSets = struct { signatures: []Signature, // idx correspond with fecset idxs signature_map: SignatureMap, + /// Authoritative `(slot, fec_set_idx) → *FecSetCtx` index: exactly one + /// entry per live ctx. `signature_map` is a fast-path secondary index + /// keyed by signature and may drop entries when two ctxs collide on a + /// signature (see `assertCounts`). + /// + /// Lets us merge shreds of the same erasure set that arrived under + /// different signatures, which happens in two cases: + /// - Agave-parity equivocation: `check_merkle_root_consistency` is + /// signature-blind, so we admit both same-root variants regardless + /// of which signature they carry. + /// - Sig-verify-off fuzz: unrelated ctxs can collide on a signature. + id_map: IdMap, eviction: Eviction, const Eviction = std.PriorityQueue(Pool.ItemId, QueueContext, QueueContext.order); @@ -524,6 +596,7 @@ const InProgressSets = struct { // Key from signature-hash, rather than merkle root, as it's equivalent for lookup and we don't // have to compute it. const SignatureMap = std.ArrayHashMapUnmanaged(void, *FecSetCtx, SignatureContext, true); + const IdMap = std.AutoHashMapUnmanaged(FecSetId, *FecSetCtx); fn init(allocator: std.mem.Allocator, capacity: u32) !InProgressSets { const buf = try allocator.alloc(FecSetCtx, capacity); @@ -541,6 +614,10 @@ const InProgressSets = struct { errdefer signature_map.deinit(allocator); try signature_map.ensureTotalCapacity(allocator, capacity); + var id_map: IdMap = .empty; + errdefer id_map.deinit(allocator); + try id_map.ensureTotalCapacity(allocator, capacity); + var eviction: Eviction = .init(allocator, .{ .ids = ids }); errdefer eviction.deinit(); try eviction.ensureTotalCapacity(capacity); @@ -551,6 +628,7 @@ const InProgressSets = struct { .ids = ids, .signatures = signatures, .signature_map = signature_map, + .id_map = id_map, .eviction = eviction, }; } @@ -560,6 +638,7 @@ const InProgressSets = struct { allocator.free(self.ids); allocator.free(self.signatures); self.signature_map.deinit(allocator); + self.id_map.deinit(allocator); self.eviction.allocator = allocator; self.eviction.deinit(); @@ -571,6 +650,7 @@ const InProgressSets = struct { fn reset(self: *InProgressSets) void { self.ctx_pool.reset(); self.signature_map.clearRetainingCapacity(); + self.id_map.clearRetainingCapacity(); self.eviction.items.len = 0; } @@ -602,14 +682,18 @@ const InProgressSets = struct { // eviction can't be full, we *just* evicted self.eviction.add(new_pool_id) catch unreachable; self.signatures[new_idx] = signature.*; + // Sig-collision path: newer ctx claims the sig_map slot. See + // `assertCounts`. const result = self.signature_map.getOrPutAssumeCapacityAdapted(signature, map_ctx); - // can't create a fecsetctx that already exists - if (result.found_existing) unreachable; const node: *FecSetCtx = self.ctx_pool.indexToPtr(@enumFromInt(new_idx)); result.value_ptr.* = node; + const id_result = self.id_map.getOrPutAssumeCapacity(id); + if (id_result.found_existing) unreachable; // `new_set` in processPacket only fires when this id is unoccupied + id_result.value_ptr.* = node; + return node; } @@ -646,14 +730,29 @@ const InProgressSets = struct { const evicted_idx = evicted_pool_idx.index(); const evicted_sig: *Signature = &self.signatures[evicted_idx]; + const evicted_id: FecSetId = self.ids[evicted_idx]; + const evicted_ptr: *FecSetCtx = self.ctx_pool.indexToPtr(evicted_pool_idx); // const node: *FecSetCtx = @ptrCast(&self.ctx_pool.buf[evicted_idx]); self.ids[evicted_idx] = // an impossible FecSetID which can never be matched with .{ .slot = std.math.maxInt(Slot), .fec_set_idx = std.math.maxInt(u32) - 1 }; self.ctx_pool.destroyId(evicted_pool_idx); - const removed = self.signature_map.swapRemoveContextAdapted(evicted_sig, map_ctx, map_ctx); - std.debug.assert(removed); + // Only remove the sig_map entry if it currently points at this + // ctx: under sig-collision the newer ctx owns the slot. See + // `assertCounts`. + if (self.signature_map.getAdapted(evicted_sig, map_ctx)) |cur| { + if (cur == evicted_ptr) { + const removed = self.signature_map.swapRemoveContextAdapted( + evicted_sig, + map_ctx, + map_ctx, + ); + std.debug.assert(removed); + } + } + const id_removed = self.id_map.remove(evicted_id); + std.debug.assert(id_removed); evicted_sig.* = undefined; @@ -662,16 +761,26 @@ const InProgressSets = struct { } fn containsId(self: *const InProgressSets, id: FecSetId) bool { - return for (self.signature_map.values()) |fec_set_ctx| { - const pool_id = self.ctx_pool.ptrToIndex(fec_set_ctx); - const idx = pool_id.index(); + return self.id_map.contains(id); + } - if (self.ids[idx].eql(&id)) break true; - } else false; + fn getCtxById(self: *const InProgressSets, id: FecSetId) ?*FecSetCtx { + return self.id_map.get(id); + } + + /// `FecSetId` under which `ctx` was inserted. Only valid for a live + /// pointer from `getFecSetCtx` / `getCtxById`. + pub fn fecSetIdOf(self: *const InProgressSets, ctx: *FecSetCtx) FecSetId { + const pool_id = self.ctx_pool.ptrToIndex(ctx); + return self.ids[pool_id.index()]; } fn assertCounts(self: *const InProgressSets) void { - std.debug.assert(self.signature_map.count() == self.eviction.items.len); + // id_map is the authoritative primary; signature_map is a + // fast-path index that may lose entries under sig-collision + // (newer ctx claims the slot in `createFecSetCtx`). + std.debug.assert(self.id_map.count() == self.eviction.items.len); + std.debug.assert(self.signature_map.count() <= self.eviction.items.len); tracy.plot(u32, "in-progress FEC sets", @intCast(self.eviction.items.len)); } @@ -891,6 +1000,7 @@ test "InProgressSets basic usage" { // doesn't contain anything yet try std.testing.expect(!in_progress.containsId(set_id)); try std.testing.expectEqual(null, in_progress.getFecSetCtx(&Signature.ZEROES)); + try std.testing.expectEqual(null, in_progress.getCtxById(set_id)); // add set const ctx = try in_progress.createFecSetCtx(set_id, &set_signature); @@ -899,6 +1009,7 @@ test "InProgressSets basic usage" { const found_ctx = in_progress.getFecSetCtx(&set_signature) orelse unreachable; try std.testing.expectEqual(ctx, found_ctx); try std.testing.expect(in_progress.containsId(set_id)); + try std.testing.expectEqual(ctx, in_progress.getCtxById(set_id)); // context is evicted { @@ -958,7 +1069,13 @@ const DoneSets = struct { // This signature+id must not be inside DoneSet already - any shred inside DoneSets should be // dropped early, so setDone should be unreachable in this case. - fn setDone(self: *DoneSets, signature: *const Signature, id: FecSetId) void { + fn setDone( + self: *DoneSets, + signature: *const Signature, + id: FecSetId, + merkle_root: *const Hash, + chained_merkle_root: *const Hash, + ) void { const done_ctx: DoneContext = .{ .done_map = &self.done_map }; self.assertCounts(); @@ -981,7 +1098,12 @@ const DoneSets = struct { }; const new_done: *DoneItem = self.done_pool.indexToPtr(new_pool_id); - new_done.* = .{ .id = id, .signature_hashed = hashSignature(signature) }; + new_done.* = .{ + .id = id, + .signature_hashed = hashSignature(signature), + .merkle_root = merkle_root.*, + .chained_merkle_root = chained_merkle_root.*, + }; self.eviction.add(new_pool_id) catch unreachable; const entry = self.done_map.getOrPutAssumeCapacityAdapted(&id, done_ctx); std.debug.assert(!entry.found_existing); @@ -1003,12 +1125,27 @@ const DoneSets = struct { .mismatching_signature; } + /// Roots pinned for a completed FEC set, or null if `id` is unknown. + fn getRoots(self: *const DoneSets, id: FecSetId) ?FecSetRoots { + const done_ctx: DoneContext = .{ .done_map = &self.done_map }; + const entry = self.done_map.getAdapted(&id, done_ctx) orelse return null; + return .{ + .merkle_root = entry.merkle_root, + .chained_merkle_root = entry.chained_merkle_root, + }; + } + fn assertCounts(self: *const DoneSets) void { std.debug.assert(self.eviction.items.len == self.done_map.count()); tracy.plot(u32, "done FEC sets", @intCast(self.eviction.items.len)); } - const DoneItem = extern struct { signature_hashed: u32, id: FecSetId }; + const DoneItem = extern struct { + signature_hashed: u32, + id: FecSetId, + merkle_root: Hash, + chained_merkle_root: Hash, + }; const Eviction = std.PriorityQueue(Pool.ItemId, QueueContext, QueueContext.order); const Pool = lib.collections.Pool(DoneItem, u32); const DoneMap = std.ArrayHashMapUnmanaged(void, *DoneItem, DoneContext, true); @@ -1056,19 +1193,19 @@ test "DoneSets basic usage" { const id_2: FecSetId = .{ .slot = 2, .fec_set_idx = 0 }; const id_3: FecSetId = .{ .slot = 3, .fec_set_idx = 0 }; - done_sets.setDone(&sig_1, id_1); + done_sets.setDone(&sig_1, id_1, &Hash.ZEROES, &Hash.ZEROES); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_1, &sig_1)); try std.testing.expectEqual(.missing, done_sets.lookupStatus(id_2, &sig_2)); try std.testing.expectEqual(.missing, done_sets.lookupStatus(id_3, &sig_3)); - done_sets.setDone(&sig_2, id_2); + done_sets.setDone(&sig_2, id_2, &Hash.ZEROES, &Hash.ZEROES); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_1, &sig_1)); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_2, &sig_2)); try std.testing.expectEqual(.missing, done_sets.lookupStatus(id_3, &sig_3)); - done_sets.setDone(&sig_3, id_3); + done_sets.setDone(&sig_3, id_3, &Hash.ZEROES, &Hash.ZEROES); try std.testing.expectEqual(.missing, done_sets.lookupStatus(id_1, &sig_1)); // 1 was evicted try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_2, &sig_2)); @@ -1091,8 +1228,8 @@ test "DoneSets reset clears state without freeing" { const id_1: FecSetId = .{ .slot = 1, .fec_set_idx = 0 }; const id_2: FecSetId = .{ .slot = 2, .fec_set_idx = 0 }; - done_sets.setDone(&sig_1, id_1); - done_sets.setDone(&sig_2, id_2); + done_sets.setDone(&sig_1, id_1, &Hash.ZEROES, &Hash.ZEROES); + done_sets.setDone(&sig_2, id_2, &Hash.ZEROES, &Hash.ZEROES); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_1, &sig_1)); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_2, &sig_2)); @@ -1104,8 +1241,35 @@ test "DoneSets reset clears state without freeing" { // Capacity is retained — refilling to the original size must not allocate // (eviction.allocator is the testing failing allocator after init). - done_sets.setDone(&sig_1, id_1); - done_sets.setDone(&sig_2, id_2); + done_sets.setDone(&sig_1, id_1, &Hash.ZEROES, &Hash.ZEROES); + done_sets.setDone(&sig_2, id_2, &Hash.ZEROES, &Hash.ZEROES); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_1, &sig_1)); try std.testing.expectEqual(.matching_signature, done_sets.lookupStatus(id_2, &sig_2)); } + +test "DoneSets.getRoots returns the pinned roots" { + const allocator = std.testing.allocator; + + var done_sets: DoneSets = try .init(allocator, 4); + defer done_sets.deinit(allocator); + + const sig_1: Signature = .ZEROES; + const id_1: FecSetId = .{ .slot = 7, .fec_set_idx = 0 }; + + const merkle: Hash = .{ .data = @splat(0xAA) }; + const chained: Hash = .{ .data = @splat(0xBB) }; + + try std.testing.expectEqual(null, done_sets.getRoots(id_1)); + + done_sets.setDone(&sig_1, id_1, &merkle, &chained); + + const got = done_sets.getRoots(id_1) orelse return error.TestUnexpectedNull; + try std.testing.expect(got.merkle_root.eql(&merkle)); + try std.testing.expect(got.chained_merkle_root.eql(&chained)); + + // Unknown id still misses. + try std.testing.expectEqual( + null, + done_sets.getRoots(.{ .slot = 7, .fec_set_idx = 32 }), + ); +} diff --git a/v2/services/replay.zig b/v2/services/replay.zig index 55c571e9bd..884f6b2d7b 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -89,14 +89,10 @@ const shred = @import("shred_api"); const accounts_db = @import("accounts_db_api"); const api = @import("replay_api"); +const replay = @import("replay"); const Hash = lib.solana.Hash; -const Shred = shred.Shred; -const FecSetId = shred.FecSetId; - -const Pool = lib.collections.Pool; - comptime { _ = start; } @@ -122,7 +118,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n var fba: std.heap.FixedBufferAllocator = .init(rw.scratch_memory); const allocator = fba.allocator(); - var forest: MerkleForest = try .init(allocator); + var forest: replay.MerkleForest = try .init(allocator); const unrooted: *Unrooted = try allocator.create(Unrooted); unrooted.init(); @@ -224,7 +220,7 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n zone.value(deshredded_fec_set.id.slot); zone.value(deshredded_fec_set.id.fec_set_idx); - const inserted = (try insertFecSet( + const inserted = (try replay.insertFecSet( logger, deshredded_fec_set, &forest, @@ -294,7 +290,7 @@ fn bootstrap( logger: tel.Logger("main"), runner: lib.runner.Connection, snapshot_metadata: *accounts_db.RuntimeMetadata, - forest: *MerkleForest, + forest: *replay.MerkleForest, block_pool: *api.BlockPool, exec_states: *BlockExecStates, blockhash_states: *BlockHashStates, @@ -337,13 +333,14 @@ fn bootstrap( // create a synthetic fec-set node that doesn't have all information about // the fec set, but it is enough to get started processing the first block // after the root - const root_node = try insertFecSet(logger, &.{ + const root_node = try replay.insertFecSet(logger, &.{ .merkle_root = snapshot_metadata.block_id, .chained_merkle_root = .ZEROES, // cannot be determined from the snapshot .id = .{ .slot = root_slot, .fec_set_idx = 0, // cannot be determined from the snapshot }, + .parent_offset = 0, // cannot be determined from the snapshot .data_complete = true, .slot_complete = true, .payload_len = 0, // cannot be determined from the snapshot @@ -356,7 +353,7 @@ fn bootstrap( // whose `merkle_root` happens to equal `Hash.ZEROES`. std.debug.assert(forest.orphan_map.swapRemoveAdapted( &root_node.chained_merkle_root, - MerkleForest.OrphanContext{ .map = &forest.orphan_map }, + replay.MerkleForest.OrphanContext{ .map = &forest.orphan_map }, )); // Mark the root block as fully executed so `maybeContinueBlockExec` will immediately @@ -569,303 +566,15 @@ fn fetchBlocking( return response.account_index; } -/// Finds a node's parent, and attaches the new node to it. -/// -/// In the case of a missing parent, also adds the current node (keyed by the parent's merkle root) -/// into the orphan map. -/// -/// NOTE: when removing nodes from the orphan map, make sure to handle all sibling nodes. -fn attachParent( - logger: tel.Logger("main"), - node: *MerkleNode, - forest: *MerkleForest, -) ?*MerkleNode { - const zone = tracy.Zone.init(@src(), .{ .name = "attachParent" }); - defer zone.deinit(); - - std.debug.assert(node.parent == .null); - const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; - const orphan_map_ctx: MerkleForest.OrphanContext = .{ .map = &forest.orphan_map }; - - const parent = forest.map.getAdapted(&node.chained_merkle_root, map_ctx) orelse { - zone.text("parent not found"); - - // No parent found, insert current node into orphan map (keyed by its parent's merkle-root) - // NOTE: it's probably a good idea to preemptively send repair requests for the parent here. - - const orphan_map_result = forest.orphan_map.getOrPutAssumeCapacityAdapted( - &node.chained_merkle_root, - orphan_map_ctx, - ); - - logger.info().logf( - "inserting ({}:{}) into orphans, already other orphans with same parent?: {}", - .{ node.id.slot, node.id.fec_set_idx, orphan_map_result.found_existing }, - ); - - if (orphan_map_result.found_existing) { - @branchHint(.unlikely); - // this could happen under equivocation or forking, should be unlikely to hit this. - // i.e. there's multiple fec sets currently missing the same parent - - // NOTE: the code that finds this orphan entry later *must* attach all of the orphan's - // siblings - - // insert at tail of the existing node's siblings - var orphan_sibling_tail: ?*MerkleNode = orphan_map_result.value_ptr.*; - while (orphan_sibling_tail) |tail_node| { - const next_sibling = (tail_node.sibling.opt() orelse break).ptr(&forest.pool); - - if (next_sibling.id.eql(&tail_node.id)) @panic("equivocation"); - - orphan_sibling_tail = next_sibling; - } - std.debug.assert(orphan_sibling_tail.?.sibling == .null); - orphan_sibling_tail.?.sibling = .init(forest.pool.ptrToIndex(node)); - } else { - orphan_map_result.value_ptr.* = node; // TODO: double check this - } - - return null; // no parent found - }; - - if (!parent.id.mayFollowWith(&node.id)) { - @branchHint(.cold); // this would be malicious behaviour, chaining in an invalid order - zone.text("mayFollowWith check failed"); - - return null; // parent found, but fec set ids don't align - } - - // insert node into tree of parent - { - std.debug.assert(node.parent == .null); - node.parent = .init(forest.pool.ptrToIndex(parent)); - - if (parent.child == .null) { - @branchHint(.likely); // no equivocation or forking - - parent.child = .init(forest.pool.ptrToIndex(node)); - return parent; - } - - std.debug.assert(parent.child != .null); - var last_child_of_parent: *MerkleNode = parent.child.opt().?.ptr(&forest.pool); - while (true) { - // NOTE: We should get rid of this panic once we're confident that we're handling it - // correctly downstream. - if (last_child_of_parent.id.eql(&node.id)) @panic("equivocation"); - const next = (last_child_of_parent.sibling.opt() orelse break).ptr(&forest.pool); - last_child_of_parent = next; - } - - last_child_of_parent.sibling = .init(forest.pool.ptrToIndex(node)); - return parent; - } -} - -fn attachChildren(node: *MerkleNode, forest: *MerkleForest) void { - std.debug.assert(node.child == .null); - - const zone = tracy.Zone.init(@src(), .{ .name = "attachChildren" }); - defer zone.deinit(); - - const orphan_map_ctx: MerkleForest.OrphanContext = .{ .map = &forest.orphan_map }; - const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; - - var children_head: ?*MerkleNode = forest.orphan_map.getAdapted( - &node.merkle_root, - orphan_map_ctx, - ) orelse return; - - // Iterate over all child nodes, setting their parent node, and deleting any bad child nodes - { - var maybe_child_node: ?*MerkleNode = children_head; - while (maybe_child_node) |child_node| { - std.debug.assert(child_node.parent == .null); - maybe_child_node = if (child_node.sibling.opt()) |s| s.ptr(&forest.pool) else null; - - if (node.id.mayFollowWith(&child_node.id)) { - @branchHint(.likely); - child_node.parent = .init(forest.pool.ptrToIndex(node)); - - continue; - } - - // Invalid chaining should only happen if the leader is being malicious. - // I'm not sure if this branch will ever be hit, but we must still handle this case. - // This node is illegal, let's remove it now. - - // Remove this node from the linked list of siblings - const next_node = if (child_node.sibling.opt()) |s| s.ptr(&forest.pool) else null; - const prev_node: ?*MerkleNode = node: { - if (child_node == children_head) break :node null; - - var prev_node: ?*MerkleNode = children_head; - break :node while (prev_node) |n| { - const next = if (n.sibling.opt()) |s| s.ptr(&forest.pool) else null; - if (next == child_node) break n; - prev_node = next.?; - } else unreachable; // either we're the head node, or we have a prev - }; - - if (prev_node) |prev| { - prev.sibling = if (next_node) |next| - .init(forest.pool.ptrToIndex(next)) - else - .null; - } else { - // head of list is invalid, let's move it forward - children_head = next_node; - } - - if (child_node.child != .null) - // Remove the node from the map + delete it - // if this orphan has invalid chaining, this means *all* of - // its children are also invalid - @panic("TODO: handle recursive removal from invalid orphan chaining"); - const removed = forest.map.swapRemoveAdapted(&child_node.merkle_root, map_ctx); - std.debug.assert(removed); - forest.pool.destroy(child_node); - } - } - - if (children_head) |c_h| node.child = .init(forest.pool.ptrToIndex(c_h)); - - // NOTE: this 2nd map lookup could be removed (we looked up this entry earlier) - const removed = forest.orphan_map.swapRemoveAdapted(&node.merkle_root, orphan_map_ctx); - std.debug.assert(removed); -} - -// Either: -// a) does nothing (parent has no BlockRef) -// b) allocates a new BlockRef due to reaching the slot boundary -// c) allocates a new BlockRef due to the parent already having a child (forking/equivocation) -// d) carries the parent's BlockRef forward (same slot + no forking/equivocation) -fn setChildBlockRef( - parent: *const MerkleNode, - child: *MerkleNode, - forest_pool: *MerkleForest.NodePool, - block_pool: *api.BlockPool, -) !void { - std.debug.assert(child.block_ref == .null); - const parent_block_ref = parent.block_ref.opt() orelse return; // a) - - // optionally allocate a new BlockRef - child.block_ref = if (parent.id.slot != child.id.slot) ref: { - // new slot, let's create a new BlockRef - const new_block = try block_pool.create(); - new_block.* = .{ - .parent = .init(parent_block_ref), - .slot = .init(child.id.slot), - }; - - break :ref .init(block_pool.ptrToIndex(new_block)); // b) - } else ref: { - // treat the first child as the canonical path - if (parent.child.opt()) |child_id| if (child_id == forest_pool.ptrToIndex(child)) { - break :ref .init(parent_block_ref); // d) - }; - - // forking/equivocation - const new_block = try block_pool.create(); - new_block.* = .{ - .parent = .init(parent_block_ref), - .slot = .init(child.id.slot), - }; - - break :ref .init(block_pool.ptrToIndex(new_block)); // c) - - }; -} - -fn setChildTreeBlockRefs( - parent: *MerkleNode, - child: *MerkleNode, - forest_pool: *MerkleForest.NodePool, - block_pool: *api.BlockPool, -) !void { - const zone = tracy.Zone.init(@src(), .{ .name = "setChildTreeBlockRefs" }); - defer zone.deinit(); - - // If we have a BlockRef we must have a parent (except in the case of an evicted parent, which - // doesn't apply here) - std.debug.assert(child.parent != .null); - - try setChildBlockRef(parent, child, forest_pool, block_pool); - - // recursively apply BlockRefs to reachable merkle nodes - // NOTE: it is possible to do this without recursion *or* a stack, as a non-null block_ref can - // be used to mark a node as visited. - var maybe_child = if (child.child.opt()) |id| id.ptr(forest_pool) else null; - while (maybe_child) |child_node| { - try setChildTreeBlockRefs(child, child_node, forest_pool, block_pool); - maybe_child = if (child_node.sibling.opt()) |id| id.ptr(forest_pool) else null; - } -} - -fn insertFecSet( - logger: tel.Logger("main"), - // to be transformed and inserted into the forest - deshredded_node: *const shred.DeshreddedFecSet, - forest: *MerkleForest, - // block associated parameters - // additional blocks may be allocated when inserting a fec set - block_pool: *api.BlockPool, -) error{OutOfSpace}!?*MerkleNode { - const zone = tracy.Zone.init(@src(), .{ .name = "insertFecSet" }); - defer zone.deinit(); - - forest.assertCounts(); - defer forest.assertCounts(); - - const map_ctx: MerkleForest.MerkleContext = .{ .map = &forest.map }; - - const node: *MerkleNode = newly_inserted: { - const map_result = forest.map.getOrPutAssumeCapacityAdapted( - &deshredded_node.merkle_root, - map_ctx, - ); - if (map_result.found_existing) return null; // node already known - - const node = try forest.pool.create(); - map_result.value_ptr.* = node; - - node.* = .{ - .merkle_root = deshredded_node.merkle_root, - .chained_merkle_root = deshredded_node.chained_merkle_root, - .id = deshredded_node.id, - .data_complete = deshredded_node.data_complete, - .slot_complete = deshredded_node.slot_complete, - - .block_ref = .null, - - .payload_len = deshredded_node.payload_len, - .payload_buf = deshredded_node.payload_buf, - }; - - break :newly_inserted node; - }; - - const maybe_parent = attachParent(logger, node, forest); - attachChildren(node, forest); - - // "propagate" BlockRefs (see `setChildBlockRef` for details) - if (maybe_parent) |parent| { - try setChildTreeBlockRefs(parent, node, &forest.pool, block_pool); - } - - return node; -} - fn maybeContinueBlockExec( logger: tel.Logger("main"), // newly inserted node (or, rarely, when called recursively, the idx=0 ancestor of the block) - node: *MerkleNode, + node: *replay.MerkleNode, // the block_ref of the newly inserted node block_ref: api.BlockRef, // pools - forest_pool: *MerkleForest.NodePool, + forest_pool: *replay.MerkleForest.NodePool, block_pool: *api.BlockPool, transaction_pool: *api.TransactionPool, @@ -1154,7 +863,7 @@ fn maybeContinueBlockExec( } const BlockDeserialState = struct { - pos_node: *const MerkleNode, + pos_node: *const replay.MerkleNode, pos_offset: usize, n_transactions_left: ?u64, @@ -1169,7 +878,7 @@ const BlockDeserialState = struct { const Reader = struct { deserial_state: *BlockDeserialState, - merkle_pool: *const MerkleForest.NodePool, + merkle_pool: *const replay.MerkleForest.NodePool, bytes_consumed: usize = 0, fn currentReadableSlice(self: *Reader) []const u8 { @@ -1263,7 +972,7 @@ const BlockDeserialState = struct { } }; - fn init(root_node: *const MerkleNode) BlockDeserialState { + fn init(root_node: *const replay.MerkleNode) BlockDeserialState { std.debug.assert(root_node.block_ref != .null); std.debug.assert(root_node.id.fec_set_idx == 0); @@ -1280,13 +989,16 @@ const BlockDeserialState = struct { }; } - fn getReader(self: *BlockDeserialState, merkle_pool: *const MerkleForest.NodePool) Reader { + fn getReader( + self: *BlockDeserialState, + merkle_pool: *const replay.MerkleForest.NodePool, + ) Reader { return .{ .deserial_state = self, .merkle_pool = merkle_pool }; } fn nextTransaction( self: *BlockDeserialState, - merkle_pool: *const MerkleForest.NodePool, + merkle_pool: *const replay.MerkleForest.NodePool, tx_buf: *[1232]u8, ) !?[]const u8 { const zone = tracy.Zone.init(@src(), .{ .name = "nextTransaction" }); @@ -1306,7 +1018,7 @@ const BlockDeserialState = struct { fn nextTransactionInner( self: *BlockDeserialState, - merkle_pool: *const MerkleForest.NodePool, + merkle_pool: *const replay.MerkleForest.NodePool, tx_buf: *[1232]u8, ) !?[]const u8 { var reader = self.getReader(merkle_pool); @@ -1401,359 +1113,3 @@ const BlockExecState = struct { self.n_transactions_completed == self.n_transactions_requested; } }; - -/// Represents a deshredded FEC set. -/// -/// Used as a hashmap value, and a tree node (these are the same memory) -/// This node is also used for the keys of hashmaps. When doing so, be careful of which adapted -/// context you use. -/// -/// NOTE: When used inside the Pool, these may be items in a free list. However such nodes should -/// not be in either map or the tree. -const MerkleNode = extern struct { - parent: MerkleForest.NodePool.ItemId.Optional = .null, - child: MerkleForest.NodePool.ItemId.Optional = .null, - sibling: MerkleForest.NodePool.ItemId.Optional = .null, - - merkle_root: Hash, - chained_merkle_root: Hash, - id: FecSetId, - data_complete: bool, - slot_complete: bool, - - // allocated upon insertion of 1st fec set, copied down through children - // TODO: eviction - block_ref: api.BlockRef.Optional, - - payload_len: u16, - - // TODO: this shouldn't be copied, and should instead come in via a pool - // NOTE: it is an advantage for MerkleNode to be small! (cache locality for map lookup and tree - // traversal). - payload_buf: [32 * Shred.data_payload_max]u8, - - fn payload(node: *const MerkleNode) []const u8 { - return node.payload_buf[0..node.payload_len]; - } - - pub fn format(node: *const MerkleNode, writer: *std.io.Writer) !void { - try writer.print( - \\ {{ - \\ id: {}, slot_complete: {} - \\ parent: {}, child: {}, sibling: {} - \\ root: {f}, chained_root: {f} - \\ data_complete: {}, slot_complete: {} - \\ block_ref: {} - \\ }} - \\ - , .{ - node.id, - node.slot_complete, - node.parent, - node.child, - node.sibling, - node.merkle_root, - node.chained_merkle_root, - node.data_complete, - node.slot_complete, - node.block_ref, - }); - } -}; - -// TODO: handle eviction -/// A tree of FEC sets, which are also keyed by their merkle (and chained) merkle roots. -const MerkleForest = struct { - // owns all of the memory of nodes used in the map/tree nodes - pool: NodePool, - - // Nodes are inserted, keyed by their merkle root. - // New nodes can look for their parent using this map. - // - // merkle-hash -> node - map: MerkleMap, - - // Nodes are inserted, keyed by their *chained* merkle root. - // New nodes can look for their child using this map. - // - // chained-merkle-hash -> node - orphan_map: OrphanMap, - - const capacity = 4096; - - // keyed by merkle root - const OrphanMap = std.ArrayHashMapUnmanaged(void, *MerkleNode, OrphanContext, true); - - // keyed by chained merkle root - const MerkleMap = std.ArrayHashMapUnmanaged(void, *MerkleNode, MerkleContext, true); - - const NodePool = Pool(MerkleNode, u32); - - const MerkleContext = struct { - map: *const MerkleMap, - - pub fn hash(ctx: MerkleContext, key: *const Hash) u32 { - _ = ctx; - return @bitCast(key.data[0..4].*); - } - pub fn eql(ctx: MerkleContext, a: *const Hash, _: void, key_idx: usize) bool { - const b: *const Hash = &ctx.map.values()[key_idx].merkle_root; - return a.eql(b); - } - }; - - const OrphanContext = struct { - map: *const OrphanMap, - - pub fn hash(ctx: OrphanContext, key: *const Hash) u32 { - _ = ctx; - return @bitCast(key.data[0..4].*); - } - pub fn eql(ctx: OrphanContext, a: *const Hash, _: void, key_idx: usize) bool { - const b: *const Hash = &ctx.map.values()[key_idx].chained_merkle_root; - return a.eql(b); - } - }; - - fn init(allocator: std.mem.Allocator) !MerkleForest { - const pool_buf = try allocator.alloc(MerkleNode, capacity); - errdefer allocator.free(pool_buf); - - var map: MerkleMap = .empty; - errdefer map.deinit(allocator); - try map.ensureTotalCapacity(allocator, capacity); - - var orphan_map: OrphanMap = .empty; - errdefer orphan_map.deinit(allocator); - try orphan_map.ensureTotalCapacity(allocator, capacity); - - return .{ - // NOTE: the pool and the tree share the exact same buffer - this is intentional - .pool = .init(pool_buf[0..capacity]), - .map = map, - .orphan_map = orphan_map, - }; - } - - fn deinit(self: *MerkleForest, allocator: std.mem.Allocator) void { - allocator.free(self.pool.buf[0..self.pool.len]); - self.map.deinit(allocator); - self.orphan_map.deinit(allocator); - } - - fn assertCounts(self: *const MerkleForest) void { - std.debug.assert(self.orphan_map.count() <= self.map.count()); - tracy.plot(u32, "Merkle forest fec sets", @intCast(self.map.count())); - tracy.plot(u32, "Merkle forest fec sets (orphaned)", @intCast(self.orphan_map.count())); - } -}; - -test "MerkleForest tree put" { - var tree: MerkleForest = try .init(std.testing.allocator); - defer tree.deinit(std.testing.allocator); - - const a_hash: Hash = .parse("ByzshhkRgXWnTkHjapkkqaKgEFnsg8ceY3bw4MWBzFE"); - const b_hash: Hash = .parse("BMHr4knWhDp8JhqCYhA2K5DUYQsYUVXdy2zWahzt5jLd"); - const c_hash: Hash = .parse("2GyMeUytf6fcsfNP2QQ6F5e5qwAUoMtKUbnH6QU6bTNm"); - const d_hash: Hash = .parse("4UahX8LzYC7xnubvP9QzRHmPPYovtcNYo7rBXKpp3ADM"); - const e_hash: Hash = .parse("An7mDXKMpRninZw6rvqc4wnQ6ukqd3ARko6QmPitjx8B"); - - const a: shred.DeshreddedFecSet = .{ - .chained_merkle_root = .parse("DWCWjQciWoWDzJKwqUZ1ntKqTyXtLVt4C8aL7biBJZ4z"), // prev slot - .merkle_root = a_hash, - - .id = .{ .slot = 409284941, .fec_set_idx = 0 }, - - .data_complete = true, - .slot_complete = false, - - .payload_len = 0, - .payload_buf = undefined, - }; - - const b: shred.DeshreddedFecSet = .{ - .chained_merkle_root = a_hash, - .merkle_root = b_hash, - - .id = .{ .slot = 409284941, .fec_set_idx = 32 }, - - .data_complete = true, - .slot_complete = false, - - .payload_len = 0, - .payload_buf = undefined, - }; - - const c: shred.DeshreddedFecSet = .{ - .chained_merkle_root = b_hash, - .merkle_root = c_hash, - - .id = .{ .slot = 409284941, .fec_set_idx = 64 }, - - .data_complete = true, - .slot_complete = false, - - .payload_len = 0, - .payload_buf = undefined, - }; - - const d: shred.DeshreddedFecSet = .{ - .chained_merkle_root = c_hash, - .merkle_root = d_hash, - - .id = .{ .slot = 409284941, .fec_set_idx = 96 }, - - .data_complete = true, - .slot_complete = true, - - .payload_len = 0, - .payload_buf = undefined, - }; - - // new slot - const e: shred.DeshreddedFecSet = .{ - .chained_merkle_root = d_hash, - .merkle_root = e_hash, - - .id = .{ .slot = 409284942, .fec_set_idx = 0 }, - - .data_complete = true, - .slot_complete = true, - - .payload_len = 0, - .payload_buf = undefined, - }; - - var pool_buf: [api.BlockPool.size()]u8 align(@alignOf(api.BlockPool)) = undefined; - const pool: *api.BlockPool = @ptrCast(&pool_buf); - pool.init(); - - const logger = tel.Logger("main").noop; - - const a_inserted = (try insertFecSet(logger, &a, &tree, pool)).?; - try std.testing.expect(a_inserted.parent == .null); - try std.testing.expect(a_inserted.child == .null); - try std.testing.expect(a_inserted.block_ref == .null); - // give the ancestor block a BlockRef, so that it may propagate - // NOTE: it is expected that the root-most fec set to be inserted first this way as a special - // case. In a real environment this would be the last fec set in the rooted slot. - a_inserted.block_ref = .init(api.BlockRef.fromInt(8053)); - - const expected_block_ref: api.BlockRef.Optional = - .init(api.BlockRef.fromInt(8053)); - - const d_inserted = (try insertFecSet(logger, &d, &tree, pool)).?; - try std.testing.expect(d_inserted.parent == .null); - try std.testing.expect(d_inserted.child == .null); - try std.testing.expect(d_inserted.block_ref == .null); // no path to a => null - - const b_inserted = (try insertFecSet(logger, &b, &tree, pool)).?; - try std.testing.expect(b_inserted.parent != .null); - try std.testing.expect(b_inserted.child == .null); - try std.testing.expect(b_inserted.block_ref == expected_block_ref); - - const c_inserted = (try insertFecSet(logger, &c, &tree, pool)).?; - try std.testing.expect(c_inserted.parent != .null); - try std.testing.expect(c_inserted.child != .null); - try std.testing.expect(c_inserted.block_ref == expected_block_ref); - try std.testing.expect(d_inserted.block_ref == expected_block_ref); - - const e_inserted = (try insertFecSet(logger, &e, &tree, pool)).?; - try std.testing.expect(e_inserted.parent != .null); - try std.testing.expect(e_inserted.child == .null); - // new slot => new BlockRef - try std.testing.expect(e_inserted.block_ref != .null); - try std.testing.expect(e_inserted.block_ref != expected_block_ref); - - // We cannot insert duplicates - try std.testing.expectEqual(null, try insertFecSet(logger, &a, &tree, pool)); - try std.testing.expectEqual(null, try insertFecSet(logger, &b, &tree, pool)); - try std.testing.expectEqual(null, try insertFecSet(logger, &c, &tree, pool)); - try std.testing.expectEqual(null, try insertFecSet(logger, &d, &tree, pool)); - try std.testing.expectEqual(null, try insertFecSet(logger, &e, &tree, pool)); -} - -test "bootstrap creates root block and chains blockhashes" { - const allocator = std.testing.allocator; - - var activity: lib.runner.Activity = .{}; - var service_view = activity.serviceView(); - const runner: lib.runner.Connection = .{ .activity = &service_view }; - - var metadata: accounts_db.RuntimeMetadata = undefined; - metadata.init(); - metadata.block_id = .parse("ByzshhkRgXWnTkHjapkkqaKgEFnsg8ceY3bw4MWBzFE"); - - // Prefill the blockhash ring with N > 1 hashes as a single writer batch, - // then close the writer end so bootstrap's drain loop terminates. - const test_hashes = [_]Hash{ - .parse("BMHr4knWhDp8JhqCYhA2K5DUYQsYUVXdy2zWahzt5jLd"), - .parse("2GyMeUytf6fcsfNP2QQ6F5e5qwAUoMtKUbnH6QU6bTNm"), - .parse("4UahX8LzYC7xnubvP9QzRHmPPYovtcNYo7rBXKpp3ADM"), - .parse("Hh8DjJdpQRGeZ6bUxYyt1PBktnFtNAwZoQuwZqGWLPfB"), - }; - { - var writer = metadata.blockhash_queue.hashes.getView(.writer); - const buf = writer.getBuffer().?; - try std.testing.expect(buf.len >= test_hashes.len); - @memcpy(buf[0..test_hashes.len], &test_hashes); - writer.advance(test_hashes.len); - writer.close(); - } - - const root_slot: lib.solana.Slot = 100; - metadata.populateSlot(root_slot); - - var pool_buf: [api.BlockPool.size()]u8 align(@alignOf(api.BlockPool)) = undefined; - const pool: *api.BlockPool = @ptrCast(&pool_buf); - pool.init(); - - var forest: MerkleForest = try .init(allocator); - defer forest.deinit(allocator); - - const exec_states = try allocator.create(BlockExecStates); - defer allocator.destroy(exec_states); - @memset(exec_states, null); - - const blockhash_states = try allocator.create(BlockHashStates); - defer allocator.destroy(blockhash_states); - @memset(blockhash_states, null); - - const logger = tel.Logger("main").noop; - - try bootstrap(logger, runner, &metadata, &forest, pool, exec_states, blockhash_states); - - // find root in pool - var root_opt: ?api.BlockRef = null; - for (pool.buf(), 0..) |block, i| { - if (block.item.slot.opt()) |slot| if (slot == root_slot) { - try std.testing.expectEqual(null, root_opt); - root_opt = api.BlockRef.fromInt(@intCast(i)); - }; - } - const root = root_opt orelse return error.NoRoot; - - try std.testing.expectEqual(root_slot, root.ptr(pool).slot.opt().?); - try std.testing.expect(exec_states[root.index()].?.finished()); - - // walk backwards from root, checking each block's hash, slot, and that the - // parent and child link properly. - var current: ?api.BlockRef = root; - var expected_child: ?api.BlockRef = null; - for (0..test_hashes.len) |i| { - const block_ref = current orelse return error.ParentNotSpecified; - const block = block_ref.ptr(pool); - try std.testing.expectEqual( - test_hashes[test_hashes.len - 1 - i], - blockhash_states[block_ref.index()].?, - ); - try std.testing.expectEqual( - if (i == 0) root_slot else null, - block.slot.opt(), - ); - try std.testing.expectEqual(expected_child, block.child.opt()); - expected_child = block_ref; - current = block.parent.opt(); - } - try std.testing.expectEqual(null, current); -}