pub fn SuperBlockType(comptime Storage: type) type {                     std.mem.zeroes(vsr.Header),                 );             } else {-                assert(context.caller == .checkpoint);+                assert(!context.caller.updates_vsr_headers());             } -            if (context.caller != .view_change) {+            if (context.caller.updates_trailers()) {+                superblock.staging.parent_checkpoint_id = superblock.staging.checkpoint_id();

 pub const SuperBlockHeader = extern struct {         pub fn monotonic(old: VSRState, new: VSRState) bool {

+const std = @import("std");+const assert = std.debug.assert;+const maybe = stdx.maybe;++const vsr = @import("../vsr.zig");+const constants = @import("../constants.zig");+const stdx = @import("../stdx.zig");++/// Initial stage: .none+///+/// Transitions:+///+///   .none                  → .cancel_commit | .cancel_grid | .request_target+///   .cancel_commit         → .cancel_grid+///   .cancel_grid           → .request_target+///   .request_target        → .write_sync_start+///   .write_sync_start      → .write_sync_start | .request_trailers+///   .request_trailers      → .write_sync_start | .cancel_grid | .request_manifest_logs+///   .request_manifest_logs → .write_sync_start | .cancel_grid | .write_sync_done+///   .write_sync_done       → .write_sync_start | .done+///   .done                  → .write_sync_start | .none+///+pub const SyncStage = union(enum) {+    /// Not syncing.+    none,++    /// Waiting for a uninterruptible step of the commit chain.+    cancel_commit,++    /// Waiting for `Grid.cancel()`.+    cancel_grid,++    /// We need to sync, but are waiting for a usable `sync_target_max`.+    request_target,++    /// superblock.sync_start()+    write_sync_start: struct { target: SyncTargetCanonical },++    request_trailers: struct {+        target: SyncTargetCanonical,+        manifest: SyncTrailer = .{},+        free_set: SyncTrailer = .{},+        client_sessions: SyncTrailer = .{},++        pub fn done(self: *const @This()) bool {+            return self.manifest.done and self.free_set.done and self.client_sessions.done;+        }+    },++    request_manifest_logs: struct { target: SyncTargetCanonical },++    /// superblock.sync_done()+    write_sync_done: struct { target: SyncTargetCanonical },++    done: struct { target: SyncTargetCanonical },++    pub fn target(stage: *const SyncStage) ?SyncTargetCanonical {+        return switch (stage.*) {+            .none,+            .cancel_commit,+            .cancel_grid,+            .request_target,+            => null,+            .write_sync_start => |s| s.target,+            .request_trailers => |s| s.target,+            .request_manifest_logs => |s| s.target,+            .write_sync_done => |s| s.target,+            .done => |s| s.target,+        };+    }+};++pub const SyncTargetCandidate = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,++    pub fn canonical(target: SyncTargetCandidate) SyncTargetCanonical {+        return .{+            .checkpoint_id = target.checkpoint_id,+            .checkpoint_op = target.checkpoint_op,+            .checkpoint_op_checksum = target.checkpoint_op_checksum,+        };+    }+};++pub const SyncTargetCanonical = struct {

 pub const SuperBlockHeader = extern struct {         [_]u8{0} ** vsr_headers_reserved_size,      pub const VSRState = extern struct {+        pub const Status = enum(u8) {

+//! Verification related to state sync.+//!+//! Check:+//! - the number of simultaneously state-syncing replicas.+//! - that every sync target is a canonical checkpoint.+//! - that every *potential* sync target is a canonical checkpoint.+const std = @import("std");+const assert = std.debug.assert;+const log = std.log.scoped(.sync_checker);++const constants = @import("../../constants.zig");+const vsr = @import("../../vsr.zig");++pub fn SyncCheckerType(comptime Replica: type) type {+    return struct {+        const SyncChecker = @This();++        /// A list of canonical checkpoint ids.+        /// Indexed by checkpoint_index(checkpoint_op).+        canonical: std.ArrayList(u128),++        replicas_syncing: std.StaticBitSet(constants.nodes_max) =+            std.StaticBitSet(constants.nodes_max).initEmpty(),++        pub fn init(allocator: std.mem.Allocator) SyncChecker {+            var canonical = std.ArrayList(u128).init(allocator);+            errdefer canonical.deinit();++            return SyncChecker{+                .canonical = canonical,+            };+        }++        pub fn deinit(checker: *SyncChecker) void {+            checker.canonical.deinit();+        }++        /// Verify that the number of simultaneous syncing (non-standby) replicas does not exceed+        /// the safe limit.+        pub fn replica_sync_start(+            checker: *SyncChecker,+            replica_index: u8,+            replica: *const Replica,+        ) void {+            if (replica.standby()) return;+            checker.replicas_syncing.set(replica_index);++            // This implicitly checks that R=1 and R=2 clusters never state sync.+            // Don't count standbys since they aren't part of the replication quorum.+            const quorums = vsr.quorums(replica.replica_count);+            const replicas_syncing_max = max: {+                if (replica.replica_count == 2) {+                    // See Replica.jump_sync_target() and the test "Cluster: sync: R=2" for details.+                    break :max 1;+                } else {+                    break :max replica.replica_count - quorums.replication;+                }+            };++            assert(checker.replicas_syncing.count() <= replicas_syncing_max);+        }++        pub fn replica_sync_done(+            checker: *SyncChecker,+            replica_index: u8,+            replica: *const Replica,

+const std = @import("std");+const assert = std.debug.assert;+const maybe = stdx.maybe;++const vsr = @import("../vsr.zig");+const constants = @import("../constants.zig");+const stdx = @import("../stdx.zig");++/// Initial stage: .none+///+/// Transitions:+///+///   .none                  → .cancel_commit | .cancel_grid | .request_target+///   .cancel_commit         → .cancel_grid+///   .cancel_grid           → .request_target+///   .request_target        → .write_sync_start+///   .write_sync_start      → .write_sync_start | .request_trailers+///   .request_trailers      → .write_sync_start | .cancel_grid | .request_manifest_logs+///   .request_manifest_logs → .write_sync_start | .cancel_grid | .write_sync_done+///   .write_sync_done       → .write_sync_start | .done+///   .done                  → .write_sync_start | .none+///+pub const SyncStage = union(enum) {+    /// Not syncing.+    none,++    /// Waiting for a uninterruptible step of the commit chain.+    cancel_commit,++    /// Waiting for `Grid.cancel()`.+    cancel_grid,++    /// We need to sync, but are waiting for a usable `sync_target_max`.+    request_target,++    /// superblock.sync_start()+    write_sync_start: struct { target: SyncTargetCanonical },++    request_trailers: struct {+        target: SyncTargetCanonical,+        manifest: SyncTrailer = .{},+        free_set: SyncTrailer = .{},+        client_sessions: SyncTrailer = .{},++        pub fn done(self: *const @This()) bool {+            return self.manifest.done and self.free_set.done and self.client_sessions.done;+        }+    },++    request_manifest_logs: struct { target: SyncTargetCanonical },++    /// superblock.sync_done()+    write_sync_done: struct { target: SyncTargetCanonical },++    done: struct { target: SyncTargetCanonical },++    pub fn target(stage: *const SyncStage) ?SyncTargetCanonical {+        return switch (stage.*) {+            .none,+            .cancel_commit,+            .cancel_grid,+            .request_target,+            => null,+            .write_sync_start => |s| s.target,+            .request_trailers => |s| s.target,+            .request_manifest_logs => |s| s.target,+            .write_sync_done => |s| s.target,+            .done => |s| s.target,+        };+    }+};++pub const SyncTargetCandidate = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,++    pub fn canonical(target: SyncTargetCandidate) SyncTargetCanonical {+        return .{+            .checkpoint_id = target.checkpoint_id,+            .checkpoint_op = target.checkpoint_op,+            .checkpoint_op_checksum = target.checkpoint_op_checksum,+        };+    }+};++pub const SyncTargetCanonical = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,+};++pub const SyncTargetQuorum = struct {+    /// The latest known checkpoint identifier from every *other* replica.+    /// Unlike sync_target_max, these SyncTargets are *not* known to be canonical.+    candidates: [constants.replicas_max]?SyncTargetCandidate =+        [_]?SyncTargetCandidate{null} ** constants.replicas_max,++    pub fn replace(+        quorum: *SyncTargetQuorum,+        replica: u8,+        candidate: *const SyncTargetCandidate,+    ) bool {+        if (quorum.candidates[replica]) |candidate_existing| {+            // Ignore old candidate.+            if (candidate.checkpoint_op < candidate_existing.checkpoint_op) {+                return false;+            }++            maybe(candidate.checkpoint_op == candidate_existing.checkpoint_op);+        }+        quorum.candidates[replica] = candidate.*;+        return true;+    }++    pub fn count(quorum: *const SyncTargetQuorum, candidate: *const SyncTargetCandidate) usize {+        var candidates_matching: usize = 0;+        for (quorum.candidates) |candidate_target| {+            if (candidate_target != null and+                candidate_target.?.checkpoint_op == candidate.checkpoint_op and+                candidate_target.?.checkpoint_id == candidate.checkpoint_id)+            {+                assert(std.meta.eql(candidate_target, candidate.*));+                candidates_matching += 1;+            }+        }+        return candidates_matching;

+const std = @import("std");+const assert = std.debug.assert;+const maybe = stdx.maybe;++const vsr = @import("../vsr.zig");+const constants = @import("../constants.zig");+const stdx = @import("../stdx.zig");++/// Initial stage: .none+///+/// Transitions:+///+///   .none                  → .cancel_commit | .cancel_grid | .request_target+///   .cancel_commit         → .cancel_grid+///   .cancel_grid           → .request_target+///   .request_target        → .write_sync_start+///   .write_sync_start      → .write_sync_start | .request_trailers+///   .request_trailers      → .write_sync_start | .cancel_grid | .request_manifest_logs+///   .request_manifest_logs → .write_sync_start | .cancel_grid | .write_sync_done+///   .write_sync_done       → .write_sync_start | .done+///   .done                  → .write_sync_start | .none+///+pub const SyncStage = union(enum) {+    /// Not syncing.+    none,++    /// Waiting for a uninterruptible step of the commit chain.+    cancel_commit,++    /// Waiting for `Grid.cancel()`.+    cancel_grid,++    /// We need to sync, but are waiting for a usable `sync_target_max`.+    request_target,++    /// superblock.sync_start()+    write_sync_start: struct { target: SyncTargetCanonical },++    request_trailers: struct {+        target: SyncTargetCanonical,+        manifest: SyncTrailer = .{},+        free_set: SyncTrailer = .{},+        client_sessions: SyncTrailer = .{},++        pub fn done(self: *const @This()) bool {+            return self.manifest.done and self.free_set.done and self.client_sessions.done;+        }+    },++    request_manifest_logs: struct { target: SyncTargetCanonical },++    /// superblock.sync_done()+    write_sync_done: struct { target: SyncTargetCanonical },++    done: struct { target: SyncTargetCanonical },++    pub fn target(stage: *const SyncStage) ?SyncTargetCanonical {+        return switch (stage.*) {+            .none,+            .cancel_commit,+            .cancel_grid,+            .request_target,+            => null,+            .write_sync_start => |s| s.target,+            .request_trailers => |s| s.target,+            .request_manifest_logs => |s| s.target,+            .write_sync_done => |s| s.target,+            .done => |s| s.target,+        };+    }+};++pub const SyncTargetCandidate = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,++    pub fn canonical(target: SyncTargetCandidate) SyncTargetCanonical {+        return .{+            .checkpoint_id = target.checkpoint_id,+            .checkpoint_op = target.checkpoint_op,+            .checkpoint_op_checksum = target.checkpoint_op_checksum,+        };+    }+};++pub const SyncTargetCanonical = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,+};++pub const SyncTargetQuorum = struct {+    /// The latest known checkpoint identifier from every *other* replica.+    /// Unlike sync_target_max, these SyncTargets are *not* known to be canonical.+    candidates: [constants.replicas_max]?SyncTargetCandidate =+        [_]?SyncTargetCandidate{null} ** constants.replicas_max,++    pub fn replace(+        quorum: *SyncTargetQuorum,+        replica: u8,+        candidate: *const SyncTargetCandidate,+    ) bool {+        if (quorum.candidates[replica]) |candidate_existing| {+            // Ignore old candidate.+            if (candidate.checkpoint_op < candidate_existing.checkpoint_op) {+                return false;+            }++            maybe(candidate.checkpoint_op == candidate_existing.checkpoint_op);+        }+        quorum.candidates[replica] = candidate.*;+        return true;+    }++    pub fn count(quorum: *const SyncTargetQuorum, candidate: *const SyncTargetCandidate) usize {+        var candidates_matching: usize = 0;+        for (quorum.candidates) |candidate_target| {+            if (candidate_target != null and+                candidate_target.?.checkpoint_op == candidate.checkpoint_op and+                candidate_target.?.checkpoint_id == candidate.checkpoint_id)+            {+                assert(std.meta.eql(candidate_target, candidate.*));+                candidates_matching += 1;+            }+        }+        return candidates_matching;+    }+};++pub const SyncTrailer = struct {

+//! Verification related to state sync.+//!+//! Check:+//! - the number of simultaneously state-syncing replicas.+//! - that every sync target is a canonical checkpoint.+//! - that every *potential* sync target is a canonical checkpoint.+const std = @import("std");+const assert = std.debug.assert;+const log = std.log.scoped(.sync_checker);++const constants = @import("../../constants.zig");+const vsr = @import("../../vsr.zig");++pub fn SyncCheckerType(comptime Replica: type) type {+    return struct {+        const SyncChecker = @This();++        /// A list of canonical checkpoint ids.+        /// Indexed by checkpoint_index(checkpoint_op).+        canonical: std.ArrayList(u128),++        replicas_syncing: std.StaticBitSet(constants.nodes_max) =+            std.StaticBitSet(constants.nodes_max).initEmpty(),++        pub fn init(allocator: std.mem.Allocator) SyncChecker {+            var canonical = std.ArrayList(u128).init(allocator);+            errdefer canonical.deinit();++            return SyncChecker{+                .canonical = canonical,+            };+        }++        pub fn deinit(checker: *SyncChecker) void {+            checker.canonical.deinit();+        }++        /// Verify that the number of simultaneous syncing (non-standby) replicas does not exceed+        /// the safe limit.+        pub fn replica_sync_start(+            checker: *SyncChecker,+            replica_index: u8,+            replica: *const Replica,+        ) void {+            if (replica.standby()) return;+            checker.replicas_syncing.set(replica_index);++            // This implicitly checks that R=1 and R=2 clusters never state sync.+            // Don't count standbys since they aren't part of the replication quorum.+            const quorums = vsr.quorums(replica.replica_count);+            const replicas_syncing_max = max: {+                if (replica.replica_count == 2) {+                    // See Replica.jump_sync_target() and the test "Cluster: sync: R=2" for details.+                    break :max 1;+                } else {+                    break :max replica.replica_count - quorums.replication;+                }+            };++            assert(checker.replicas_syncing.count() <= replicas_syncing_max);+        }++        pub fn replica_sync_done(+            checker: *SyncChecker,+            replica_index: u8,+            replica: *const Replica,+        ) void {+            _ = replica;+            checker.replicas_syncing.unset(replica_index);

+const std = @import("std");+const assert = std.debug.assert;+const maybe = stdx.maybe;++const vsr = @import("../vsr.zig");+const constants = @import("../constants.zig");+const stdx = @import("../stdx.zig");++/// Initial stage: .none+///+/// Transitions:+///+///   .none                  → .cancel_commit | .cancel_grid | .request_target+///   .cancel_commit         → .cancel_grid+///   .cancel_grid           → .request_target+///   .request_target        → .write_sync_start+///   .write_sync_start      → .write_sync_start | .request_trailers+///   .request_trailers      → .write_sync_start | .cancel_grid | .request_manifest_logs+///   .request_manifest_logs → .write_sync_start | .cancel_grid | .write_sync_done+///   .write_sync_done       → .write_sync_start | .done+///   .done                  → .write_sync_start | .none+///+pub const SyncStage = union(enum) {+    /// Not syncing.+    none,++    /// Waiting for a uninterruptible step of the commit chain.+    cancel_commit,++    /// Waiting for `Grid.cancel()`.+    cancel_grid,++    /// We need to sync, but are waiting for a usable `sync_target_max`.+    request_target,++    /// superblock.sync_start()+    write_sync_start: struct { target: SyncTargetCanonical },++    request_trailers: struct {+        target: SyncTargetCanonical,+        manifest: SyncTrailer = .{},+        free_set: SyncTrailer = .{},+        client_sessions: SyncTrailer = .{},++        pub fn done(self: *const @This()) bool {+            return self.manifest.done and self.free_set.done and self.client_sessions.done;+        }+    },++    request_manifest_logs: struct { target: SyncTargetCanonical },++    /// superblock.sync_done()+    write_sync_done: struct { target: SyncTargetCanonical },++    done: struct { target: SyncTargetCanonical },++    pub fn target(stage: *const SyncStage) ?SyncTargetCanonical {+        return switch (stage.*) {+            .none,+            .cancel_commit,+            .cancel_grid,+            .request_target,+            => null,+            .write_sync_start => |s| s.target,+            .request_trailers => |s| s.target,+            .request_manifest_logs => |s| s.target,+            .write_sync_done => |s| s.target,+            .done => |s| s.target,+        };+    }+};++pub const SyncTargetCandidate = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,++    pub fn canonical(target: SyncTargetCandidate) SyncTargetCanonical {+        return .{+            .checkpoint_id = target.checkpoint_id,+            .checkpoint_op = target.checkpoint_op,+            .checkpoint_op_checksum = target.checkpoint_op_checksum,+        };+    }+};++pub const SyncTargetCanonical = struct {+    /// The target's checkpoint identifier.+    checkpoint_id: u128,+    /// The op_checkpoint() that corresponds to the checkpoint id.+    checkpoint_op: u64,+    /// The checksum of the prepare corresponding to checkpoint_op.+    checkpoint_op_checksum: u128,+};++pub const SyncTargetQuorum = struct {+    /// The latest known checkpoint identifier from every *other* replica.+    /// Unlike sync_target_max, these SyncTargets are *not* known to be canonical.+    candidates: [constants.replicas_max]?SyncTargetCandidate =+        [_]?SyncTargetCandidate{null} ** constants.replicas_max,++    pub fn replace(+        quorum: *SyncTargetQuorum,+        replica: u8,+        candidate: *const SyncTargetCandidate,+    ) bool {+        if (quorum.candidates[replica]) |candidate_existing| {+            // Ignore old candidate.+            if (candidate.checkpoint_op < candidate_existing.checkpoint_op) {+                return false;+            }++            maybe(candidate.checkpoint_op == candidate_existing.checkpoint_op);+        }+        quorum.candidates[replica] = candidate.*;+        return true;+    }++    pub fn count(quorum: *const SyncTargetQuorum, candidate: *const SyncTargetCandidate) usize {+        var candidates_matching: usize = 0;+        for (quorum.candidates) |candidate_target| {+            if (candidate_target != null and+                candidate_target.?.checkpoint_op == candidate.checkpoint_op and+                candidate_target.?.checkpoint_id == candidate.checkpoint_id)+            {+                assert(std.meta.eql(candidate_target, candidate.*));+                candidates_matching += 1;+            }+        }+        return candidates_matching;+    }+};++pub const SyncTrailer = struct {+    pub const chunk_size_max = constants.message_body_size_max;++    /// The next offset to fetch.+    offset: u32 = 0,+    done: bool = false,+    final: ?struct { size: u32, checksum: u128 } = null,++    pub fn write(+        trailer: *SyncTrailer,+        data: struct {+            total_buffer: []align(@alignOf(u128)) u8,+            total_size: u32,+            total_checksum: u128,+            chunk: []const u8,+            chunk_offset: u32,+        },+    ) ?[]align(@alignOf(u128)) u8 {+        assert(data.chunk.len <= chunk_size_max);+        assert(data.total_size <= data.total_buffer.len);++        if (trailer.final) |*final| {+            assert(final.checksum == data.total_checksum);+            assert(final.size == data.total_size);+        } else {+            assert(trailer.offset == 0);+            assert(!trailer.done);++            trailer.final = .{+                .checksum = data.total_checksum,+                .size = data.total_size,+            };+        }++        if (trailer.done) return null;++        const total_buffer = data.total_buffer[data.chunk_offset..][0..data.chunk.len];+        if (trailer.offset == data.chunk_offset) {+            stdx.copy_disjoint(.exact, u8, total_buffer, data.chunk);+            trailer.offset += @intCast(u32, data.chunk.len);+            assert(trailer.offset <= data.total_size);++            if (trailer.offset == data.total_size) {+                assert(vsr.checksum(data.total_buffer[0..data.total_size]) == data.total_checksum);++                trailer.done = true;+                return data.total_buffer[0..data.total_size];+            }+        } else {+            if (trailer.offset < data.chunk_offset) {+                // Already received this chunk.