feat(metadata): state transfer for snapshot and client table

core/binary_protocol/src/consensus/command.rs

@@ -59,6 +59,17 @@ pub enum Command2 {
     RepairPrepare = 19,
     RepairDone = 20,
     RangeEvicted = 21,
+
+    // State transfer (metadata plane): a restarted replica replaces its
+    // snapshot-shaped state (metadata snapshot + client table) from the
+    // current primary, then journal repair covers the tail. Pull-based:
+    // the requester asks for the target descriptor, then fetches each
+    // artifact in bounded chunks (per-peer bus queues drop overruns
+    // silently, so push cannot work).
+    RequestStateTransfer = 22,
+    StateTransferTarget = 23,
+    RequestStateChunk = 24,
+    StateChunk = 25,
 }
 
 // SAFETY: Command2 is #[repr(u8)] with no padding bytes.
@@ -69,7 +80,7 @@ unsafe impl CheckedBitPattern for Command2 {
     type Bits = u8;
 
     fn is_valid_bit_pattern(bits: &u8) -> bool {
-        *bits <= Self::RangeEvicted as u8
+        *bits <= Self::StateChunk as u8
     }
 }
 
@@ -90,16 +101,16 @@ mod tests {
 
     #[test]
     fn replica_auth_commands_are_valid_bit_patterns() {
-        // Locks the is_valid_bit_pattern bump: 14..=21 parse, 22 still rejects.
-        for command in 14u8..=21 {
+        // Locks the is_valid_bit_pattern bump: 14..=25 parse, 26 still rejects.
+        for command in 14u8..=25 {
             let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
             buf.resize(256, 0);
             buf[60] = command;
             assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_ok());
         }
         let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
         buf.resize(256, 0);
-        buf[60] = 22;
+        buf[60] = 26;
         assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
     }
 }

core/binary_protocol/src/consensus/header.rs

@@ -1280,6 +1280,277 @@ impl ConsensusHeader for RepairRangeReplyHeader {
     }
 }
 
+// State transfer: descriptor + chunk pull frames.
+//
+// Plane-agnostic: the descriptor's BODY carries a state manifest (see the
+// consensus crate's `state_manifest`) listing N artifacts, and the chunk
+// frames address bytes by `(manifest index, offset)`. The metadata plane
+// ships two artifacts (snapshot + client table); the partition plane ships
+// its own set (segment logs, offsets) through the same frames.
+
+// RequestStateTransferHeader - restarted replica -> current primary
+
+/// Ask the current primary for a state-transfer target descriptor.
+///
+/// Answered with `StateTransferTarget`. Sent by a restarted replica after it
+/// adopts a view from a live primary; `nonce` correlates the whole transfer
+/// session.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RequestStateTransferHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    pub namespace: u64,
+    pub reserved: [u8; 104],
+}
+const _: () = {
+    assert!(size_of::<RequestStateTransferHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(RequestStateTransferHeader, nonce)
+            == offset_of!(RequestStateTransferHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(
+        offset_of!(RequestStateTransferHeader, reserved) + size_of::<[u8; 104]>() == HEADER_SIZE
+    );
+};
+
+impl ConsensusHeader for RequestStateTransferHeader {
+    const COMMAND: Command2 = Command2::RequestStateTransfer;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::RequestStateTransfer {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RequestStateTransfer,
+                found: self.command,
+            });
+        }
+        Ok(())
+    }
+}
+
+// StateTransferTargetHeader - serving primary -> requester
+
+/// The transfer target descriptor.
+///
+/// The artifact list rides the BODY as an encoded state manifest (`size`
+/// spans header + manifest); the header carries only the session nonce and
+/// the serving peer's applied frontier. `available == 0` means the serving
+/// peer cannot serve right now (not a caught-up primary, or it has never
+/// checkpointed, so the requester's journal repair can cover the whole gap)
+/// and the frame is header-only.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct StateTransferTargetHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    /// Serving primary's applied frontier (`commit_min`) when the descriptor
+    /// was built. The receiver's tail repair targets past this.
+    pub commit_op: u64,
+    pub namespace: u64,
+    pub available: u8,
+    pub reserved: [u8; 95],
+}
+const _: () = {
+    assert!(size_of::<StateTransferTargetHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(StateTransferTargetHeader, nonce)
+            == offset_of!(StateTransferTargetHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 95]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for StateTransferTargetHeader {
+    const COMMAND: Command2 = Command2::StateTransferTarget;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::StateTransferTarget {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::StateTransferTarget,
+                found: self.command,
+            });
+        }
+        if self.available > 1 {
+            return Err(ConsensusError::InvalidField(
+                "available must be 0 or 1".to_string(),
+            ));
+        }
+        // Unavailable is a bare refusal; a manifest body on it would be
+        // ambiguous (which offer would the chunks belong to?).
+        if self.available == 0 && self.size as usize != HEADER_SIZE {
+            return Err(ConsensusError::InvalidField(
+                "unavailable descriptor must be header-only".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+// RequestStateChunkHeader - requester -> serving primary
+
+/// Pull one bounded chunk of an artifact.
+///
+/// Lockstep per artifact: the requester keeps at most one chunk in flight,
+/// so the bounded per-peer bus queue can never drop a burst tail.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RequestStateChunkHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    pub offset: u64,
+    pub namespace: u64,
+    pub len: u32,
+    /// Index into the offered state manifest. Range-checked by the serving
+    /// handler against the cached offer (the header cannot know the count).
+    pub artifact: u32,
+    pub reserved: [u8; 88],
+}
+const _: () = {
+    assert!(size_of::<RequestStateChunkHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(RequestStateChunkHeader, nonce)
+            == offset_of!(RequestStateChunkHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(RequestStateChunkHeader, reserved) + size_of::<[u8; 88]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for RequestStateChunkHeader {
+    const COMMAND: Command2 = Command2::RequestStateChunk;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::RequestStateChunk {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RequestStateChunk,
+                found: self.command,
+            });
+        }
+        if self.len == 0 {
+            return Err(ConsensusError::InvalidField(
+                "chunk len must be non-zero".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+// StateChunkHeader - serving primary -> requester (carries payload)
+
+/// One chunk of artifact bytes at `offset`.
+///
+/// The payload rides the body (`size` spans header + payload). Transit
+/// integrity is checked at the artifact level (descriptor checksums), not
+/// per chunk.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct StateChunkHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    pub offset: u64,
+    pub namespace: u64,
+    /// Index into the offered state manifest. Range-checked by the receiving
+    /// handler against its accepted manifest.
+    pub artifact: u32,
+    pub reserved: [u8; 92],
+}
+const _: () = {
+    assert!(size_of::<StateChunkHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(StateChunkHeader, nonce)
+            == offset_of!(StateChunkHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(StateChunkHeader, reserved) + size_of::<[u8; 92]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for StateChunkHeader {
+    const COMMAND: Command2 = Command2::StateChunk;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::StateChunk {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::StateChunk,
+                found: self.command,
+            });
+        }
+        if (self.size as usize) < HEADER_SIZE {
+            return Err(ConsensusError::InvalidField(
+                "state chunk size below header size".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
 // Tests
 
 #[cfg(test)]

core/binary_protocol/src/consensus/mod.rs

@@ -48,7 +48,8 @@ pub use header::{
     CommitHeader, ConsensusHeader, DoViewChangeHeader, EvictionHeader, EvictionReason,
     GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN,
     RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
-    RequestStartViewHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+    RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader, SIZE_FIELD_OFFSET,
+    StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
     read_size_field,
 };
 pub use operation::Operation;

core/binary_protocol/src/lib.rs

@@ -74,8 +74,9 @@ pub use consensus::{
     Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
     EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader,
     RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
-    RequestPreparesHeader, RequestStartViewHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
-    StartViewHeader, read_size_field, result_code, result_section_len,
+    RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
+    RequestStateTransferHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+    StateChunkHeader, StateTransferTargetHeader, read_size_field, result_code, result_section_len,
 };
 pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
 pub use error::WireError;

core/consensus/Cargo.toml

@@ -41,6 +41,7 @@ rand = { workspace = true }
 rand_xoshiro = { workspace = true }
 server_common = { workspace = true }
 tracing = { workspace = true }
+twox-hash = { workspace = true }
 
 [dev-dependencies]
 futures = { workspace = true }