You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/01 11:21:52 UTC
(iggy) branch master updated: feat(consensus): persist metadata VSR state in a durable superblock (#3767)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2afd4b12b feat(consensus): persist metadata VSR state in a durable superblock (#3767)
2afd4b12b is described below
commit 2afd4b12bb34ef4147c62e58f003315cf33074e5
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sat Aug 1 16:51:47 2026 +0530
feat(consensus): persist metadata VSR state in a durable superblock (#3767)
---
Cargo.lock | 5 +
core/binary_protocol/src/consensus/header.rs | 41 +-
core/consensus/Cargo.toml | 1 +
core/consensus/src/client_table.rs | 514 ++++++++++-
core/consensus/src/impls.rs | 445 +++++++++-
core/consensus/src/lib.rs | 7 +-
core/consensus/src/vsr_state.rs | 224 +++++
core/integration/Cargo.toml | 8 +-
.../tests/server/cluster_view_durability_vsr.rs | 266 ++++++
.../server/metadata_checkpoint_recovery_vsr.rs | 138 +++
core/integration/tests/server/mod.rs | 8 +
core/journal/Cargo.toml | 2 +
core/journal/src/lib.rs | 15 +
core/journal/src/prepare_journal.rs | 495 ++++++++++-
core/journal/src/superblock.rs | 824 +++++++++++++++++
core/metadata/src/impls/metadata.rs | 777 +++++++++++++---
core/metadata/src/impls/recovery.rs | 976 +++++++++++++++++++--
core/metadata/src/lib.rs | 4 +-
core/metadata/src/stm/snapshot.rs | 25 +-
core/metadata/src/stm/stream.rs | 86 ++
core/metadata/src/stm/user.rs | 5 +
core/server-ng/src/bootstrap.rs | 224 +++--
core/server-ng/src/dispatch.rs | 7 +-
core/server-ng/src/partition_reconciler.rs | 2 +-
core/shard/src/lib.rs | 94 +-
core/simulator/src/deps.rs | 121 +++
core/simulator/src/lib.rs | 517 ++++++++++-
core/simulator/src/replica.rs | 114 ++-
28 files changed, 5589 insertions(+), 356 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 50939ed39..8c4ddbf28 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3225,6 +3225,7 @@ dependencies = [
"message_bus",
"rand 0.10.2",
"rand_xoshiro",
+ "serde",
"server_common",
"tracing",
]
@@ -7424,6 +7425,7 @@ dependencies = [
"compio",
"configs",
"configs_derive",
+ "consensus",
"ctor 1.0.9",
"deltalake",
"dtor 1.0.5",
@@ -7437,6 +7439,7 @@ dependencies = [
"iggy_common",
"iggy_connector_doris_sink",
"iggy_connector_sdk",
+ "journal",
"jsonwebtoken",
"keyring-core",
"lazy_static",
@@ -7678,6 +7681,8 @@ dependencies = [
"iggy_binary_protocol",
"server_common",
"tempfile",
+ "tracing",
+ "twox-hash",
]
[[package]]
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index b819037a7..d2b5009e9 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1048,15 +1048,30 @@ pub struct StartViewHeader {
/// max(commit) from all DVCs.
pub commit: u64,
pub namespace: u64,
- pub reserved: [u8; 104],
+ pub reserved: [u8; 88],
+ /// Sender's incarnation, echoed from the `RequestStartView` this answers so a
+ /// recovering replica can prove the reply post-dates its restart (see
+ /// `RequestStartViewHeader::incarnation`). `0` on an unsolicited `StartView`
+ /// (a normal view-change completion), which carries no freshness claim.
+ ///
+ /// Carved from the tail of the former `reserved` region and placed LAST so it
+ /// lands 16-aligned with no padding WITHOUT moving `op`/`commit`/`namespace`.
+ /// A peer that predates it sends zeros, decoding as `incarnation == 0`, which
+ /// the `handle_start_view` guard treats as no claim rather than as a foreign
+ /// one, so a mixed-version rolling upgrade is wire-compatible: the pre-upgrade
+ /// peer's `StartView` is judged by the view checks alone, as before the field.
+ pub incarnation: u128,
}
const _: () = {
assert!(size_of::<StartViewHeader>() == HEADER_SIZE);
+ // op/commit/namespace keep their pre-incarnation offsets.
assert!(
offset_of!(StartViewHeader, op)
== offset_of!(StartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
);
- assert!(offset_of!(StartViewHeader, reserved) + size_of::<[u8; 104]>() == HEADER_SIZE);
+ // `incarnation` is last and 16-aligned, so the struct has no padding.
+ assert!(offset_of!(StartViewHeader, incarnation) + size_of::<u128>() == HEADER_SIZE);
+ assert!(offset_of!(StartViewHeader, incarnation) % 16 == 0);
};
impl ConsensusHeader for StartViewHeader {
@@ -1097,8 +1112,11 @@ impl ConsensusHeader for StartViewHeader {
/// Recovering replica -> all replicas: resend me the current `StartView`.
///
/// Header-only; only the current view's primary answers, with a targeted
-/// `StartView` (adoption is fenced by the receiver's view monotonicity and
-/// the sender-is-primary check, so no nonce is needed).
+/// `StartView`. Adoption is fenced by the receiver's view monotonicity and the
+/// sender-is-primary check; `incarnation` additionally proves the reply
+/// post-dates this replica's restart, so a `StartView` from a previous
+/// incarnation still in flight cannot be adopted (see the `handle_start_view`
+/// recovering-status guard).
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct RequestStartViewHeader {
@@ -1113,15 +1131,26 @@ pub struct RequestStartViewHeader {
pub reserved_frame: [u8; 66],
pub namespace: u64,
- pub reserved: [u8; 120],
+ pub reserved: [u8; 104],
+ /// The requester's per-boot incarnation, echoed back in the answering
+ /// `StartView` so a reply from a previous incarnation is detectable.
+ ///
+ /// Carved from the tail of the former `reserved` region and placed LAST so it
+ /// lands 16-aligned with no padding WITHOUT moving `namespace`. A peer that
+ /// predates it sends zeros, decoding as `incarnation == 0`, so a mixed-version
+ /// rolling upgrade is wire-compatible.
+ pub incarnation: u128,
}
const _: () = {
assert!(size_of::<RequestStartViewHeader>() == HEADER_SIZE);
+ // namespace keeps its pre-incarnation offset.
assert!(
offset_of!(RequestStartViewHeader, namespace)
== offset_of!(RequestStartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
);
- assert!(offset_of!(RequestStartViewHeader, reserved) + size_of::<[u8; 120]>() == HEADER_SIZE);
+ // `incarnation` is last and 16-aligned, so the struct has no padding.
+ assert!(offset_of!(RequestStartViewHeader, incarnation) + size_of::<u128>() == HEADER_SIZE);
+ assert!(offset_of!(RequestStartViewHeader, incarnation) % 16 == 0);
};
impl ConsensusHeader for RequestStartViewHeader {
diff --git a/core/consensus/Cargo.toml b/core/consensus/Cargo.toml
index a612dca77..7de02ec5b 100644
--- a/core/consensus/Cargo.toml
+++ b/core/consensus/Cargo.toml
@@ -39,6 +39,7 @@ iggy_common = { workspace = true }
message_bus = { workspace = true }
rand = { workspace = true }
rand_xoshiro = { workspace = true }
+serde = { workspace = true }
server_common = { workspace = true }
tracing = { workspace = true }
diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs
index 1011a5ae8..57f4f76e8 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -16,8 +16,11 @@
// under the License.
use iggy_binary_protocol::ReplyHeader;
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
+use iggy_binary_protocol::consensus::ConsensusError;
+use serde::{Deserialize, Serialize};
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen, iobuf::Owned};
use std::collections::{HashMap, VecDeque};
+use std::fmt;
use std::mem::size_of;
use tracing::trace;
@@ -65,12 +68,25 @@ impl CachedReply {
bytes: msg.into_generic().into_frozen(),
}
}
+
+ /// Raw reply bytes for checkpoint serialization, round-tripped through
+ /// [`Self::from_message`] on decode.
+ fn as_bytes(&self) -> &[u8] {
+ self.bytes.as_slice()
+ }
}
/// Reserved request number for [`Operation::Register`](iggy_binary_protocol::Operation::Register).
/// Real requests start at 1 (header validation enforces `request > 0`).
pub const REGISTER_REQUEST_ID: u64 = 0;
+/// Exclusive ceiling on a checkpointed slot index.
+///
+/// Bounds the table [`ClientTable::from_snapshot`] allocates from an index it read off
+/// disk. Mirrors the config's `MAX_METADATA_CLIENTS_TABLE_MAX`, the largest capacity an
+/// operator can configure, so no valid checkpoint can carry an index at or above it.
+pub const CLIENTS_TABLE_SLOT_MAX: usize = 1 << 16;
+
/// Committed replies retained per entry, newest at the back.
///
/// The back is the latest committed reply and is structurally safe:
@@ -124,6 +140,175 @@ struct ClientEntry {
latest_commit: u64,
}
+/// Serializable form of one occupied slot.
+///
+/// Folded into the metadata checkpoint (`MetadataSnapshot`) so fence epochs and
+/// dedup watermarks survive a restart that drained the WAL prefix they committed
+/// in. Carries `client_id` explicitly because the index is rebuilt from it on
+/// decode.
+///
+/// Only the entry's latest reply is carried, not the whole ring: `latest_commit`
+/// is re-derived from its header, which is what keeps `evict_oldest` picking the
+/// same victim on a checkpoint-restored replica as on a WAL-replayed one. The
+/// older ring entries are volatile by design (see [`REPLY_RING_CAPACITY`]), so a
+/// retransmit that would have hit them answers
+/// [`RequestStatus::AlreadyApplied`] instead of replaying bytes: a worse answer,
+/// never a re-execution.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ClientEntrySnapshot {
+ pub client_id: u128,
+ pub epoch: u64,
+ pub user_id: u32,
+ pub watermark: u64,
+ pub watermark_checksum: u128,
+ /// Wire bytes of the entry's latest committed reply, round-tripped through
+ /// [`CachedReply::from_message`]. Never empty: registration seeds the ring.
+ ///
+ /// Serialized as a msgpack `bin` blob, not the integer array a plain `Vec<u8>`
+ /// produces, which spends 2 bytes on every byte >= 0x80 and runs a checkpoint's
+ /// reply payload up to roughly double on disk.
+ #[serde(with = "reply_bytes")]
+ pub reply: Vec<u8>,
+}
+
+/// Serializable [`ClientTable`]: the occupied slots, each with its index.
+///
+/// Slot positions are carried explicitly rather than by array position, so the
+/// encoded form is proportional to live clients instead of to configured capacity.
+/// The alternative (a full `Vec<Option<_>>`) makes the slot count self-perpetuating:
+/// [`ClientTable::from_snapshot`] would have to honour the array's length, so
+/// lowering `clients_table_max` could never take effect, and every checkpoint would
+/// serde-walk `clients_table_max` entries while shard 0 is blocked on fsyncs.
+///
+/// Deterministic eviction order is unaffected: the index is what places each entry,
+/// and it survives here verbatim.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ClientTableSnapshot {
+ pub slots: Vec<(u32, ClientEntrySnapshot)>,
+}
+
+/// Serializes reply bytes as a msgpack `bin` blob. See [`ClientEntrySnapshot::reply`].
+mod reply_bytes {
+ use serde::de::{Error, SeqAccess, Visitor};
+ use serde::{Deserializer, Serializer};
+ use std::fmt;
+
+ pub fn serialize<S: Serializer>(bytes: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
+ serializer.serialize_bytes(bytes)
+ }
+
+ pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
+ deserializer.deserialize_byte_buf(BytesVisitor)
+ }
+
+ struct BytesVisitor;
+
+ impl<'de> Visitor<'de> for BytesVisitor {
+ type Value = Vec<u8>;
+
+ fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.write_str("reply bytes")
+ }
+
+ fn visit_bytes<E: Error>(self, bytes: &[u8]) -> Result<Self::Value, E> {
+ Ok(bytes.to_vec())
+ }
+
+ fn visit_byte_buf<E: Error>(self, bytes: Vec<u8>) -> Result<Self::Value, E> {
+ Ok(bytes)
+ }
+
+ /// A checkpoint written before the `bin` encoding holds an integer array.
+ /// Accepting it keeps this a read-compatible change rather than one that
+ /// refuses a checkpoint it could decode.
+ fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
+ let mut bytes = Vec::with_capacity(seq.size_hint().unwrap_or_default());
+ while let Some(byte) = seq.next_element()? {
+ bytes.push(byte);
+ }
+ Ok(bytes)
+ }
+ }
+}
+
+/// A [`ClientTableSnapshot`] could not be decoded into a [`ClientTable`], so a
+/// corrupt or torn checkpoint refuses boot with a typed error rather than
+/// panicking mid-decode.
+#[derive(Debug)]
+pub enum ClientTableDecodeError {
+ /// A slot's serialized reply bytes are not a valid reply message.
+ InvalidReply {
+ /// Slot whose reply bytes failed to decode.
+ slot: usize,
+ /// The underlying wire-decode failure.
+ source: ConsensusError,
+ },
+ /// Two occupied slots carry the same `client_id`. Rebuilding the index would
+ /// collapse them onto one slot and leave the other occupied but unindexed, so
+ /// the decode is rejected.
+ DuplicateClientId {
+ /// Slot repeating an already-seen `client_id`.
+ slot: usize,
+ /// Slot that first declared it.
+ first_slot: usize,
+ /// The duplicated client id.
+ client_id: u128,
+ },
+ /// Two entries claim the same slot index. The second would overwrite the first,
+ /// leaving that client indexed onto another's state, so the decode is rejected.
+ DuplicateSlot {
+ /// The repeated slot index.
+ slot: usize,
+ },
+ /// A slot index is past what any configured capacity can produce, so honouring
+ /// it would size the table from a corrupt length.
+ SlotOutOfRange {
+ /// The out-of-range slot index.
+ slot: usize,
+ /// Exclusive ceiling on a slot index.
+ max: usize,
+ },
+}
+
+impl fmt::Display for ClientTableDecodeError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::InvalidReply { slot, source } => write!(
+ f,
+ "client-table checkpoint slot {slot} holds invalid reply bytes: {source}"
+ ),
+ Self::DuplicateClientId {
+ slot,
+ first_slot,
+ client_id,
+ } => write!(
+ f,
+ "client-table checkpoint slot {slot} repeats client_id {client_id} already in \
+ slot {first_slot}"
+ ),
+ Self::DuplicateSlot { slot } => write!(
+ f,
+ "client-table checkpoint holds two entries for slot {slot}"
+ ),
+ Self::SlotOutOfRange { slot, max } => write!(
+ f,
+ "client-table checkpoint slot index {slot} is past the {max}-slot ceiling"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for ClientTableDecodeError {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ Self::InvalidReply { source, .. } => Some(source),
+ Self::DuplicateClientId { .. }
+ | Self::DuplicateSlot { .. }
+ | Self::SlotOutOfRange { .. } => None,
+ }
+ }
+}
+
/// Result of checking a request against the client table.
///
/// In-progress dedup is the caller's job, preflights consult
@@ -204,10 +389,17 @@ pub enum CommitReply {
/// `commit_register` in the apply path, so every replica of the group
/// derives an identical table from the committed log.
///
+/// ## Durability
+///
+/// [`Self::to_snapshot`] / [`Self::from_snapshot`] fold the table into the
+/// metadata checkpoint, so sessions registered below the snapshot floor survive
+/// a restart that drained the WAL prefix they committed in.
+///
/// ## Known gaps
///
/// - **Serialization**: encode/decode for rejoin slice-fetch and state
-/// transfer TODO (IGGY-137).
+/// transfer TODO (IGGY-137). The checkpoint form above is local-recovery
+/// only; it is not a wire format.
#[derive(Debug)]
pub struct ClientTable {
/// `None` = free slot. Deterministic iteration for eviction + serialization.
@@ -251,6 +443,123 @@ impl ClientTable {
*self = Self::new(max_clients);
}
+ /// Snapshot the table for the metadata checkpoint: every occupied slot with its
+ /// index, so positions (and with them deterministic eviction order) survive, plus
+ /// each entry's `client_id` so the index rebuilds on decode.
+ ///
+ /// Walks only occupied slots. This runs on shard 0's checkpoint task, inside the
+ /// section where that core is already blocked on the snapshot's fsyncs, so it must
+ /// not also allocate and serde-walk one element per configured slot.
+ #[must_use]
+ pub fn to_snapshot(&self) -> ClientTableSnapshot {
+ let slots = self
+ .slots
+ .iter()
+ .enumerate()
+ .filter_map(|(slot_idx, slot)| {
+ let entry = slot.as_ref()?;
+ let slot_idx = u32::try_from(slot_idx).ok()?;
+ Some((
+ slot_idx,
+ ClientEntrySnapshot {
+ client_id: entry.client_id,
+ epoch: entry.epoch,
+ user_id: entry.user_id,
+ watermark: entry.watermark,
+ watermark_checksum: entry.watermark_checksum,
+ reply: entry.latest().as_bytes().to_vec(),
+ },
+ ))
+ })
+ .collect();
+ ClientTableSnapshot { slots }
+ }
+
+ /// Rebuild a table from a checkpoint snapshot: restore each slot in place and
+ /// rebuild the client-to-slot index.
+ ///
+ /// The restored ring holds only the entry's latest reply, so `latest_commit`
+ /// (and with it `evict_oldest`'s victim order) is reproduced exactly while
+ /// older retransmits degrade from [`RequestStatus::Duplicate`] to
+ /// [`RequestStatus::AlreadyApplied`].
+ ///
+ /// `min_slots` is the configured capacity. The rebuilt table is that, or enough
+ /// to hold the highest occupied index when the checkpoint was taken under a
+ /// larger capacity: a capacity *lowered* below a live entry's slot cannot be
+ /// honoured without dropping a recovered session, so the larger count stands
+ /// until those entries drain, and a later checkpoint then rebuilds at the
+ /// configured size. Slot positions are preserved either way, so eviction order is
+ /// unchanged.
+ ///
+ /// # Errors
+ /// [`ClientTableDecodeError`] if a slot's reply bytes are not a valid reply
+ /// message, if two slots share a `client_id`, if two entries claim the same slot,
+ /// or if a slot index is past [`CLIENTS_TABLE_SLOT_MAX`] (a corrupt, torn, or
+ /// foreign checkpoint). Surfaced rather than panicked so a bad checkpoint refuses
+ /// boot instead of unwinding the shard. Callers verify the checkpoint's checksum
+ /// against the superblock first, so a correct boot never hits this.
+ pub fn from_snapshot(
+ snapshot: ClientTableSnapshot,
+ min_slots: usize,
+ ) -> Result<Self, ClientTableDecodeError> {
+ // Bound the capacity on a slot index read off disk before allocating from it,
+ // as the superblock and WAL do with their length fields.
+ let mut capacity = min_slots;
+ for (slot_idx, _) in &snapshot.slots {
+ let slot = *slot_idx as usize;
+ if slot >= CLIENTS_TABLE_SLOT_MAX {
+ return Err(ClientTableDecodeError::SlotOutOfRange {
+ slot,
+ max: CLIENTS_TABLE_SLOT_MAX,
+ });
+ }
+ capacity = capacity.max(slot + 1);
+ }
+
+ let mut index = HashMap::with_capacity(snapshot.slots.len());
+ let mut slots = Vec::with_capacity(capacity);
+ slots.resize_with(capacity, || None);
+ for (slot_idx, entry) in snapshot.slots {
+ let slot_idx = slot_idx as usize;
+ let reply = Message::<ReplyHeader>::try_from(Owned::<MESSAGE_ALIGN>::copy_from_slice(
+ &entry.reply,
+ ))
+ .map_err(|source| ClientTableDecodeError::InvalidReply {
+ slot: slot_idx,
+ source,
+ })?;
+ // Reject rather than collapse the index onto one slot, leaving the
+ // other occupied but unindexed. Slot `client_id`s are unique in a
+ // table this crate produced, so a duplicate means a corrupt or
+ // foreign checkpoint.
+ if let Some(first_slot) = index.insert(entry.client_id, slot_idx) {
+ return Err(ClientTableDecodeError::DuplicateClientId {
+ slot: slot_idx,
+ first_slot,
+ client_id: entry.client_id,
+ });
+ }
+ let latest_commit = reply.header().commit;
+ let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
+ ring.push_back(CachedReply::from_message(reply));
+ // Two entries claiming one slot would silently drop the first, leaving it
+ // indexed but pointing at another client's state.
+ if slots[slot_idx].is_some() {
+ return Err(ClientTableDecodeError::DuplicateSlot { slot: slot_idx });
+ }
+ slots[slot_idx] = Some(ClientEntry {
+ epoch: entry.epoch,
+ user_id: entry.user_id,
+ watermark: entry.watermark,
+ watermark_checksum: entry.watermark_checksum,
+ ring,
+ client_id: entry.client_id,
+ latest_commit,
+ });
+ }
+ Ok(Self { slots, index })
+ }
+
/// Check a request against the table. Epoch fence first, then the
/// watermark. Register does not come through here: every bind proposes
/// unconditionally so its fence actually moves, see
@@ -667,6 +976,7 @@ impl ClientEntry {
}
#[cfg(test)]
+#[allow(clippy::cast_possible_truncation)]
mod tests {
use super::*;
use iggy_binary_protocol::{Command2, Operation};
@@ -688,6 +998,7 @@ mod tests {
commit,
command: Command2::Reply,
operation: Operation::Register,
+ size: header_size as u32,
..ReplyHeader::default()
};
msg
@@ -716,11 +1027,210 @@ mod tests {
request_checksum,
command: Command2::Reply,
operation: Operation::SendMessages,
+ size: header_size as u32,
..ReplyHeader::default()
};
msg
}
+ #[test]
+ fn to_from_snapshot_round_trips_epochs_and_watermarks() {
+ let mut table = ClientTable::new(8);
+ table.commit_register(1, 11, make_register_reply(1, 10));
+ table.commit_register(2, 22, make_register_reply(2, 20));
+ // Client 1 committed request 5; its reply is the entry's latest.
+ table.commit_reply(1, make_reply_with_checksum(1, 5, 30, 0xbeef));
+
+ let restored = ClientTable::from_snapshot(table.to_snapshot(), 0).unwrap();
+
+ // Fences and dedup history survive, and the index is rebuilt (every
+ // accessor reads through it).
+ assert_eq!(restored.get_epoch(1), Some(10));
+ assert_eq!(restored.get_epoch(2), Some(20));
+ assert_eq!(restored.get_watermark(1), Some(5));
+ assert_eq!(restored.get_watermark(2), Some(0));
+ assert_eq!(restored.get_user_id(1), Some(11));
+ // Replaying request 5 is a dedup hit, not a re-execution: at-most-once
+ // holds across a restart, and the original bytes still answer it.
+ match restored.check_request(1, 10, 5, 0xbeef) {
+ RequestStatus::Duplicate(cached) => assert_eq!(cached.header().request, 5),
+ other => panic!("expected Duplicate, got {other:?}"),
+ }
+ // The persisted watermark checksum still catches request-id reuse.
+ assert!(matches!(
+ restored.check_request(1, 10, 5, 0xfeed),
+ RequestStatus::ChecksumMismatch { request: 5 }
+ ));
+ // A zombie holding the pre-restart epoch of a since-rebound client is
+ // still fenced, so the fence is not weakened by the round trip.
+ assert!(matches!(
+ restored.check_request(1, 9, 6, 0),
+ RequestStatus::Fenced {
+ current: 10,
+ received: 9
+ }
+ ));
+ // A client that never registered is still unknown.
+ assert!(matches!(
+ restored.check_request(3, 1, 1, 0),
+ RequestStatus::NoSession
+ ));
+ }
+
+ // Only the entry's latest reply is persisted, so a retransmit of an older
+ // ring entry is refused execution rather than answered from cache.
+ #[test]
+ fn snapshot_drops_stale_ring_replies_but_keeps_at_most_once() {
+ let mut table = ClientTable::new(4);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ table.commit_reply(1, make_reply_for(1, 5, 20));
+ table.commit_reply(1, make_reply_for(1, 6, 21));
+
+ let restored = ClientTable::from_snapshot(table.to_snapshot(), 0).unwrap();
+
+ assert!(matches!(
+ restored.check_request(1, 10, 5, 0),
+ RequestStatus::AlreadyApplied {
+ request: 5,
+ watermark: 6
+ }
+ ));
+ }
+
+ // Slot positions are preserved, so a checkpoint-restored replica picks the
+ // same eviction victim as one that replayed the whole WAL.
+ #[test]
+ fn snapshot_preserves_slot_order_for_eviction() {
+ let mut table = ClientTable::new(2);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ table.commit_register(2, TEST_USER_ID, make_register_reply(2, 20));
+
+ let mut restored = ClientTable::from_snapshot(table.to_snapshot(), 0).unwrap();
+ assert_eq!(restored.client_ids().collect::<Vec<_>>(), vec![1, 2]);
+
+ // Full table: the oldest latest_commit (client 1, op 10) is evicted, and
+ // latest_commit came back from the persisted reply header.
+ restored.commit_register(3, TEST_USER_ID, make_register_reply(3, 30));
+ assert_eq!(restored.get_epoch(1), None);
+ assert_eq!(restored.get_epoch(2), Some(20));
+ assert_eq!(restored.get_epoch(3), Some(30));
+ }
+
+ // A LOWERED `clients_table_max` must take effect too. It cannot while the encoded
+ // form is one element per configured slot, since the rebuilt table has to be at
+ // least as long as that array.
+ #[test]
+ fn from_snapshot_shrinks_to_the_configured_capacity() {
+ let mut table = ClientTable::new(8);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+
+ let mut restored = ClientTable::from_snapshot(table.to_snapshot(), 2).unwrap();
+
+ // Capacity 2: the recovered session plus one, and the third register evicts.
+ restored.commit_register(2, TEST_USER_ID, make_register_reply(2, 20));
+ restored.commit_register(3, TEST_USER_ID, make_register_reply(3, 30));
+ assert_eq!(restored.count(), 2);
+ assert_eq!(
+ restored.get_epoch(1),
+ None,
+ "the oldest committed entry is the eviction victim at the lowered capacity"
+ );
+ }
+
+ // A capacity lowered below a live entry's slot cannot be honoured without dropping
+ // a recovered session, so the table keeps room for it.
+ #[test]
+ fn from_snapshot_keeps_a_slot_above_the_configured_capacity() {
+ let mut table = ClientTable::new(8);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ let mut snapshot = table.to_snapshot();
+ snapshot.slots[0].0 = 5;
+
+ let restored = ClientTable::from_snapshot(snapshot, 2).unwrap();
+ assert_eq!(
+ restored.get_epoch(1),
+ Some(10),
+ "an entry above the configured capacity must survive, not be dropped"
+ );
+ }
+
+ #[test]
+ fn from_snapshot_rejects_two_entries_in_one_slot() {
+ let mut table = ClientTable::new(2);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ table.commit_register(2, TEST_USER_ID, make_register_reply(2, 20));
+ let mut snapshot = table.to_snapshot();
+ snapshot.slots[1].0 = snapshot.slots[0].0;
+
+ assert!(matches!(
+ ClientTable::from_snapshot(snapshot, 0),
+ Err(ClientTableDecodeError::DuplicateSlot { slot: 0 })
+ ));
+ }
+
+ #[test]
+ fn from_snapshot_rejects_a_slot_index_past_the_ceiling() {
+ // The capacity is allocated from this index, so an out-of-range one must be
+ // refused before the allocation rather than sized from.
+ let mut table = ClientTable::new(2);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ let mut snapshot = table.to_snapshot();
+ snapshot.slots[0].0 = u32::MAX;
+
+ assert!(matches!(
+ ClientTable::from_snapshot(snapshot, 0),
+ Err(ClientTableDecodeError::SlotOutOfRange { .. })
+ ));
+ }
+
+ // A raised `clients_table_max` must take effect on the next boot rather than
+ // staying inert behind the checkpoint's slot count.
+ #[test]
+ fn from_snapshot_grows_to_the_configured_capacity() {
+ let mut table = ClientTable::new(1);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+
+ let mut restored = ClientTable::from_snapshot(table.to_snapshot(), 3).unwrap();
+
+ // Two free slots were padded on, so the next registers land without
+ // evicting the recovered session.
+ restored.commit_register(2, TEST_USER_ID, make_register_reply(2, 20));
+ restored.commit_register(3, TEST_USER_ID, make_register_reply(3, 30));
+ assert_eq!(restored.count(), 3);
+ assert_eq!(restored.get_epoch(1), Some(10));
+ }
+
+ #[test]
+ fn from_snapshot_rejects_duplicate_client_ids() {
+ let mut table = ClientTable::new(2);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ table.commit_register(2, TEST_USER_ID, make_register_reply(2, 20));
+ let mut snapshot = table.to_snapshot();
+ snapshot.slots[1].1 = snapshot.slots[0].1.clone();
+
+ assert!(matches!(
+ ClientTable::from_snapshot(snapshot, 0),
+ Err(ClientTableDecodeError::DuplicateClientId {
+ slot: 1,
+ first_slot: 0,
+ client_id: 1
+ })
+ ));
+ }
+
+ #[test]
+ fn from_snapshot_rejects_invalid_reply_bytes() {
+ let mut table = ClientTable::new(2);
+ table.commit_register(1, TEST_USER_ID, make_register_reply(1, 10));
+ let mut snapshot = table.to_snapshot();
+ snapshot.slots[0].1.reply = vec![0xff; 8];
+
+ assert!(matches!(
+ ClientTable::from_snapshot(snapshot, 0),
+ Err(ClientTableDecodeError::InvalidReply { slot: 0, .. })
+ ));
+ }
+
/// Register client 1 (register commit stamped at op 10). Returns
/// (table, epoch=1).
fn table_with_client() -> (ClientTable, u64) {
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index f969cb022..4b9467e17 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -20,8 +20,9 @@ use crate::vsr_timeout::{TimeoutKind, TimeoutManager};
use crate::{
AckLogEvent, Consensus, ControlActionLogEvent, DvcQuorumArray, IgnoreReason, Pipeline,
PlaneKind, PrepareLogEvent, Project, ReplicaLogContext, SimEventKind, StoredDvc,
- ViewChangeLogEvent, ViewChangeReason, dvc_count, dvc_max_commit, dvc_quorum_array_empty,
- dvc_record, dvc_reset, dvc_select_winner, emit_replica_event, emit_sim_event,
+ ViewChangeLogEvent, ViewChangeReason, VsrState, dvc_count, dvc_max_commit,
+ dvc_quorum_array_empty, dvc_record, dvc_reset, dvc_select_winner, emit_replica_event,
+ emit_sim_event,
};
use bit_set::BitSet;
use clock::{Clock, IggySystemClock};
@@ -30,6 +31,7 @@ use iggy_binary_protocol::{
ReplyHeader, RequestHeader, RequestStartViewHeader, StartViewChangeHeader, StartViewHeader,
};
use iggy_common::IggyTimestamp;
+use iggy_common::calculate_checksum;
use message_bus::IggyMessageBus;
use message_bus::MessageBus;
use server_common::Message;
@@ -695,11 +697,22 @@ pub enum VsrAction {
/// Stamped with the prober's view so peers can fence stale duplicates
/// out of the probed-primary election path.
SendRequestStartView { view: u32, namespace: u64 },
- /// Send `StartView` to all backups (as new primary).
+ /// Send `StartView`, as the view's primary.
+ ///
+ /// `incarnation` echoes the requester's nonce when this answers a
+ /// `RequestStartView` probe (freshness proof), and is `0` on an unsolicited
+ /// send at view-change completion (carries no freshness claim).
+ ///
+ /// `target` is the probing replica for an echo and `None` for an unsolicited
+ /// broadcast. An echo must not be broadcast: the nonce it carries is one
+ /// replica's freshness proof, and every other peer that is itself recovering
+ /// reads a foreign nonce and rejects an otherwise current `StartView`.
SendStartView {
view: u32,
op: u64,
commit: u64,
+ incarnation: u128,
+ target: Option<u8>,
namespace: u64,
},
/// Send `PrepareOK` for each op in `[from_op, to_op]` that is present in the WAL.
@@ -788,6 +801,21 @@ where
// * `replica.log_view ≥ replica.log_view_durable`
// * `replica.log_view = 0` when replica_count=1.
log_view: Cell<u32>,
+ /// The `view` last made durable in the superblock. `view_durable <= view`; a
+ /// view-scoped message must not leave until they are equal, or a crash could
+ /// recover an older view than one this replica already acted in (split brain).
+ /// Advanced by [`Self::mark_superblock_durable`].
+ view_durable: Cell<u32>,
+ /// The `log_view` last made durable in the superblock. Realizes the
+ /// long-standing invariant `log_view >= log_view_durable`.
+ log_view_durable: Cell<u32>,
+ /// Per-boot incarnation nonce, stamped on outbound `RequestStartView` probes
+ /// and echoed in the answering `StartView`, so a recovering replica ignores a
+ /// `StartView` from a previous incarnation still in flight (see the
+ /// [`Self::handle_start_view`] recovering-status guard). `0` means unset
+ /// (partition plane, tests), leaving the guard inert. Set by
+ /// [`Self::set_incarnation`] at boot, before `init`.
+ incarnation: Cell<u128>,
/// Commit point the recovered WAL suffix must re-reach before admitting
/// client requests as primary (`0` = no recovered suffix pending).
recovery_barrier: Cell<u64>,
@@ -925,6 +953,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
namespace,
view: Cell::new(0),
log_view: Cell::new(0),
+ view_durable: Cell::new(0),
+ log_view_durable: Cell::new(0),
+ incarnation: Cell::new(0),
recovery_barrier: Cell::new(0),
recovery_deadline: Cell::new(Duration::ZERO),
ceded_primaryship: Cell::new(false),
@@ -1031,9 +1062,12 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
/// primary for different planes and clients route to the wrong one. Join
/// as a backup instead; either the peers' heartbeat timeout elects a
/// primary and its `StartView` brings this replica forward, or this
- /// replica's own silence provokes that election. Unlike
- /// [`Self::init_recovering`] the local journal is intact, so the normal
- /// commit walk applies it -- no commit-floor fast-forward.
+ /// replica's own silence provokes that election.
+ ///
+ /// The local journal is intact here, so the normal commit walk applies it and no
+ /// commit-floor fast-forward is needed. Callers pair this with
+ /// [`Self::begin_view_probe`], which moves the replica to `Status::Recovering` so
+ /// it stays quorum-invisible until a `StartView` answers.
pub fn init_as_backup(&self) {
self.status.set(Status::Normal);
self.ceded_primaryship.set(true);
@@ -1048,6 +1082,20 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
self.ceded_primaryship.get()
}
+ /// Set the per-boot incarnation nonce, at boot before `init`. Production
+ /// supplies a random `u128`; the deterministic simulator supplies a
+ /// seed-derived value bumped per restart. See the `incarnation` field.
+ pub fn set_incarnation(&self, incarnation: u128) {
+ self.incarnation.set(incarnation);
+ }
+
+ /// This replica's per-boot incarnation nonce (`0` when unset). Stamped on
+ /// outbound `RequestStartView` probes by the shard.
+ #[must_use]
+ pub const fn incarnation(&self) -> u128 {
+ self.incarnation.get()
+ }
+
#[must_use]
// cast_lossless: `u32::from()` unavailable in const fn.
// cast_possible_truncation: modulo by replica_count (u8) guarantees result fits in u8.
@@ -1407,6 +1455,58 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
self.log_view.set(log_view);
}
+ /// Snapshot the durable VSR state for the superblock. `view`, `log_view`,
+ /// `commit_max`, and the replica identity come from consensus; the caller
+ /// supplies the paired checkpoint, which consensus does not own.
+ #[must_use]
+ pub const fn vsr_state(&self, checkpoint_op: u64, checkpoint_checksum: u128) -> VsrState {
+ VsrState {
+ cluster: self.cluster,
+ replica_id: self.replica,
+ replica_count: self.replica_count,
+ view: self.view.get(),
+ log_view: self.log_view.get(),
+ commit_max: self.commit_max.get(),
+ checkpoint_op,
+ checkpoint_checksum,
+ }
+ }
+
+ /// True when the current `(view, log_view)` is not yet in the superblock, so a
+ /// view-scoped send would advertise a view a crash could lose. The split-brain
+ /// gate: the dispatcher persists first when this holds. `commit_max` is
+ /// excluded deliberately, since it advances on every commit and rides the
+ /// checkpoint write rather than each view change.
+ #[must_use]
+ pub const fn needs_superblock_persist(&self) -> bool {
+ self.view.get() != self.view_durable.get()
+ || self.log_view.get() != self.log_view_durable.get()
+ }
+
+ /// Record that the superblock now durably holds `(view, log_view)`: the exact
+ /// values written, NOT a re-read of the current in-memory view.
+ ///
+ /// The caller passes what it wrote because the in-memory view can advance
+ /// across the write's `.await`, when a concurrent checkpoint holds the
+ /// superblock lock while the pump adopts a newer view via `handle_start_view`.
+ /// Re-reading `self.view` here would mark that newer, unwritten view durable,
+ /// and [`Self::needs_superblock_persist`] would wrongly report it safe to
+ /// send: the split-brain footgun this signature removes.
+ ///
+ /// Called after a successful write with the state written, and on boot with
+ /// the recovered `(view, log_view)`, durable by definition.
+ pub fn mark_superblock_durable(&self, view: u32, log_view: u32) {
+ debug_assert!(
+ view <= self.view.get() && log_view <= self.log_view.get(),
+ "durable (view={view}, log_view={log_view}) cannot exceed in-memory \
+ (view={}, log_view={})",
+ self.view.get(),
+ self.log_view.get(),
+ );
+ self.view_durable.set(view);
+ self.log_view_durable.set(log_view);
+ }
+
#[must_use]
pub const fn is_primary_for_view(&self, view: u32) -> bool {
self.primary_index(view) == self.replica
@@ -2216,10 +2316,17 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
if self.log_view.get() != self.view.get() {
return Vec::new();
}
+ // Echo the requester's incarnation so it can prove this StartView
+ // post-dates its restart (a probe reply, not a stale in-flight message).
+ // Addressed to the requester alone: the nonce proves freshness for that
+ // replica only, and a peer recovering at the same time would read it as
+ // foreign and reject a StartView that is in fact current.
vec![VsrAction::SendStartView {
view: self.view.get(),
op: self.sequencer.current_sequence(),
commit: self.commit_max.get(),
+ incarnation: header.incarnation,
+ target: Some(header.replica),
namespace: self.namespace,
}]
}
@@ -2284,11 +2391,61 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
return Vec::new();
}
- // Skip equal-view SV with old op.
- // Already in this view; re-running reset_view_change_state would
- // cancel subscribers (waking register awaiters Canceled) and clear
- // pipeline for nothing. log_view (not self.view) tracks last-normal view.
- if msg_view == self.log_view.get() && msg_op < self.sequencer.current_sequence() {
+ // Incarnation guard. While Recovering (probing after a restart), only adopt
+ // a StartView that is provably post-restart: a strictly newer view, a head
+ // at or past ours, or one echoing our current incarnation (a reply to our
+ // own probe). Otherwise it may be addressed to a previous incarnation still
+ // in flight, and adopting it could install a stale head and let this replica
+ // act in a view it will not remember after another crash. Inert when no
+ // incarnation is set (partition plane, tests).
+ //
+ // A zero `header.incarnation` makes no claim either way: it is what an
+ // unsolicited StartView carries, and what a peer predating the field sends.
+ // Classifying it stale would have this replica reject a current StartView
+ // from a healthy primary purely because that primary is older, so it falls
+ // through to the view checks that governed before the field existed.
+ let self_incarnation = self.incarnation.get();
+ if self.status.get() == Status::Recovering
+ && self_incarnation != 0
+ && header.incarnation != 0
+ && msg_view <= self.view.get()
+ && msg_op < self.sequencer.current_sequence()
+ && header.incarnation != self_incarnation
+ {
+ tracing::debug!(
+ replica = self.replica,
+ view = msg_view,
+ op = msg_op,
+ incarnation = header.incarnation,
+ self_incarnation,
+ "ignoring StartView while recovering: stale incarnation"
+ );
+ return Vec::new();
+ }
+
+ // Skip an equal-view StartView whose op is below our COMMITTED floor. Such a
+ // message can only be stale: a live primary's head covers every op it ever
+ // told us was committed. Already in this view, so re-running
+ // reset_view_change_state for one would cancel subscribers (waking register
+ // awaiters Canceled) and clear the pipeline for nothing. log_view (not
+ // self.view) tracks the last-normal view.
+ //
+ // The bound is deliberately the commit floor and NOT the sequencer head.
+ // Adopting a StartView drops the head (below) without truncating the WAL, so
+ // a replica that adopted at head H and then restarted re-derives a LONGER
+ // head from its own journal while log_view stays at that same view. Skipping
+ // on the head would drop the primary's StartView there -- including the reply
+ // echoing this replica's own probe -- leaving it to time the probe out and
+ // elect instead, with a DoViewChange of (log_view, discarded_head) that
+ // outranks the real primary's and pushes ops that view already discarded back
+ // over committed bodies.
+ //
+ // TODO(suffix-truncation): re-adoption drops the head again, but the WAL
+ // still holds the discarded suffix, so the primary's next prepare lands on an
+ // op that suffix already occupies and fails `append`'s slot-collision check,
+ // poisoning the journal. Loud beats the silent divergence above; a durable
+ // truncate-from-op primitive is what actually closes it.
+ if msg_view == self.log_view.get() && msg_op < self.commit_min() {
return Vec::new();
}
@@ -2504,10 +2661,15 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
emit_replica_event(SimEventKind::PrimaryElected, &state);
emit_replica_event(SimEventKind::ReplicaStateChanged, &state);
+ // Unsolicited StartView at view-change completion: no probe to answer,
+ // so it carries no incarnation (freshness comes from the newer view) and
+ // goes to every backup.
let action = VsrAction::SendStartView {
view: self.view.get(),
op: new_op,
commit: max_commit,
+ incarnation: 0,
+ target: None,
namespace: self.namespace,
};
emit_sim_event(
@@ -2704,6 +2866,30 @@ where
// `VsrConsensus::next_monotonic_timestamp`.
let timestamp = consensus.next_monotonic_timestamp();
+ // Seal the body integrity field over the payload past the 256-byte header
+ // (the client request bytes, carried through the transmute unchanged).
+ // Computed once by the primary and replicated verbatim, so every backup
+ // stores the same value and the scan verifies it after a crash. The body is
+ // never re-stamped (`restamp_prepare_view` patches only `view`), so this
+ // survives view-change retransmits. The header `checksum` and its `parent`
+ // chain stay `0`: activating them needs the retransmit path to re-seal a
+ // re-stamped header, a separate change.
+ //
+ // Metadata plane only. A partition produce prepare already carries a verified
+ // `batch_checksum` over the same bytes, so a second full-payload pass is pure
+ // cost on the produce path, and it would describe the WRONG bytes:
+ // `stamp_prepare_for_persistence` rewrites the command header INSIDE this
+ // sealed region before the entry is journaled. Leaving those prepares at `0`
+ // is the designed "nothing to verify" sentinel, so a future durable partition
+ // journal skips verification instead of failing every entry as corrupt.
+ let checksum_body = if consensus.namespace == METADATA_CONSENSUS_NAMESPACE {
+ u128::from(calculate_checksum(
+ &self.as_slice()[size_of::<PrepareHeader>()..],
+ ))
+ } else {
+ 0
+ };
+
self.transmute_header(|old, new| {
*new = PrepareHeader {
cluster: consensus.cluster,
@@ -2721,6 +2907,7 @@ where
timestamp,
operation: old.operation,
namespace: old.namespace,
+ checksum_body,
// Copied verbatim: carries the stamped acting user for client
// ops (and the authenticated user on Register), so the in-apply
// RBAC gate resolves the same identity on every backup.
@@ -3173,4 +3360,240 @@ mod timestamp_clamp_tests {
"a clock ahead of the log must stamp real time, not floor + 1"
);
}
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn make_start_view(
+ view: u32,
+ op: u64,
+ replica: u8,
+ incarnation: u128,
+ ) -> Message<StartViewHeader> {
+ let size = std::mem::size_of::<StartViewHeader>();
+ let mut msg = Message::<StartViewHeader>::new(size);
+ let header = bytemuck::checked::try_from_bytes_mut::<StartViewHeader>(
+ &mut msg.as_mut_slice()[..size],
+ )
+ .expect("zeroed bytes are a valid StartViewHeader");
+ header.command = Command2::StartView;
+ header.cluster = 1;
+ header.view = view;
+ header.op = op;
+ header.commit = op;
+ header.replica = replica;
+ header.incarnation = incarnation;
+ header.namespace = METADATA_CONSENSUS_NAMESPACE;
+ header.size = size as u32;
+ msg
+ }
+
+ #[test]
+ fn given_recovering_replica_when_start_view_incarnation_foreign_should_ignore() {
+ // A StartView addressed to a PREVIOUS incarnation, still in flight when the
+ // replica crashed, must be ignored after restart, while the reply echoing
+ // the CURRENT incarnation is adopted. Otherwise the replica could act in a
+ // view it will not remember after another crash.
+ const CURRENT: u128 = 0xB;
+ const STALE: u128 = 0xA;
+
+ // Replica 0 of 3, recovered at view 1 with head op 5, still Recovering
+ // (probing). The primary for view 1 is replica 1 (view % replica_count),
+ // and log_view stays 0 so the equal-view-old-op skip does not fire.
+ let mut consensus = VsrConsensus::with_clock(
+ 1,
+ 0,
+ 3,
+ METADATA_CONSENSUS_NAMESPACE,
+ NoopBus,
+ LocalPipeline::new(),
+ ConsensusClock::system(),
+ );
+ consensus.set_incarnation(CURRENT);
+ consensus.set_view(1);
+ consensus.sequencer().set_sequence(5);
+ assert_eq!(consensus.status(), Status::Recovering);
+
+ // Same view, head behind ours, foreign incarnation: ignored.
+ let stale = make_start_view(1, 4, 1, STALE);
+ assert!(
+ consensus
+ .handle_start_view(PlaneKind::Metadata, stale.header())
+ .is_empty(),
+ "a StartView echoing a previous incarnation must be ignored while recovering"
+ );
+ assert_eq!(
+ consensus.status(),
+ Status::Recovering,
+ "an ignored StartView must not transition status"
+ );
+ assert_eq!(
+ consensus.view(),
+ 1,
+ "an ignored StartView must not change the view"
+ );
+
+ // Same view and head but echoing our current incarnation: adopted, since
+ // the match proves the reply post-dates our restart.
+ let fresh = make_start_view(1, 4, 1, CURRENT);
+ assert!(
+ !consensus
+ .handle_start_view(PlaneKind::Metadata, fresh.header())
+ .is_empty(),
+ "a StartView echoing our current incarnation must be adopted"
+ );
+ assert_eq!(
+ consensus.status(),
+ Status::Normal,
+ "adopting a StartView transitions to Normal"
+ );
+ }
+
+ #[test]
+ fn given_partition_namespace_when_projecting_should_leave_the_body_unsealed() {
+ // The body seal is metadata-only. A partition produce prepare already carries a
+ // verified `batch_checksum` over the same bytes, so a second full-payload hash
+ // is pure cost on the produce path, and it would describe bytes that never reach
+ // the journal: `stamp_prepare_for_persistence` rewrites the command header
+ // inside the sealed region. `0` is the designed "nothing to verify" sentinel, so
+ // a durable partition journal skips verification rather than reading every entry
+ // as corrupt.
+ let seal = |namespace: u64| -> u128 {
+ // Fixed clock: `project` stamps the prepare timestamp, and Miri covers this
+ // crate, where a real clock read is an unsupported syscall.
+ let consensus = VsrConsensus::with_clock(
+ 1,
+ 0,
+ 1,
+ namespace,
+ NoopBus,
+ LocalPipeline::new(),
+ ConsensusClock::new(Rc::new(FixedClock(100_000))),
+ );
+ let header_size = size_of::<RequestHeader>();
+ let body = b"produce payload";
+ let mut msg = Message::<RequestHeader>::new(header_size + body.len());
+ msg.as_mut_slice()[header_size..].copy_from_slice(body);
+ let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ &mut msg.as_mut_slice()[..header_size],
+ )
+ .expect("zeroed bytes are a valid RequestHeader");
+ header.command = Command2::Request;
+ header.client = 1;
+ header.request = 1;
+ header.operation = iggy_binary_protocol::Operation::SendMessages;
+ header.size = u32::try_from(header_size + body.len()).expect("fits u32");
+ msg.project(&consensus).header().checksum_body
+ };
+
+ assert_ne!(
+ seal(METADATA_CONSENSUS_NAMESPACE),
+ 0,
+ "a metadata prepare must be sealed: the WAL scan verifies it after a crash"
+ );
+ assert_eq!(
+ seal(1),
+ 0,
+ "a partition prepare must be left unsealed, since its sealed region is \
+ rewritten before it is journaled"
+ );
+ }
+
+ #[test]
+ fn given_restored_log_view_when_start_view_head_behind_wal_should_adopt() {
+ // A replica that adopted a StartView at head 105 in view 7, then crashed,
+ // recovers log_view = 7 from the superblock but re-derives head 120 from its
+ // own WAL: adoption drops the head without truncating the journal. The
+ // primary's StartView for view 7 then carries an op BEHIND that head, and
+ // skipping it on the head comparison would leave this replica probing until it
+ // elected instead, carrying a DoViewChange of (7, 120) that outranks the real
+ // primary's (7, 105) and resurrects ops view 7 already discarded.
+ //
+ // Replica 0 of 3 at view 7, whose primary is replica 1 (7 % 3). Incarnation
+ // left at 0 so the recovering-replica guard stays inert and this exercises the
+ // equal-view path alone.
+ let mut consensus = VsrConsensus::with_clock(
+ 1,
+ 0,
+ 3,
+ METADATA_CONSENSUS_NAMESPACE,
+ NoopBus,
+ LocalPipeline::new(),
+ ConsensusClock::system(),
+ );
+ consensus.set_view(7);
+ consensus.set_log_view(7);
+ consensus.restore_commit_state(105, 105);
+ consensus.sequencer().set_sequence(120);
+
+ // Below the committed floor: stale by construction, since a live primary's
+ // head covers every op it told us was committed.
+ assert!(
+ consensus
+ .handle_start_view(PlaneKind::Metadata, make_start_view(7, 104, 1, 0).header())
+ .is_empty(),
+ "an equal-view StartView below the commit floor must be skipped"
+ );
+ assert_eq!(
+ consensus.sequencer().current_sequence(),
+ 120,
+ "a skipped StartView must not move the head"
+ );
+
+ // At the committed floor but behind our WAL head: the primary's real head.
+ // Adopt it and drop the discarded suffix.
+ assert!(
+ !consensus
+ .handle_start_view(PlaneKind::Metadata, make_start_view(7, 105, 1, 0).header())
+ .is_empty(),
+ "an equal-view StartView at or above the commit floor must be adopted, \
+ even when its head is behind a WAL suffix the view already discarded"
+ );
+ assert_eq!(
+ consensus.sequencer().current_sequence(),
+ 105,
+ "adoption must drop the head to the primary's, so a later DoViewChange \
+ cannot outrank it with a discarded suffix"
+ );
+ assert_eq!(consensus.status(), Status::Normal);
+ }
+
+ /// The split-brain gate's predicate: `view` and `log_view` each independently
+ /// make the superblock stale, and only a matching `mark_superblock_durable`
+ /// clears it. The simulator proves the withheld-send behavior end to end; this
+ /// pins the predicate the dispatch sites and the debug tripwire both read.
+ #[test]
+ fn given_view_change_when_needs_superblock_persist_should_track_durability() {
+ let mut consensus = VsrConsensus::new(
+ 1,
+ 0,
+ 3,
+ METADATA_CONSENSUS_NAMESPACE,
+ NoopBus,
+ LocalPipeline::new(),
+ );
+ assert!(
+ !consensus.needs_superblock_persist(),
+ "fresh replica: view == view_durable == 0"
+ );
+
+ consensus.set_view(3);
+ assert!(
+ consensus.needs_superblock_persist(),
+ "view advanced but not yet persisted"
+ );
+
+ consensus.mark_superblock_durable(consensus.view(), consensus.log_view());
+ assert!(
+ !consensus.needs_superblock_persist(),
+ "marked durable clears the gate"
+ );
+
+ consensus.set_log_view(3);
+ assert!(
+ consensus.needs_superblock_persist(),
+ "log_view advanced but not yet persisted"
+ );
+
+ consensus.mark_superblock_durable(consensus.view(), consensus.log_view());
+ assert!(!consensus.needs_superblock_persist());
+ }
}
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index 3937eb0e6..8f1ebe756 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -148,7 +148,10 @@ where
}
pub mod client_table;
-pub use client_table::{CachedReply, ClientTable, CommitReply};
+pub use client_table::{
+ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableSnapshot,
+ CommitReply,
+};
// One-shot per `PipelineEntry` for in-process commit awaiters.
pub(crate) mod oneshot;
pub use oneshot::{Canceled, Receiver};
@@ -166,5 +169,7 @@ pub use observability::*;
mod view_change_quorum;
pub use view_change_quorum::*;
+mod vsr_state;
+pub use vsr_state::{VsrState, VsrStateError};
mod vsr_timeout;
pub use vsr_timeout::TimeoutManager;
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
new file mode 100644
index 000000000..6523f38dc
--- /dev/null
+++ b/core/consensus/src/vsr_state.rs
@@ -0,0 +1,224 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The durable VSR state: the consensus numbers a replica must recover from its
+//! own disk after a crash, rather than infer from the WAL or relearn from a
+//! peer.
+//!
+//! The payload `journal::superblock` persists. Consensus owns its meaning and
+//! byte layout; the superblock adds framing, versioning, and integrity.
+//!
+//! Fixed little-endian, so it is byte-identical across replicas and stable
+//! across restarts. Field order is load-bearing: reorders and width changes are
+//! breaking and must ride a superblock version bump.
+
+use std::fmt;
+
+/// Number of bytes [`VsrState::to_bytes`] produces and [`VsrState::try_from`]
+/// expects: `cluster`(16) + `replica_id`(1) + `replica_count`(1) + `view`(4)
+/// + `log_view`(4) + `commit_max`(8) + `checkpoint_op`(8)
+/// + `checkpoint_checksum`(16).
+pub const ENCODED_LEN: usize = 58;
+
+/// The durable consensus state of one replica for one consensus group.
+///
+/// A view a replica acted in must survive a crash, or it can re-participate in
+/// an old view and split the log.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct VsrState {
+ /// Cluster this superblock belongs to. Recovery rejects a mismatch: a copied or
+ /// misplaced data directory. Catches misplacement only, never staleness -- an
+ /// older backup of this replica's own directory carries the right identity and
+ /// passes.
+ pub cluster: u128,
+ /// Replica index this superblock belongs to. Recovery rejects a mismatch, with
+ /// the same misplacement-only scope as `cluster`.
+ pub replica_id: u8,
+ /// Cluster size at write time. Recovery rejects a mismatch: quorum size and the
+ /// `view % replica_count` primary mapping both derive from it, so booting a
+ /// resized cluster without reconfiguration splits the log.
+ pub replica_count: u8,
+ /// Current view.
+ pub view: u32,
+ /// Latest view in which this replica changed its head (adopted a `StartView`
+ /// as a backup, or completed a DVC quorum as primary). `log_view <= view`.
+ pub log_view: u32,
+ /// Highest op known committed by the cluster at write time.
+ ///
+ /// A recovery lower bound and nothing more: recovery reports a gap against what the
+ /// WAL can prove, but cannot act on one, since closing it needs state transfer.
+ /// Kept in the durable record because that is what state transfer will read, and
+ /// adding it later would mean a version bump that invalidates every record.
+ pub commit_max: u64,
+ /// Op of the paired checkpoint (the metadata snapshot's sequence number).
+ pub checkpoint_op: u64,
+ /// Integrity tag of the paired checkpoint, detecting a torn
+ /// snapshot/superblock pairing across a crash.
+ pub checkpoint_checksum: u128,
+}
+
+impl VsrState {
+ /// Encode to the fixed little-endian on-disk layout.
+ #[must_use]
+ pub fn to_bytes(&self) -> [u8; ENCODED_LEN] {
+ let mut out = [0u8; ENCODED_LEN];
+ out[0..16].copy_from_slice(&self.cluster.to_le_bytes());
+ out[16] = self.replica_id;
+ out[17] = self.replica_count;
+ out[18..22].copy_from_slice(&self.view.to_le_bytes());
+ out[22..26].copy_from_slice(&self.log_view.to_le_bytes());
+ out[26..34].copy_from_slice(&self.commit_max.to_le_bytes());
+ out[34..42].copy_from_slice(&self.checkpoint_op.to_le_bytes());
+ out[42..58].copy_from_slice(&self.checkpoint_checksum.to_le_bytes());
+ out
+ }
+}
+
+impl TryFrom<&[u8]> for VsrState {
+ type Error = VsrStateError;
+
+ fn try_from(bytes: &[u8]) -> Result<Self, Self::Error> {
+ // One length check up front puts every field slice below in bounds by
+ // construction, so the `try_into`s cannot fail.
+ let bytes: &[u8; ENCODED_LEN] =
+ bytes.try_into().map_err(|_| VsrStateError::WrongLength {
+ expected: ENCODED_LEN,
+ actual: bytes.len(),
+ })?;
+ let state = Self {
+ cluster: u128::from_le_bytes(field(bytes, 0)),
+ replica_id: bytes[16],
+ replica_count: bytes[17],
+ view: u32::from_le_bytes(field(bytes, 18)),
+ log_view: u32::from_le_bytes(field(bytes, 22)),
+ commit_max: u64::from_le_bytes(field(bytes, 26)),
+ checkpoint_op: u64::from_le_bytes(field(bytes, 34)),
+ checkpoint_checksum: u128::from_le_bytes(field(bytes, 42)),
+ };
+ // A record violating `log_view <= view` decodes into a replica that looks
+ // healthy locally while `DoViewChangeHeader::validate` makes every peer drop
+ // its DVCs, so it can never conclude a view change. Length validation alone
+ // would let corruption inside the checksummed region through as a live
+ // consensus state.
+ if state.log_view > state.view {
+ return Err(VsrStateError::LogViewAheadOfView {
+ view: state.view,
+ log_view: state.log_view,
+ });
+ }
+ Ok(state)
+ }
+}
+
+/// Copy a fixed-width field out of the length-validated record. Always in bounds
+/// for a `[u8; ENCODED_LEN]`, so the slice conversion is infallible.
+fn field<const N: usize>(bytes: &[u8; ENCODED_LEN], start: usize) -> [u8; N] {
+ let mut out = [0u8; N];
+ out.copy_from_slice(&bytes[start..start + N]);
+ out
+}
+
+/// Failure decoding a [`VsrState`] from bytes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum VsrStateError {
+ /// The byte slice was not exactly [`ENCODED_LEN`] long.
+ WrongLength { expected: usize, actual: usize },
+ /// The record violates `log_view <= view`, so it cannot be a state any replica
+ /// reached.
+ LogViewAheadOfView { view: u32, log_view: u32 },
+}
+
+impl fmt::Display for VsrStateError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::WrongLength { expected, actual } => {
+ write!(f, "VsrState needs {expected} bytes, got {actual}")
+ }
+ Self::LogViewAheadOfView { view, log_view } => write!(
+ f,
+ "VsrState log_view {log_view} exceeds view {view}, which no replica can reach"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for VsrStateError {}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_distinct_fields_when_to_bytes_should_pin_offsets_and_round_trip() {
+ // Guards the on-disk layout: a reorder or width change trips this, forcing a
+ // deliberate superblock version bump. Distinct per-field values make a swap
+ // between two same-width fields observable.
+ let state = VsrState {
+ cluster: 1,
+ replica_id: 2,
+ replica_count: 5,
+ // Distinct per-field values make a swap between two same-width fields
+ // observable, and `view > log_view` keeps the record decodable.
+ view: 4,
+ log_view: 3,
+ commit_max: 6,
+ checkpoint_op: 7,
+ checkpoint_checksum: 8,
+ };
+ let bytes = state.to_bytes();
+ assert_eq!(bytes.len(), ENCODED_LEN);
+ assert_eq!(bytes[0], 1, "cluster low byte");
+ assert_eq!(bytes[16], 2, "replica_id");
+ assert_eq!(bytes[17], 5, "replica_count");
+ assert_eq!(bytes[18], 4, "view low byte");
+ assert_eq!(bytes[22], 3, "log_view low byte");
+ assert_eq!(bytes[26], 6, "commit_max low byte");
+ assert_eq!(bytes[34], 7, "checkpoint_op low byte");
+ assert_eq!(bytes[42], 8, "checkpoint_checksum low byte");
+
+ assert_eq!(VsrState::try_from(&bytes[..]).unwrap(), state);
+ assert!(VsrState::try_from(&bytes[..ENCODED_LEN - 1]).is_err());
+ }
+
+ #[test]
+ fn given_log_view_past_view_when_decoded_should_reject() {
+ // Corruption inside the checksummed region can produce a length-valid record
+ // that violates `log_view <= view`. Decoding it yields a replica that looks
+ // healthy locally while every peer drops its DoViewChange messages
+ // (`DoViewChangeHeader::validate`), so it can never conclude a view change.
+ let mut bytes = VsrState {
+ cluster: 1,
+ replica_id: 0,
+ replica_count: 3,
+ view: 4,
+ log_view: 4,
+ commit_max: 0,
+ checkpoint_op: 0,
+ checkpoint_checksum: 0,
+ }
+ .to_bytes();
+ bytes[22] = 5; // log_view = 5, view stays 4
+
+ assert_eq!(
+ VsrState::try_from(&bytes[..]),
+ Err(VsrStateError::LogViewAheadOfView {
+ view: 4,
+ log_view: 5
+ })
+ );
+ }
+}
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index d0dc17dba..f01912283 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -31,7 +31,7 @@ ignored = ["cfg_aliases"]
ci-qemu = []
default = ["login-session"]
login-session = ["dep:zbus-secret-service-keyring-store"]
-vsr = ["iggy/vsr"]
+vsr = ["dep:consensus", "dep:journal", "iggy/vsr"]
[dependencies]
assert_cmd = { workspace = true }
@@ -43,6 +43,9 @@ bytes = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
configs_derive = { workspace = true }
+# vsr-only: decode a metadata replica's durable `VsrState` off disk in the
+# superblock recovery test.
+consensus = { workspace = true, optional = true }
ctor = { workspace = true }
deltalake = { workspace = true }
dtor = { workspace = true }
@@ -58,6 +61,9 @@ iggy_common = { workspace = true }
# `build_label` function — keeping the test and production label format in lock-step.
iggy_connector_doris_sink = { path = "../connectors/sinks/doris_sink" }
iggy_connector_sdk = { workspace = true, features = ["api"] }
+# vsr-only: locate and decode the on-disk superblock slot files in the recovery
+# test (`SLOT_FILE_NAMES`, `decode_slots`).
+journal = { workspace = true, optional = true }
jsonwebtoken = { workspace = true }
keyring-core = { workspace = true }
lazy_static = { workspace = true }
diff --git a/core/integration/tests/server/cluster_view_durability_vsr.rs b/core/integration/tests/server/cluster_view_durability_vsr.rs
new file mode 100644
index 000000000..d69d673e7
--- /dev/null
+++ b/core/integration/tests/server/cluster_view_durability_vsr.rs
@@ -0,0 +1,266 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Metadata-plane view durability across a real view change and process restarts,
+//! over `iggy-server-ng`'s production superblock path.
+//!
+//! The superblock exists so a replica recovers a view it already acted in from its
+//! OWN disk after a crash, instead of inferring a stale view from the WAL or
+//! relearning it from a peer. That is the safety property stopping a recovered
+//! replica from re-voting in an old view and splitting the log. This drives a genuine
+//! metadata view change, crashing the view-0 primary so the survivors elect a new
+//! one, then proves the elected view is durable: it lands in a survivor's on-disk
+//! `VsrState` as `view` / `log_view >= 1` and survives that survivor's own process
+//! restart. The stream committed before the crash comes back too, so metadata
+//! op/commit recovery from the WAL rides along.
+//!
+//! The production counterpart to the simulator's
+//! `given_advanced_view_when_metadata_replica_restarts_should_recover_view_from_superblock`,
+//! which proves the same property over the in-memory `SimSuperblock`; this proves it
+//! over the real `PingPongSuperblock` and `restore_metadata_consensus` the sim stubs
+//! out.
+//!
+//! Deliberately metadata-only, a stream and no topic: a topic would create a
+//! partition consensus group whose own view change is orthogonal to the plane under
+//! test and would only add flakiness. Restarts keep a metadata quorum of 2 of 3 live
+//! at every instant, bringing the crashed primary back before bouncing the survivor,
+//! so recovery is exercised without wedging the cluster below quorum.
+//!
+//! vsr-only: a metadata view change has no analog on the single-process legacy
+//! server, and the superblock is server-ng's durable consensus record.
+
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+use consensus::VsrState;
+use iggy::prelude::*;
+use integration::harness::{TestBinary, TestHarness};
+use integration::iggy_harness;
+use journal::superblock::{SLOT_FILE_NAMES, SuperblockContents, decode_slots};
+use tokio::time::sleep;
+
+const STREAM_NAME: &str = "view-durability-stream";
+
+/// A view change plus a rejoin is not instant (primary timeout, SVC, DVC, StartView,
+/// then the restarted node's probe and repair), and CI runners are slow. 60s bounds
+/// the worst case without hanging the suite.
+const CONVERGE_TIMEOUT: Duration = Duration::from_secs(60);
+const POLL_INTERVAL: Duration = Duration::from_millis(250);
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_advanced_metadata_view_when_survivor_restarts_should_recover_view_from_superblock(
+ harness: &mut TestHarness,
+) {
+ // One shard per node keeps the metadata consensus on shard 0, the shard every
+ // client connection lands on, so leadership shows up over the wire. Three
+ // nodes give quorum 2, so the cluster keeps a quorum through a single crash.
+
+ // Baseline: node 0 is the view-0 metadata primary (its replica id is 0, and
+ // the primary of view 0 is replica 0). Commit a stream through it, so the
+ // metadata group has committed state to recover later and exactly one leader
+ // is visible.
+ let client = connect(harness, 0).await;
+ client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("create stream on the view-0 primary");
+ assert_eq!(
+ leader_count(&client).await,
+ Some(1),
+ "a healthy cluster must show exactly one metadata leader before the crash"
+ );
+ drop(client);
+ // Let the committed prepare settle (fsync per entry) on the survivors before
+ // the primary dies, so the stream is durable on a node that outlives it.
+ sleep(Duration::from_secs(1)).await;
+
+ // Force a metadata view change: crash the view-0 primary. Its silence trips
+ // the survivors' primary timeout; they run SVC/DVC and elect a new primary at
+ // view >= 1, each persisting the advanced view through the superblock gate
+ // before it casts a view-scoped vote.
+ harness
+ .node_mut(0)
+ .stop()
+ .expect("crash the metadata primary (node 0)");
+
+ // Wait for the advanced view to become DURABLE on a survivor's own disk. Node 2
+ // has to take part in the election, since quorum needs both survivors, so its
+ // superblock must reach view >= 1. Reading the durable record directly is the
+ // race-free signal that the view change happened and was persisted, stronger than
+ // a leadership poll, which could read the stale view-0 roster mid-election.
+ let view_before = wait_for_advanced_view(harness, 2).await;
+ assert!(
+ view_before.view >= 1 && view_before.log_view >= 1,
+ "a survivor that took part in the view change must persist view/log_view >= 1, \
+ got {view_before:?}"
+ );
+
+ // Bring the crashed primary back so it rejoins from its own disk, via recover(),
+ // the superblock, and a view probe. The two survivors still form a quorum while it
+ // is down, so this never stalls; once back the cluster is 3/3, keeping a quorum
+ // alive when the survivor is bounced next.
+ harness
+ .node_mut(0)
+ .start()
+ .expect("restart the crashed primary (node 0)");
+ let rejoined = wait_for_advanced_view(harness, 0).await;
+ assert!(
+ rejoined.view >= view_before.view,
+ "the rejoined primary must adopt the advanced view (>= {}), not resume the stale view 0, \
+ got {rejoined:?}",
+ view_before.view
+ );
+ // Let the 3/3 mesh settle so the survivor restart below keeps a live quorum.
+ sleep(Duration::from_secs(1)).await;
+
+ // Restart the survivor that advanced its view. It drops all in-memory
+ // consensus state and must recover its advanced view from its own superblock
+ // via `restore_metadata_consensus`, not reset to a fresh 0. The other two
+ // nodes hold the quorum while it is down.
+ harness
+ .node_mut(2)
+ .stop()
+ .expect("stop the survivor (node 2)");
+ harness
+ .node_mut(2)
+ .start()
+ .expect("restart the survivor (node 2)");
+
+ // The reformed cluster must serve again under exactly one leader, and the stream
+ // committed before the view change must still resolve, proving the metadata
+ // op/commit point and state machine recovered from disk alongside the view.
+ wait_until_serving_with_single_leader(harness, &[0, 1, 2]).await;
+
+ // The advanced view survived the survivor's own restart: re-read its superblock
+ // once resettled and confirm it did not regress toward a fresh 0.
+ let view_after = wait_for_advanced_view(harness, 2).await;
+ assert!(
+ view_after.view >= view_before.view && view_after.log_view >= view_before.log_view,
+ "the durable view must survive node 2's restart without regressing: \
+ before={view_before:?}, after={view_after:?}"
+ );
+}
+
+/// Connect a root-authenticated TCP client to a specific node.
+async fn connect(harness: &TestHarness, node: usize) -> IggyClient {
+ harness
+ .node(node)
+ .tcp_client()
+ .expect("tcp client builder")
+ .with_root_login()
+ .connect()
+ .await
+ .unwrap_or_else(|e| panic!("connect to node {node}: {e}"))
+}
+
+/// Connect to the first node in `nodes` that accepts a connection, `None` when none
+/// do (mid-election, or a node still restarting).
+async fn connect_any(harness: &TestHarness, nodes: &[usize]) -> Option<IggyClient> {
+ for &node in nodes {
+ if let Ok(builder) = harness.node(node).tcp_client()
+ && let Ok(client) = builder.with_root_login().connect().await
+ {
+ return Some(client);
+ }
+ }
+ None
+}
+
+/// Number of nodes the metadata roster marks as leader, `None` if the query fails
+/// (mid-election, connection dropped).
+async fn leader_count(client: &IggyClient) -> Option<usize> {
+ let metadata = client.get_cluster_metadata().await.ok()?;
+ Some(
+ metadata
+ .nodes
+ .iter()
+ .filter(|node| node.role == ClusterNodeRole::Leader)
+ .count(),
+ )
+}
+
+/// Decode a node's durable metadata `VsrState` from its on-disk superblock, `None` if
+/// no record exists yet. Reads the two slot files with blocking I/O, since this is a
+/// tokio test off any compio runtime, and decodes them through the journal's own
+/// newest-verifying-wins selection, so the test sees exactly what
+/// `PingPongSuperblock::read_latest` would.
+///
+/// # Panics
+/// If a slot holds bytes that do not verify. No step here corrupts one, so that is a
+/// real durability bug; returning `None` would let the pollers below read it as "not
+/// written yet" and time out on a misleading message.
+fn read_superblock_state(data_path: &Path) -> Option<VsrState> {
+ // `<data_dir>/metadata/` is where shard 0 opens its `PingPongSuperblock`.
+ let dir = data_path.join("metadata");
+ let slot_a = std::fs::read(dir.join(SLOT_FILE_NAMES[0])).ok();
+ let slot_b = std::fs::read(dir.join(SLOT_FILE_NAMES[1])).ok();
+ match decode_slots(slot_a.as_deref(), slot_b.as_deref()) {
+ SuperblockContents::Present(payload) => VsrState::try_from(payload.as_slice()).ok(),
+ SuperblockContents::Empty => None,
+ SuperblockContents::Unreadable { version } => panic!(
+ "superblock at {} is unreadable (version {version:?}); no test step corrupts it",
+ dir.display()
+ ),
+ }
+}
+
+/// Poll `node`'s superblock until it holds a settled advanced view, returning that
+/// state. Panics on timeout.
+///
+/// The gate persists `view` the moment a replica advances it to VOTE, so a bare
+/// `view >= 1` check races the election: it can observe the intermediate
+/// `view = 1, log_view = 0` a backup writes before adopting the new primary's
+/// `StartView`, which moves its head and sets `log_view = view`. Waiting for
+/// `log_view >= 1`, which implies `view >= 1` since `log_view <= view`, is the settled
+/// signal that the replica durably adopted the new view's log.
+async fn wait_for_advanced_view(harness: &TestHarness, node: usize) -> VsrState {
+ let data_path = harness.node(node).data_path();
+ let deadline = Instant::now() + CONVERGE_TIMEOUT;
+ loop {
+ if let Some(state) = read_superblock_state(&data_path)
+ && state.log_view >= 1
+ {
+ return state;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "node {node} did not persist a settled advanced view (log_view >= 1) \
+ within {CONVERGE_TIMEOUT:?}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
+
+/// Poll until a node serves the pre-crash stream AND the roster shows exactly one
+/// leader. Panics on timeout.
+async fn wait_until_serving_with_single_leader(harness: &TestHarness, nodes: &[usize]) {
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+ let deadline = Instant::now() + CONVERGE_TIMEOUT;
+ loop {
+ if let Some(client) = connect_any(harness, nodes).await
+ && matches!(client.get_stream(&stream_id).await, Ok(Some(_)))
+ && leader_count(&client).await == Some(1)
+ {
+ return;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "cluster did not re-form (stream served, single leader) within {CONVERGE_TIMEOUT:?}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
diff --git a/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs b/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs
new file mode 100644
index 000000000..4fca67b0b
--- /dev/null
+++ b/core/integration/tests/server/metadata_checkpoint_recovery_vsr.rs
@@ -0,0 +1,138 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Metadata checkpoint-fold recovery across a solo restart, over
+//! `iggy-server-ng`'s production snapshot and WAL path.
+//!
+//! Between checkpoints a replica recovers its metadata by replaying the WAL. Once the
+//! WAL fills, the `SnapshotCoordinator` checkpoints: it persists `snapshot.bin`, pairs
+//! it in the superblock, and DRAINS the snapshotted prefix from the WAL. A restart
+//! after that must fold the snapshot back in as the recovery floor and replay only the
+//! committed suffix on top, rather than rely on a full WAL replay, which no longer
+//! holds the drained ops. This drives a real checkpoint by pushing the metadata WAL
+//! past `CHECKPOINT_MARGIN`, restarts the process, and asserts that a stream from the
+//! drained prefix, recoverable only from the snapshot, and one from the WAL suffix
+//! both survive.
+//!
+//! Solo on purpose: 1-of-1 quorum commits every op the instant it is journaled, so
+//! bulk creation is fast and the WAL is fully committed with no uncommitted suffix to
+//! reconcile, exercising checkpoint and snapshot-fold recovery in isolation without an
+//! election in the mix.
+//!
+//! vsr-only: the metadata snapshot/superblock checkpoint pairing is server-ng's.
+
+use std::time::{Duration, Instant};
+
+use iggy::prelude::*;
+use integration::harness::{TestBinary, TestHarness};
+use integration::iggy_harness;
+use tokio::time::sleep;
+
+/// The prepare WAL has `SLOT_COUNT = 1024` slots and the coordinator checkpoints at
+/// `<= CHECKPOINT_MARGIN` (64) remaining, so after ~960 uncheckpointed ops. 1024
+/// stream creates clears that with room for a WAL suffix above the checkpoint, and
+/// stays under the 4096 stream namespace cap.
+const STREAMS: u32 = 1024;
+
+const RECOVER_TIMEOUT: Duration = Duration::from_secs(60);
+const POLL_INTERVAL: Duration = Duration::from_millis(200);
+
+fn stream_name(index: u32) -> String {
+ format!("ckpt-stream-{index}")
+}
+
+#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_checkpointed_metadata_when_solo_replica_restarts_should_recover_from_snapshot_and_wal(
+ harness: &mut TestHarness,
+) {
+ let client = connect(harness).await;
+ for index in 0..STREAMS {
+ client
+ .create_stream(&stream_name(index))
+ .await
+ .unwrap_or_else(|e| panic!("create stream {index}: {e}"));
+ }
+ drop(client);
+
+ // Crossing CHECKPOINT_MARGIN must have driven the coordinator to persist a snapshot
+ // and drain the WAL prefix behind it.
+ let snapshot_path = harness
+ .node(0)
+ .data_path()
+ .join("metadata")
+ .join("snapshot.bin");
+ let snapshot_len = std::fs::metadata(&snapshot_path).map(|m| m.len()).ok();
+ assert!(
+ snapshot_len.is_some_and(|len| len > 0),
+ "{STREAMS} committed metadata ops must cross CHECKPOINT_MARGIN and persist a \
+ non-empty snapshot at {}, got {snapshot_len:?}",
+ snapshot_path.display()
+ );
+
+ // Restart the solo node: recovery loads the snapshot, holding the drained prefix,
+ // and replays the committed WAL suffix on top.
+ harness.node_mut(0).stop().expect("stop the solo node");
+ harness.node_mut(0).start().expect("restart the solo node");
+
+ // The first stream sits far below the checkpoint op, so it was drained from the WAL
+ // and can come back only from the snapshot; the last stream is in the WAL suffix
+ // above the checkpoint. Both surviving proves snapshot-fold plus suffix recovery,
+ // not a bare WAL replay.
+ let client = wait_for_stream(harness, &stream_name(0)).await;
+ for name in [stream_name(0), stream_name(STREAMS - 1)] {
+ assert!(
+ client
+ .get_stream(&Identifier::named(&name).unwrap())
+ .await
+ .expect("get stream after restart")
+ .is_some(),
+ "stream {name} must survive the checkpointed restart"
+ );
+ }
+}
+
+/// Connect a root-authenticated TCP client to the solo node.
+async fn connect(harness: &TestHarness) -> IggyClient {
+ harness
+ .node(0)
+ .tcp_client()
+ .expect("tcp client builder")
+ .with_root_login()
+ .connect()
+ .await
+ .expect("connect to the solo node")
+}
+
+/// Poll the solo node until it is back up and serving `stream`, returning the connected
+/// client. Panics on timeout.
+async fn wait_for_stream(harness: &TestHarness, stream: &str) -> IggyClient {
+ let stream_id = Identifier::named(stream).unwrap();
+ let deadline = Instant::now() + RECOVER_TIMEOUT;
+ loop {
+ if let Ok(builder) = harness.node(0).tcp_client()
+ && let Ok(client) = builder.with_root_login().connect().await
+ && matches!(client.get_stream(&stream_id).await, Ok(Some(_)))
+ {
+ return client;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "solo node did not recover and serve {stream} within {RECOVER_TIMEOUT:?}"
+ );
+ sleep(POLL_INTERVAL).await;
+ }
+}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 25cac503d..5681bf714 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -44,6 +44,14 @@ mod http_tls;
// Binary GetClusterMetadata must serve the real roster from a VSR cluster.
#[cfg(feature = "vsr")]
mod cluster_metadata_vsr;
+// A metadata view change must persist the advanced view and recover it from disk
+// across a replica restart.
+#[cfg(feature = "vsr")]
+mod cluster_view_durability_vsr;
+// A metadata checkpoint must drain the WAL and recover from the snapshot fold plus
+// the WAL suffix across a restart.
+#[cfg(feature = "vsr")]
+mod metadata_checkpoint_recovery_vsr;
// 80-case race matrix with hardcoded HTTP variants (test_matrix bypasses
// the harness transport filter).
mod concurrent_addition;
diff --git a/core/journal/Cargo.toml b/core/journal/Cargo.toml
index fb76787a6..0a522f1c7 100644
--- a/core/journal/Cargo.toml
+++ b/core/journal/Cargo.toml
@@ -33,6 +33,8 @@ bytemuck = { workspace = true }
compio = { workspace = true }
iggy_binary_protocol = { workspace = true }
server_common = { workspace = true }
+tracing = { workspace = true }
+twox-hash = { workspace = true }
[dev-dependencies]
futures = { workspace = true }
diff --git a/core/journal/src/lib.rs b/core/journal/src/lib.rs
index 629a2e4ec..a8096055e 100644
--- a/core/journal/src/lib.rs
+++ b/core/journal/src/lib.rs
@@ -17,9 +17,11 @@
use std::io;
use std::ops::{Deref, RangeInclusive};
+use std::rc::Rc;
pub mod file_storage;
pub mod prepare_journal;
+pub mod superblock;
pub trait Journal<S>
where
@@ -80,3 +82,16 @@ pub trait JournalHandle {
fn handle(&self) -> &Self::Target;
}
+
+/// Forwarding impl so a journal held behind an `Rc` still drives the shard
+/// through [`JournalHandle`]. The deterministic simulator needs this to retain
+/// the metadata WAL across a replica restart: the bytes and index survive the
+/// shard being dropped and rebuilt.
+impl<T: JournalHandle> JournalHandle for Rc<T> {
+ type Storage = T::Storage;
+ type Target = T::Target;
+
+ fn handle(&self) -> &Self::Target {
+ (**self).handle()
+ }
+}
diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs
index 802b36d81..bc24d5c16 100644
--- a/core/journal/src/prepare_journal.rs
+++ b/core/journal/src/prepare_journal.rs
@@ -25,6 +25,7 @@ use std::fmt;
use std::io;
use std::ops::RangeInclusive;
use std::path::{Path, PathBuf};
+use twox_hash::XxHash3_64;
const HEADER_SIZE: usize = size_of::<PrepareHeader>();
@@ -35,6 +36,21 @@ const HEADER_SIZE: usize = size_of::<PrepareHeader>();
/// multi-GiB allocation during the WAL scan.
const MAX_ENTRY_SIZE: u64 = 64 * 1024 * 1024;
+/// `checksum_body` of an entry no producer sealed: written by a build predating
+/// body sealing, or replicated from a primary predating it (backups journal the
+/// header verbatim, so the zero travels along). Indistinguishable from
+/// sealed-then-corrupted, so the scan skips verification and counts these rather
+/// than reject a WAL it cannot prove is damaged. Sound as a sentinel because a sealed
+/// body hashing to `0` is a 1-in-2^64 event, not an impossible one: some input maps
+/// there, we just do not know which. The consequence of that collision is a single
+/// entry replayed unverified, the same treatment a genuinely unsealed one gets, which
+/// is why the sentinel is worth the odds.
+///
+/// Never re-sealed on receipt: the producer seals once and every replica stores that
+/// verbatim, so re-sealing locally would diverge the header bytes and break the
+/// parent chain in the TODO below.
+const CHECKSUM_BODY_UNSEALED: u128 = 0;
+
/// Number of slots in the journal ring buffer.
///
/// Must be larger than the maximum number of entries between consecutive
@@ -147,6 +163,10 @@ pub struct PrepareJournal {
/// let more committed-but-unsnapshotted entries accumulate between
/// checkpoints (more WAL churn headroom, more memory).
slot_count: usize,
+ /// How many entries the opening scan accepted unverified because no producer
+ /// sealed them ([`CHECKSUM_BODY_UNSEALED`]). Fixed at `open`, since every later
+ /// entry comes from a sealing producer.
+ unsealed_entries: u64,
}
/// Captured cause of journal poisoning. `stage` names the drain step
@@ -174,31 +194,55 @@ const fn slot_for_op(op: u64, slot_count: usize) -> usize {
/// Repair a damaged WAL tail by truncating to `pos`, or surface a loud
/// error when truncation would be unsafe.
///
-/// Truncation is only sound for a torn final append, which writes at most
-/// one entry's worth of bytes. If more than `MAX_ENTRY_SIZE` bytes follow
-/// `pos`, the damage is mid-file: truncating would silently discard every
-/// committed entry after it, so this hard-errors instead of repairing.
+/// Truncation is sound only for a torn final append. The question that decides it
+/// is the one the interior-corruption branch in `scan` asks: does a complete entry
+/// follow? An entry only exists past `pos` if an `append` completed after the damaged
+/// region, and each `append` fsyncs before its `PrepareOk`, so discarding it would
+/// drop an entry that was durable, acked, and possibly quorum-committed. `pos` alone
+/// cannot answer this: a bit-flip in a header's `size` field loses the entry boundary
+/// while leaving intact entries behind it, and those bytes are what the probe finds.
+///
+/// The `> MAX_ENTRY_SIZE` test is kept as a second refusal, not as the classifier:
+/// one entry per `append` + fsync means a torn tail is at most one entry wide, so a
+/// larger unparsable region is damage of some other kind and not this function's to
+/// repair.
+///
+/// Both outcomes are traced. A silent truncation is the failure mode that hides
+/// durable data loss from an operator who has no other signal.
#[allow(clippy::future_not_send)]
async fn truncate_or_fail(
storage: &FileStorage,
pos: u64,
reason: &'static str,
) -> Result<(), JournalError> {
- let trailing = storage.file_len().saturating_sub(pos);
- // Sound only because this WAL appends exactly one entry per `append`
- // + fsync, so a torn tail is at most one entry wide. A batched-append
- // WAL could leave a torn tail many entries wide, and this
- // `> MAX_ENTRY_SIZE` check would misclassify it as mid-file damage.
+ let file_len = storage.file_len();
+ let trailing = file_len.saturating_sub(pos);
+ if let Some(entry_pos) = find_complete_entry(storage, pos, file_len).await? {
+ return Err(JournalError::Io(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "mid-file WAL corruption at pos {pos}: {reason}; a complete entry \
+ starts at pos {entry_pos} ({trailing} trailing bytes), so this is not \
+ a torn tail; refusing to truncate and discard committed entries"
+ ),
+ )));
+ }
if trailing > MAX_ENTRY_SIZE {
return Err(JournalError::Io(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"mid-file WAL corruption at pos {pos}: {reason}; {trailing} bytes \
- follow (> MAX_ENTRY_SIZE {MAX_ENTRY_SIZE}), refusing to truncate \
- and discard committed entries"
+ follow (> MAX_ENTRY_SIZE {MAX_ENTRY_SIZE}), wider than one append can \
+ tear; refusing to truncate and discard committed entries"
),
)));
}
+ tracing::warn!(
+ pos,
+ trailing,
+ reason,
+ "truncating torn WAL tail; no complete entry follows the damage"
+ );
storage.truncate(pos).await?;
// The repair must be crash-durable. `FileStorage::truncate` is a
// bare `set_len`; without this fsync a power loss right after the
@@ -208,12 +252,80 @@ async fn truncate_or_fail(
Ok(())
}
-/// Best-effort unlink of the drain temp file on any error path between
-/// `File::create(wal.tmp)` and the atomic `rename`. Without this, every
-/// failed drain leaks a `wal.tmp` next to the WAL; the next drain
-/// truncates it on re-create so safety holds, but operators see the tmp
-/// files accumulate across crashed/aborted drains. `defuse` is called
-/// after a successful rename so the now-renamed file is not removed.
+/// Position of the first structurally complete entry starting after `from`, or
+/// `None` when the region holds no entry a scan could read.
+///
+/// Entry starts are byte-aligned in the file (`append` writes exact-sized buffers
+/// with no padding) and the damaged entry's own `size` cannot be trusted, so every
+/// offset is a candidate. `command` is a `#[repr(u8)]` discriminant at a fixed offset,
+/// so it pre-filters ~255 of every 256 offsets down to a byte compare, and the full
+/// structural validation runs only on the rest. Cost is bounded by the trailing
+/// region, which the caller refuses above `MAX_ENTRY_SIZE`, and this runs once on a
+/// boot that is already repairing.
+#[allow(clippy::future_not_send)]
+async fn find_complete_entry(
+ storage: &FileStorage,
+ from: u64,
+ file_len: u64,
+) -> Result<Option<u64>, JournalError> {
+ const COMMAND_OFFSET: usize = std::mem::offset_of!(PrepareHeader, command);
+ /// Fresh bytes read per pass, over the `HEADER_SIZE` overlap that lets a
+ /// candidate straddle two passes.
+ const PROBE_STRIDE: usize = 64 * 1024;
+
+ // The entry AT `from` already failed to parse, so it is not a candidate.
+ let mut base = from.saturating_add(1);
+ let mut buf: Vec<u8> = Vec::new();
+ let mut scratch = Owned::<16>::zeroed(HEADER_SIZE);
+ while base + HEADER_SIZE as u64 <= file_len {
+ let want = usize::try_from((file_len - base).min((PROBE_STRIDE + HEADER_SIZE) as u64))
+ .unwrap_or(PROBE_STRIDE + HEADER_SIZE);
+ if buf.len() != want {
+ buf = vec![0u8; want];
+ }
+ buf = storage.read_at(base, buf).await?;
+
+ let last_start = want - HEADER_SIZE;
+ for offset in 0..=last_start {
+ if buf[offset + COMMAND_OFFSET] != Command2::Prepare as u8 {
+ continue;
+ }
+ let candidate = &buf[offset..offset + HEADER_SIZE];
+ let Some(header) = valid_entry_header(&mut scratch, candidate) else {
+ continue;
+ };
+ let start = base + offset as u64;
+ if start + u64::from(header.size) <= file_len {
+ return Ok(Some(start));
+ }
+ }
+ base += last_start as u64 + 1;
+ }
+ Ok(None)
+}
+
+/// The structural checks `scan` applies before it trusts a header's `size`: a valid
+/// bit pattern, the `Prepare` command, and a size that spans at least the header and
+/// at most one entry.
+fn valid_entry_header(scratch: &mut Owned<16>, bytes: &[u8]) -> Option<PrepareHeader> {
+ scratch.as_mut_slice().copy_from_slice(bytes);
+ let header = *bytemuck::checked::try_from_bytes::<PrepareHeader>(scratch.as_slice()).ok()?;
+ if header.command != Command2::Prepare
+ || (header.size as usize) < HEADER_SIZE
+ || u64::from(header.size) > MAX_ENTRY_SIZE
+ {
+ return None;
+ }
+ Some(header)
+}
+
+/// Best-effort unlink of a temp file on any error path between its
+/// `File::create` and the atomic `rename`. Without this, every failed
+/// write leaks a tmp file next to its target; the next attempt truncates
+/// it on re-create so safety holds, but operators see the tmp files
+/// accumulate across crashed/aborted writes. `defuse` is called after a
+/// successful rename so the now-renamed file is not removed. Shared with
+/// `superblock::atomic_replace`, which has the same window.
///
/// `Drop` cannot be async, so the unlink is a blocking `std::fs::remove_file`.
/// This only runs on the drain failure path (already returning an error),
@@ -221,17 +333,17 @@ async fn truncate_or_fail(
/// may already be gone (e.g. rename succeeded but a later step failed
/// and we defused too late) and there is no useful recovery from a
/// failed cleanup unlink.
-struct TmpFileGuard {
+pub(crate) struct TmpFileGuard {
path: PathBuf,
armed: bool,
}
impl TmpFileGuard {
- const fn new(path: PathBuf) -> Self {
+ pub(crate) const fn new(path: PathBuf) -> Self {
Self { path, armed: true }
}
- fn defuse(mut self) {
+ pub(crate) fn defuse(mut self) {
self.armed = false;
}
}
@@ -310,11 +422,17 @@ impl PrepareJournal {
let mut headers: Vec<Option<PrepareHeader>> = vec![None; slot_count];
let mut offsets: Vec<Option<u64>> = vec![None; slot_count];
let mut last_op: Option<u64> = None;
+ let mut unsealed_entries: u64 = 0;
let mut pos: u64 = 0;
let mut header_buf = vec![0u8; HEADER_SIZE];
// Reused 16-aligned scratch (PrepareHeader has u128 fields). Avoids
// per-iteration 4 KiB-aligned alloc; bytes never become a `Message`.
let mut aligned = Owned::<16>::zeroed(HEADER_SIZE);
+ // Reused across entries, skipping a fresh allocation when consecutive
+ // bodies match in size (the common case for metadata prepares). The read
+ // below fills the buffer to capacity, so it must be sized to exactly the
+ // entry body; a size change takes a new exact-sized buffer.
+ let mut body_buf: Vec<u8> = Vec::new();
while pos + HEADER_SIZE as u64 <= file_len {
// Read the 256-byte header
@@ -344,26 +462,72 @@ impl PrepareJournal {
let entry_size = u64::from(header.size);
- // TODO(hubcio): verify `header.checksum` / `header.checksum_body`
- // against the entry body during scan and route a mismatch
- // through `truncate_or_fail`. Blocked on the writer side: the
- // `PrepareHeader` projection in consensus builds prepares with
- // `..Default::default()` so the integrity fields are always 0.
- // Until a producer computes them, verification here would be
- // trivially-passing noise. Without it, a body bit-flip that
- // leaves the header valid is replayed silently as corrupt
- // state. Committed bytes are meant to be byte-identical across
- // replicas (deterministic apply, timestamp replicated not
- // re-projected), so once the producer computes the integrity fields
- // they should agree on every node and this check can be turned on
- // without per-replica false positives.
-
// Check if the full entry fits
if pos + entry_size > file_len {
truncate_or_fail(&storage, pos, "truncated entry at tail").await?;
break;
}
+ // Verify the body integrity field the primary sealed at prepare-build
+ // (`checksum_body`, XxHash3_64 over the payload past the header,
+ // replicated verbatim so it agrees on every replica), catching a body
+ // bit-flip that leaves the header structurally valid. A completed entry
+ // after the corrupt one means interior bit-rot and refuses boot below;
+ // only a genuine torn tail is truncated. An unsealed entry has nothing to
+ // verify against and is skipped, not rejected: see
+ // [`CHECKSUM_BODY_UNSEALED`].
+ //
+ // TODO(wal-integrity): the header `checksum` and its `parent` chain stay
+ // unverified, since the producer does not seal them yet (blocked on
+ // re-sealing re-stamped retransmits), so a bit-flip in a
+ // structurally-valid header field slips through. Recovery derives
+ // `commit_watermark = max(header.commit)`, so a flipped `commit` makes it
+ // apply prepared-but-uncommitted ops as committed, the very ops a view
+ // change may have truncated cluster-wide, diverging this replica. Seal
+ // and verify the header checksum + parent chain.
+ if header.checksum_body == CHECKSUM_BODY_UNSEALED {
+ // Skip the body read too, so a WAL written entirely by a pre-sealing
+ // build scans without touching its payload.
+ unsealed_entries += 1;
+ } else {
+ let body_len = (entry_size - HEADER_SIZE as u64) as usize;
+ // `read_at` (read_exact_at) fills the buffer to capacity, so it must
+ // hold exactly `body_len`. A prior buffer of the same length is
+ // reused as-is; any size change replaces it, since capacity cannot
+ // shrink in place and an oversized buffer would read past the entry.
+ if body_buf.len() != body_len {
+ body_buf = vec![0u8; body_len];
+ }
+ body_buf = storage.read_at(pos + HEADER_SIZE as u64, body_buf).await?;
+ if u128::from(XxHash3_64::oneshot(&body_buf)) != header.checksum_body {
+ // The header passed the structural checks above, so `entry_size`
+ // is trustworthy. Bytes following this entry mean a later append
+ // completed after it, so this is interior bit-rot of a durable
+ // entry, NOT a torn tail: one append+fsync per entry means a torn
+ // write can only be the final entry. Truncating forward would
+ // discard the committed entries that follow, so refuse boot. A
+ // cluster repairs the entry from a peer, and a solo node keeps
+ // its WAL for manual recovery instead of silently losing
+ // committed data.
+ if pos + entry_size < file_len {
+ return Err(JournalError::Io(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "interior WAL corruption at pos {pos} (op {}, operation {:?}): \
+ prepare body checksum mismatch with {} bytes of entries \
+ following; refusing to truncate and discard the committed suffix",
+ header.op,
+ header.operation,
+ file_len - (pos + entry_size),
+ ),
+ )));
+ }
+ truncate_or_fail(&storage, pos, "prepare body checksum mismatch at tail")
+ .await?;
+ break;
+ }
+ }
+
let slot = slot_for_op(header.op, slot_count);
// Note: Regarding duplicate op in WAL. We rewrite it with whichever
@@ -394,6 +558,7 @@ impl PrepareJournal {
poisoned: OnceCell::new(),
drain_in_flight: Cell::new(false),
slot_count,
+ unsealed_entries,
})
}
@@ -413,6 +578,13 @@ impl PrepareJournal {
self.last_op.get()
}
+ /// How many entries the opening scan replayed unverified
+ /// ([`CHECKSUM_BODY_UNSEALED`]). `0` once every producer seals; the boot path
+ /// warns while it is not, so the fail-open stretch is visible to an operator.
+ pub const fn unsealed_entry_count(&self) -> u64 {
+ self.unsealed_entries
+ }
+
/// Advance the snapshot watermark. The caller must ensure `op` is
/// monotonically increasing and corresponds to a durable snapshot.
///
@@ -709,6 +881,25 @@ impl Journal<FileStorage> for PrepareJournal {
let header = *entry.header();
let slot = slot_for_op(header.op, self.slot_count);
+ // Reject a buffer with slack for the same reason as the slot-collision check
+ // below: before it reaches disk. `Message::try_from` permits `len >= size`, and
+ // `write_append` writes the WHOLE buffer, so slack would land on disk while the
+ // scan's checksum verification and its `pos += entry_size` walk both use
+ // `header.size`. The producer seals the whole buffer today, which makes an
+ // over-length entry a loud checksum failure rather than a silent one, but the
+ // seal is an integrity field and not the place to enforce framing.
+ if entry.as_slice().len() != header.size as usize {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "journal entry buffer is {} bytes but its header claims {}: \
+ the slack would be written to disk and mis-frame the scan",
+ entry.as_slice().len(),
+ header.size,
+ ),
+ ));
+ }
+
// Slot collision must be detected BEFORE `write_append + fsync`:
// a post-fsync panic would leave bytes durably on disk, and the
// recovery scan on the next boot would re-hit the same collision,
@@ -806,10 +997,31 @@ mod tests {
use iggy_binary_protocol::consensus::Operation;
use tempfile::tempdir;
+ /// An entry as the primary's `project` produces it: body checksum sealed.
fn make_prepare(op: u64, body_size: usize) -> Message<PrepareHeader> {
+ make_entry(op, body_size, |body| u128::from(XxHash3_64::oneshot(body)))
+ }
+
+ /// An entry as a pre-sealing build wrote it, or as a pre-sealing primary
+ /// still replicates it: `checksum_body == 0`.
+ fn make_unsealed_prepare(op: u64, body_size: usize) -> Message<PrepareHeader> {
+ make_entry(op, body_size, |_| CHECKSUM_BODY_UNSEALED)
+ }
+
+ fn make_entry(
+ op: u64,
+ body_size: usize,
+ seal: impl FnOnce(&[u8]) -> u128,
+ ) -> Message<PrepareHeader> {
let total_size = HEADER_SIZE + body_size;
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total_size);
+ // Recognizable pattern, then seal over it so the scan accepts the entry.
+ for (i, byte) in buffer.as_mut_slice()[HEADER_SIZE..].iter_mut().enumerate() {
+ *byte = (op as u8).wrapping_add(i as u8);
+ }
+ let checksum_body = seal(&buffer.as_slice()[HEADER_SIZE..]);
+
let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
&mut buffer.as_mut_slice()[..HEADER_SIZE],
);
@@ -817,13 +1029,144 @@ mod tests {
header.command = Command2::Prepare;
header.op = op;
header.operation = Operation::CreateStream;
+ header.checksum_body = checksum_body;
- // Fill body with recognizable pattern
- for (i, byte) in buffer.as_mut_slice()[HEADER_SIZE..].iter_mut().enumerate() {
- *byte = (op as u8).wrapping_add(i as u8);
+ Message::try_from(buffer).unwrap()
+ }
+
+ #[compio::test]
+ async fn scan_truncates_entry_with_body_checksum_mismatch() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ journal
+ .append(make_prepare(1, 64).deep_copy())
+ .await
+ .unwrap();
+ journal
+ .append(make_prepare(2, 64).deep_copy())
+ .await
+ .unwrap();
+ assert_eq!(journal.last_op(), Some(2));
}
- Message::try_from(buffer).unwrap()
+ // Flip a byte in the last entry's body, leaving its header structurally
+ // valid (command/size/op intact) so only the body checksum can catch it.
+ let mut bytes = std::fs::read(&path).unwrap();
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+
+ // Reopen: the scan recomputes the body checksum, finds the mismatch, and
+ // truncates the corrupt tail entry via the torn-tail repair.
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ assert_eq!(
+ journal.last_op(),
+ Some(1),
+ "a body-checksum mismatch on the tail entry must truncate it on scan"
+ );
+ assert!(journal.header(2).is_none());
+ }
+
+ #[compio::test]
+ async fn scan_refuses_boot_on_interior_body_checksum_mismatch() {
+ // Bit-rot in a committed entry that is NOT the tail must refuse boot:
+ // truncating forward would discard the committed entries that follow
+ // (here op 3). A torn tail, the final in-flight entry, stays truncatable.
+ const BODY: usize = 64;
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ for op in 1..=3u64 {
+ journal
+ .append(make_prepare(op, BODY).deep_copy())
+ .await
+ .unwrap();
+ }
+ assert_eq!(journal.last_op(), Some(3));
+ }
+
+ // Flip a byte inside op 2's body. Entries append in order at a fixed
+ // HEADER_SIZE + BODY stride, so op 2's body starts one full entry plus one
+ // header in.
+ let entry_size = HEADER_SIZE + BODY;
+ let op2_body_byte = entry_size + HEADER_SIZE + 5;
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[op2_body_byte] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+
+ // Reopen must refuse boot: op 3 (committed) follows the corrupt op 2, so
+ // this is interior bit-rot, not a torn tail.
+ let result = PrepareJournal::open(&path, 0).await;
+ assert!(
+ matches!(result, Err(JournalError::Io(_))),
+ "interior body-checksum mismatch must refuse boot, not truncate the committed suffix"
+ );
+ }
+
+ #[compio::test]
+ async fn scan_replays_unsealed_entries_and_counts_them() {
+ // A WAL from a pre-sealing build, or one a pre-sealing primary replicated:
+ // every entry carries `checksum_body == 0`. Verifying against that would fail
+ // every entry and brick the upgrade, so the scan replays them and reports how
+ // many it could not verify.
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ for op in 1..=3u64 {
+ journal
+ .append(make_unsealed_prepare(op, 64).deep_copy())
+ .await
+ .unwrap();
+ }
+ }
+
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ assert_eq!(
+ journal.last_op(),
+ Some(3),
+ "an unsealed WAL must boot, not be rejected as corrupt"
+ );
+ assert_eq!(journal.unsealed_entry_count(), 3);
+ }
+
+ #[compio::test]
+ async fn scan_verifies_sealed_entries_alongside_unsealed_ones() {
+ // The sentinel must exempt only the entries carrying it: a rolling upgrade
+ // leaves both kinds in one WAL, and bit-rot in a sealed one must still be
+ // caught.
+ const BODY: usize = 64;
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ // Op 1 from the old primary, ops 2 and 3 after it upgraded.
+ journal
+ .append(make_unsealed_prepare(1, BODY).deep_copy())
+ .await
+ .unwrap();
+ for op in 2..=3u64 {
+ journal
+ .append(make_prepare(op, BODY).deep_copy())
+ .await
+ .unwrap();
+ }
+ }
+
+ // Flip a byte in sealed op 2's body, one full entry plus one header in.
+ let entry_size = HEADER_SIZE + BODY;
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[entry_size + HEADER_SIZE + 5] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+
+ let result = PrepareJournal::open(&path, 0).await;
+ assert!(
+ matches!(result, Err(JournalError::Io(_))),
+ "a sealed entry must still be verified when unsealed entries precede it"
+ );
}
#[compio::test]
@@ -1058,6 +1401,82 @@ mod tests {
);
}
+ #[compio::test]
+ async fn open_refuses_when_a_complete_entry_follows_the_damage() {
+ // Bit-rot in an interior header loses that entry's boundary but leaves the
+ // entries behind it intact. Each was fsynced before its PrepareOk, so they may
+ // be quorum-committed: classifying this as a torn tail would silently discard
+ // durable data. The trailing region is a few hundred bytes here, well under
+ // MAX_ENTRY_SIZE, so only the forward probe can tell the two apart.
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ journal.append(make_prepare(1, 64)).await.unwrap();
+ journal.append(make_prepare(2, 64)).await.unwrap();
+ journal.append(make_prepare(3, 64)).await.unwrap();
+ journal.storage.fsync().await.unwrap();
+ }
+
+ let entry_2_offset = (HEADER_SIZE + 64) as u64;
+ let command_byte_offset =
+ entry_2_offset + std::mem::offset_of!(PrepareHeader, command) as u64;
+ {
+ use std::io::{Seek, SeekFrom, Write};
+ let mut file = std::fs::OpenOptions::new().write(true).open(&path).unwrap();
+ file.seek(SeekFrom::Start(command_byte_offset)).unwrap();
+ file.write_all(&[99u8]).unwrap(); // out of range for Command2
+ file.sync_all().unwrap();
+ }
+
+ let size_before = std::fs::metadata(&path).unwrap().len();
+ let error = PrepareJournal::open(&path, 0)
+ .await
+ .expect_err("damage with a complete entry behind it must refuse boot");
+ let error = format!("{error:?}");
+ assert!(
+ error.contains("a complete entry starts at pos"),
+ "the refusal must come from the forward probe, not the size heuristic: {error}"
+ );
+ assert_eq!(
+ std::fs::metadata(&path).unwrap().len(),
+ size_before,
+ "a refused scan must leave the WAL intact for repair from a peer"
+ );
+ }
+
+ #[compio::test]
+ async fn append_rejects_entry_buffer_with_slack() {
+ // `Message::try_from` permits `len >= size` and `write_append` writes the whole
+ // buffer, so slack would reach disk while the scan frames on `header.size`.
+ // Refuse before the write rather than leave a mis-framed entry durable.
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+
+ let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(HEADER_SIZE + 64);
+ let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
+ &mut buffer.as_mut_slice()[..HEADER_SIZE],
+ );
+ header.command = Command2::Prepare;
+ header.op = 1;
+ header.operation = Operation::CreateStream;
+ header.size = (HEADER_SIZE + 48) as u32; // 16 bytes of slack
+ let entry = Message::try_from(buffer).unwrap();
+
+ let error = journal
+ .append(entry)
+ .await
+ .expect_err("an over-length entry buffer must be refused");
+ assert_eq!(error.kind(), io::ErrorKind::InvalidData);
+ assert_eq!(
+ journal.storage.file_len(),
+ 0,
+ "a refused append must write nothing"
+ );
+ }
+
#[compio::test]
async fn iter_headers_from() {
let dir = tempdir().unwrap();
diff --git a/core/journal/src/superblock.rs b/core/journal/src/superblock.rs
new file mode 100644
index 000000000..9cecb318e
--- /dev/null
+++ b/core/journal/src/superblock.rs
@@ -0,0 +1,824 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Durable superblock: a small record that survives a crash intact.
+//!
+//! Persists the VSR state consensus cannot recover from its own disk
+//! (`view`, `log_view`, `commit`). The payload is opaque here: this layer
+//! makes N bytes durable and picks the latest good copy, nothing more.
+//! Consensus owns their meaning.
+//!
+//! [`PingPongSuperblock`] writes two files alternately, higher valid
+//! `sequence` wins, each replaced by writing a temp, fsyncing it, renaming it
+//! over the target, then fsyncing the directory (as `PrepareJournal` does for
+//! WAL compaction). Rename is atomic, so a torn write dies in the temp while
+//! the other file stays an intact prior generation: an update can never destroy
+//! the last good record.
+//!
+//! By that same atomicity, a slot holding bytes that fail verification is bit-rot
+//! of a once-durable record, not an interrupted write. Its `sequence` is
+//! unreadable, so it may be the NEWER generation, and reading through it would
+//! hand consensus an older `view` / `log_view` than it already acted on. Such a
+//! slot yields [`SuperblockContents::Unreadable`] and blocks writes, so the next
+//! `write` cannot stamp a fresh sequence over the evidence.
+//!
+//! The record already carries the `version`, `sequence`, and
+//! `parent_checksum` an N-copy in-place quorum variant would need, so that
+//! migration stays contained to a second `impl SuperblockStore`.
+
+// Every future here drives compio single-threaded file I/O and is `!Send` by
+// construction, like the sibling `FileStorage`/`PrepareJournal`.
+#![allow(clippy::future_not_send)]
+
+use std::cell::Cell;
+use std::hash::Hasher;
+use std::io;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+
+use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
+
+use crate::prepare_journal::TmpFileGuard;
+use twox_hash::XxHash3_64;
+
+/// Identifies a superblock file and rejects a foreign or zeroed one. "SBLK".
+const SUPERBLOCK_MAGIC: u32 = 0x5342_4C4B;
+/// On-disk record format version. Bump when the framing or payload contract
+/// changes; `read` rejects an unknown version rather than misparsing it.
+const SUPERBLOCK_VERSION: u16 = 1;
+
+/// Fixed framing ahead of the payload (28 bytes): magic, version, a reserved
+/// half-word, sequence, parent checksum, and payload length.
+const HEADER_LEN: usize = 28;
+/// Trailing `XxHash3_64` over every preceding byte.
+const CHECKSUM_LEN: usize = 8;
+/// Smallest well-formed record (empty payload).
+const MIN_RECORD_LEN: usize = HEADER_LEN + CHECKSUM_LEN;
+
+/// Ceiling on a record's payload, bounding every allocation this module makes from
+/// a length it read off disk (`PrepareJournal::MAX_ENTRY_SIZE` bounds the WAL for the
+/// same reason). The only payload today is a 58-byte [`VsrState`]; the headroom is
+/// for a payload that grows fields, not for bulk data. `read_slot` treats a longer
+/// file as corrupt WITHOUT reading it, and `build_record` refuses to write one, so a
+/// length this store could have produced is always in bounds.
+const MAX_PAYLOAD_LEN: usize = 4096;
+/// Largest record `read_slot` will read into memory.
+const MAX_RECORD_LEN: usize = HEADER_LEN + MAX_PAYLOAD_LEN + CHECKSUM_LEN;
+
+const FILE_A: &str = "superblock.a";
+const FILE_B: &str = "superblock.b";
+
+/// The two ping-pong slot file names under a superblock directory.
+///
+/// Newest wins by `sequence`. Exposed so an off-runtime reader (tests, tooling)
+/// can read the slots with blocking I/O and decode them via [`decode_slots`].
+pub const SLOT_FILE_NAMES: [&str; 2] = [FILE_A, FILE_B];
+
+/// Outcome of reading the latest record from a [`SuperblockStore`].
+///
+/// The three states stay distinct because "never written" and "written, now
+/// unreadable" demand opposite recovery responses. Collapsing them (as an
+/// `Option` would) lets a lost or corrupt superblock masquerade as a fresh
+/// deployment: the split-brain footgun this type exists to remove.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SuperblockContents {
+ /// Every slot absent or zero-length. The only state a genuinely fresh
+ /// deployment produces.
+ Empty,
+ /// The newest checksum-clean record of a supported version.
+ Present(Vec<u8>),
+ /// No slot pair this build can trust: bytes that fail verification on every
+ /// copy, or one clean record beside a copy that does not verify, whose lost
+ /// `sequence` may have been the newer. `version` names an unrecognized format
+ /// version when that was the cause (a downgrade). Never "fresh".
+ Unreadable { version: Option<u16> },
+}
+
+/// A durable store for a single small record.
+///
+/// `write` returns only once the record is durable, and a crash during it must
+/// never destroy the record a prior `write` made durable.
+pub trait SuperblockStore {
+ /// Persist `payload` as the newest record. Durable on return.
+ ///
+ /// # Errors
+ /// I/O error if the record cannot be made durable, or `InvalidInput` if
+ /// `payload` exceeds `u32::MAX`.
+ fn write(&self, payload: &[u8]) -> impl Future<Output = io::Result<()>>;
+
+ /// Read the latest record. See [`SuperblockContents`].
+ ///
+ /// # Errors
+ /// I/O error if a slot exists but cannot be read. A checksum, magic, or
+ /// version failure is NOT an I/O error: it surfaces as
+ /// [`SuperblockContents::Unreadable`] so the caller decides how to respond.
+ fn read_latest(&self) -> impl Future<Output = io::Result<SuperblockContents>>;
+}
+
+/// A boxed, lifetime-bound future, for the object-safe superblock adapter.
+type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
+
+/// Object-safe adapter over [`SuperblockStore`].
+///
+/// Lets a superblock be held as `Rc<dyn DynSuperblockStore>` without threading a
+/// generic through the consensus and metadata layers. Writes happen only on a
+/// view change or checkpoint, so the boxed future is off every hot path. The
+/// `dyn_` prefix avoids clashing with the inherent methods under the blanket impl.
+pub trait DynSuperblockStore {
+ /// See [`SuperblockStore::write`].
+ fn dyn_write<'a>(&'a self, payload: &'a [u8]) -> BoxFuture<'a, io::Result<()>>;
+
+ /// See [`SuperblockStore::read_latest`].
+ fn dyn_read_latest(&self) -> BoxFuture<'_, io::Result<SuperblockContents>>;
+}
+
+impl<T: SuperblockStore> DynSuperblockStore for T {
+ fn dyn_write<'a>(&'a self, payload: &'a [u8]) -> BoxFuture<'a, io::Result<()>> {
+ Box::pin(self.write(payload))
+ }
+
+ fn dyn_read_latest(&self) -> BoxFuture<'_, io::Result<SuperblockContents>> {
+ Box::pin(self.read_latest())
+ }
+}
+
+#[derive(Clone, Copy)]
+enum Slot {
+ A,
+ B,
+}
+
+impl Slot {
+ const fn other(self) -> Self {
+ match self {
+ Self::A => Self::B,
+ Self::B => Self::A,
+ }
+ }
+
+ const fn file_name(self) -> &'static str {
+ match self {
+ Self::A => FILE_A,
+ Self::B => FILE_B,
+ }
+ }
+}
+
+/// Two-file ping-pong superblock. See the module docs for the durability
+/// argument.
+pub struct PingPongSuperblock {
+ dir: PathBuf,
+ /// Sequence to stamp on the next `write`. Monotonic; selects the latest
+ /// record on read.
+ next_sequence: Cell<u64>,
+ /// Slot the next `write` targets: always the one NOT holding the latest
+ /// record, so an interrupted write cannot corrupt the newest good copy.
+ next_slot: Cell<Slot>,
+ /// Set at `open` when a slot held bytes that do not verify, which `write` then
+ /// refuses to overwrite (see the `write` impl). Fixed for the store's
+ /// lifetime: repair happens out of band.
+ degraded: bool,
+ /// Debug tripwire for the single-writer contract (see the `write` impl).
+ /// Set across the `.await`, so an overlapping second writer trips the assert
+ /// instead of silently tearing a slot. Absent in release builds.
+ #[cfg(debug_assertions)]
+ writing: Cell<bool>,
+}
+
+impl PingPongSuperblock {
+ /// Open the superblock rooted at `dir`, which must already exist. Classifies
+ /// both slots to resume the sequence counter and aim the next write at the
+ /// staler slot. Missing files mean a fresh store.
+ ///
+ /// Succeeds even when a slot is unusable, so the caller reads the typed verdict
+ /// from [`SuperblockStore::read_latest`] rather than a bare I/O error. Writes
+ /// are refused in that state.
+ ///
+ /// # Errors
+ /// I/O error if a slot file exists but cannot be read.
+ pub async fn open(dir: impl Into<PathBuf>) -> io::Result<Self> {
+ let dir = dir.into();
+ let slot_a = read_slot(&dir.join(FILE_A)).await?;
+ let slot_b = read_slot(&dir.join(FILE_B)).await?;
+ let seq_a = sequence_of(&slot_a);
+ let seq_b = sequence_of(&slot_b);
+
+ // Latest sequence across both slots; a slot with no readable sequence counts
+ // as 0. Sound only because `degraded` then blocks every write: such a slot
+ // may hold a higher sequence, and reusing it would break the monotonicity
+ // that selects the latest record.
+ let latest = seq_a.unwrap_or(0).max(seq_b.unwrap_or(0));
+ // Aim the next write at the slot that does NOT hold the strictly-newest
+ // record, so an interrupted write cannot clobber it. A tie or a fresh
+ // store targets A.
+ let a_is_newest = match (seq_a, seq_b) {
+ (Some(a), Some(b)) => a >= b,
+ (Some(_), None) => true,
+ (None, _) => false,
+ };
+ let next_slot = if a_is_newest { Slot::B } else { Slot::A };
+
+ Ok(Self {
+ dir,
+ next_sequence: Cell::new(latest + 1),
+ next_slot: Cell::new(next_slot),
+ degraded: has_unreadable_sequence(&slot_a) || has_unreadable_sequence(&slot_b),
+ #[cfg(debug_assertions)]
+ writing: Cell::new(false),
+ })
+ }
+}
+
+impl SuperblockStore for PingPongSuperblock {
+ async fn write(&self, payload: &[u8]) -> io::Result<()> {
+ // An unverifiable slot looks stale to the aiming logic in `open`, so this
+ // write would land on it and stamp `sequence = latest_valid + 1` over the
+ // only evidence, leaving two clean slots whose "newest" carries the OLDER
+ // payload: the regression `read_latest` refuses, laundered into durable
+ // state. Refuse until the slot is repaired out of band.
+ if self.degraded {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ "superblock slot holds bytes that do not verify; refusing to write over it",
+ ));
+ }
+ // Single-writer contract: callers MUST serialize `write` (the metadata
+ // superblock lock does). `sequence`/`slot` are read before the
+ // `atomic_replace` await and committed after it, so two overlapping
+ // writers would target the same slot and could tear it while both return
+ // `Ok`. The ping-pong guarantee, that the non-written slot is always an
+ // intact prior generation, holds only with one write in flight.
+ let sequence = self.next_sequence.get();
+ let slot = self.next_slot.get();
+ // Built before the first await so the `payload` borrow does not span it.
+ let record = build_record(sequence, payload)?;
+ // RAII, not a manual clear: a dropped write future would otherwise leave the
+ // flag latched for the store's lifetime and make every later write panic on a
+ // contract nobody violated, sending whoever debugs it to audit the lock.
+ #[cfg(debug_assertions)]
+ let _writing = WritingGuard::acquire(&self.writing);
+ atomic_replace(&self.dir, slot.file_name(), record).await?;
+ self.next_sequence.set(sequence + 1);
+ self.next_slot.set(slot.other());
+ Ok(())
+ }
+
+ async fn read_latest(&self) -> io::Result<SuperblockContents> {
+ let a = read_slot(&self.dir.join(FILE_A)).await?;
+ let b = read_slot(&self.dir.join(FILE_B)).await?;
+ Ok(combine_slots(a, b))
+ }
+}
+
+/// Holds the single-writer tripwire for one `write`, clearing it on drop so a
+/// cancelled write future cannot latch it. Debug builds only.
+#[cfg(debug_assertions)]
+struct WritingGuard<'a>(&'a Cell<bool>);
+
+#[cfg(debug_assertions)]
+impl<'a> WritingGuard<'a> {
+ fn acquire(writing: &'a Cell<bool>) -> Self {
+ assert!(
+ !writing.replace(true),
+ "PingPongSuperblock::write called concurrently; writes must be externally serialized"
+ );
+ Self(writing)
+ }
+}
+
+#[cfg(debug_assertions)]
+impl Drop for WritingGuard<'_> {
+ fn drop(&mut self) {
+ self.0.set(false);
+ }
+}
+
+fn checksum(bytes: &[u8]) -> u64 {
+ let mut hasher = XxHash3_64::new();
+ hasher.write(bytes);
+ hasher.finish()
+}
+
+fn build_record(sequence: u64, payload: &[u8]) -> io::Result<Vec<u8>> {
+ // Refuse here rather than at the next boot: a record over the read ceiling would
+ // be written durably and then classified corrupt, which blocks writes and refuses
+ // boot over a payload the caller could have been told about synchronously.
+ if payload.len() > MAX_PAYLOAD_LEN {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ format!(
+ "superblock payload is {} bytes, over the {MAX_PAYLOAD_LEN}-byte ceiling",
+ payload.len()
+ ),
+ ));
+ }
+ let payload_len = u32::try_from(payload.len())
+ .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "superblock payload too large"))?;
+
+ let mut record = Vec::with_capacity(MIN_RECORD_LEN + payload.len());
+ record.extend_from_slice(&SUPERBLOCK_MAGIC.to_le_bytes());
+ record.extend_from_slice(&SUPERBLOCK_VERSION.to_le_bytes());
+ record.extend_from_slice(&0u16.to_le_bytes()); // reserved
+ record.extend_from_slice(&sequence.to_le_bytes());
+ record.extend_from_slice(&0u64.to_le_bytes()); // parent_checksum, unused by ping-pong
+ record.extend_from_slice(&payload_len.to_le_bytes());
+ record.extend_from_slice(payload);
+ let ck = checksum(&record);
+ record.extend_from_slice(&ck.to_le_bytes());
+ Ok(record)
+}
+
+/// Reach the same verdict as [`PingPongSuperblock::read_latest`] from the two
+/// slots' raw bytes, without a runtime.
+///
+/// `slot_a` / `slot_b` are the file contents read from [`SLOT_FILE_NAMES`], or
+/// `None` for a missing slot. Shares the classification and selection rules with
+/// the async path, so an off-runtime reader (tests, tooling) cannot conclude
+/// `Present` where the server would refuse to boot.
+#[must_use]
+pub fn decode_slots(slot_a: Option<&[u8]>, slot_b: Option<&[u8]>) -> SuperblockContents {
+ combine_slots(classify_bytes(slot_a), classify_bytes(slot_b))
+}
+
+/// One slot's contents, classified. Separates "nothing here" from "something
+/// here but unusable" so [`combine_slots`] can tell `Empty` from `Unreadable`.
+enum SlotClass {
+ /// File missing or zero-length.
+ Absent,
+ /// Bytes present but unusable: too short, foreign magic, or a length/checksum
+ /// failure. `version` is `Some` when the magic matched but the format version is
+ /// unknown to this build, so nothing past that field can be trusted or
+ /// checksum-verified; [`combine_slots`] reports that case over generic corruption,
+ /// since it names a downgrade.
+ Corrupt { version: Option<u16> },
+ /// A checksum-clean record of the current version.
+ Valid { sequence: u64, payload: Vec<u8> },
+}
+
+/// Classify one slot from raw bytes, `Absent` for a missing or empty slot.
+/// Mirrors [`read_slot`]'s treatment of a zero-length file.
+fn classify_bytes(bytes: Option<&[u8]>) -> SlotClass {
+ match bytes {
+ None | Some([]) => SlotClass::Absent,
+ Some(bytes) => classify(bytes),
+ }
+}
+
+/// Classify one slot's raw, non-empty bytes: framing, version, payload length and
+/// checksum in one pass, so no slot is parsed twice.
+fn classify(bytes: &[u8]) -> SlotClass {
+ let corrupt = SlotClass::Corrupt { version: None };
+ // Too short to hold even the framing, or a foreign file: unusable bytes.
+ if bytes.len() < MIN_RECORD_LEN {
+ return corrupt;
+ }
+ if u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) != SUPERBLOCK_MAGIC {
+ return corrupt;
+ }
+ let version = u16::from_le_bytes([bytes[4], bytes[5]]);
+ if version != SUPERBLOCK_VERSION {
+ // Magic matched, so a superblock writer produced this, but the framing past
+ // the version field may differ, so neither the payload length nor the checksum
+ // can validate it.
+ return SlotClass::Corrupt {
+ version: Some(version),
+ };
+ }
+
+ let payload_len = u32::from_le_bytes([bytes[24], bytes[25], bytes[26], bytes[27]]) as usize;
+ // `decode_slots` is public and takes bytes straight from a caller, so bound the
+ // length field here too, not only via `read_slot`'s file-size ceiling.
+ if payload_len > MAX_PAYLOAD_LEN {
+ return corrupt;
+ }
+ let checksum_start = HEADER_LEN + payload_len;
+ if bytes.len() != checksum_start + CHECKSUM_LEN {
+ return corrupt;
+ }
+ let Ok(stored) = bytes[checksum_start..checksum_start + CHECKSUM_LEN].try_into() else {
+ return corrupt;
+ };
+ if checksum(&bytes[..checksum_start]) != u64::from_le_bytes(stored) {
+ return corrupt;
+ }
+
+ SlotClass::Valid {
+ sequence: u64::from_le_bytes([
+ bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
+ ]),
+ payload: bytes[HEADER_LEN..checksum_start].to_vec(),
+ }
+}
+
+/// Combine the two slots into one outcome.
+///
+/// Two valid records: higher `sequence` wins. One valid record beside an absent
+/// slot: that record, what a single write leaves behind. `Empty` only when both
+/// slots are absent, the one state a fresh deployment produces.
+///
+/// Everything else refuses boot with `Unreadable`, including a valid record beside
+/// a slot that holds bytes but does not verify. Such a slot is bit-rot, not a torn
+/// write (rename is atomic), and its `sequence` is gone, so it cannot be shown to
+/// be the older of the two. Falling back would walk `view` / `log_view` backwards
+/// and let this replica act twice in one view. An unrecognized version is reported
+/// over generic corruption, since it names a downgrade.
+///
+/// Two copies cannot do better: with no readable sequence the choice is refuse or
+/// risk regression. An N-copy quorum could repair in place instead, which the
+/// record's `sequence` and `parent_checksum` framing already allows for.
+fn combine_slots(a: SlotClass, b: SlotClass) -> SuperblockContents {
+ match (a, b) {
+ (
+ SlotClass::Valid {
+ sequence: seq_a,
+ payload: payload_a,
+ },
+ SlotClass::Valid {
+ sequence: seq_b,
+ payload: payload_b,
+ },
+ ) => SuperblockContents::Present(if seq_a >= seq_b { payload_a } else { payload_b }),
+ (SlotClass::Valid { payload, .. }, SlotClass::Absent)
+ | (SlotClass::Absent, SlotClass::Valid { payload, .. }) => {
+ SuperblockContents::Present(payload)
+ }
+ (SlotClass::Absent, SlotClass::Absent) => SuperblockContents::Empty,
+ (
+ SlotClass::Corrupt {
+ version: Some(version),
+ },
+ _,
+ )
+ | (
+ _,
+ SlotClass::Corrupt {
+ version: Some(version),
+ },
+ ) => SuperblockContents::Unreadable {
+ version: Some(version),
+ },
+ _ => SuperblockContents::Unreadable { version: None },
+ }
+}
+
+/// The slot's `sequence`, or `None` when it holds no record this build can read.
+const fn sequence_of(slot: &SlotClass) -> Option<u64> {
+ match slot {
+ SlotClass::Valid { sequence, .. } => Some(*sequence),
+ SlotClass::Absent | SlotClass::Corrupt { .. } => None,
+ }
+}
+
+/// Whether the slot holds bytes whose generation is unknowable, so it cannot be
+/// ruled out as the newer. An absent slot never held a record to lose, so it is
+/// not one of these.
+const fn has_unreadable_sequence(slot: &SlotClass) -> bool {
+ matches!(slot, SlotClass::Corrupt { .. })
+}
+
+async fn read_slot(path: &Path) -> io::Result<SlotClass> {
+ let file = match compio::fs::File::open(path).await {
+ Ok(file) => file,
+ Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(SlotClass::Absent),
+ Err(e) => return Err(e),
+ };
+ let len = file.metadata().await?.len();
+ if len == 0 {
+ return Ok(SlotClass::Absent);
+ }
+ // Bound the allocation on `st_size` before making it. A file longer than a
+ // well-formed record cannot be one: `classify` demands an exact length, so
+ // reading it would spend the allocation only to conclude "corrupt". The realistic
+ // producer is a foreign or leftover file, not bit-rot, since the value is the
+ // file's size rather than a field inside it.
+ let len = match usize::try_from(len) {
+ Ok(len) if len <= MAX_RECORD_LEN => len,
+ _ => return Ok(SlotClass::Corrupt { version: None }),
+ };
+ let buf = vec![0u8; len];
+ let (result, buf) = file.read_exact_at(buf, 0).await.into();
+ result?;
+ // A checksum/magic/version failure is not an I/O error: the other slot may
+ // still be good, so classify the bytes and let `combine_slots` decide.
+ Ok(classify(&buf))
+}
+
+/// Replace `dir/file_name` atomically: write a temp, fsync it, rename over the
+/// target, then fsync the directory so the rename itself is durable.
+async fn atomic_replace(dir: &Path, file_name: &str, bytes: Vec<u8>) -> io::Result<()> {
+ let tmp_path = dir.join(format!("{file_name}.tmp"));
+ let final_path = dir.join(file_name);
+
+ // Unlink the temp on any failure before the rename. Not a correctness matter,
+ // since `File::create` truncates and `next_sequence` / `next_slot` stay
+ // un-advanced so a retry re-targets the same slot; it keeps a failing disk from
+ // littering `superblock.{a,b}.tmp` next to the slots an operator is inspecting.
+ let guard = TmpFileGuard::new(tmp_path.clone());
+ let mut tmp = compio::fs::File::create(&tmp_path).await?;
+ let (result, _buf) = tmp.write_all_at(bytes, 0).await.into();
+ result?;
+ tmp.sync_all().await?;
+
+ compio::fs::rename(&tmp_path, &final_path).await?;
+ guard.defuse();
+
+ let dir_file = compio::fs::File::open(dir).await?;
+ dir_file.sync_all().await?;
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use tempfile::tempdir;
+
+ /// The payload when present, else `None`, so round-trip assertions can
+ /// compare against the written bytes.
+ fn present(contents: SuperblockContents) -> Option<Vec<u8>> {
+ match contents {
+ SuperblockContents::Present(payload) => Some(payload),
+ SuperblockContents::Empty | SuperblockContents::Unreadable { .. } => None,
+ }
+ }
+
+ /// Fresh store, ping-pong alternation, and sequence resume across a reopen, in
+ /// one pass: `Empty` before any write, the newer payload winning after each,
+ /// and both slot files present once two writes have landed.
+ #[compio::test]
+ async fn given_reopen_when_write_should_alternate_slots_and_advance() {
+ let dir = tempdir().unwrap();
+ {
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ assert_eq!(sb.read_latest().await.unwrap(), SuperblockContents::Empty);
+ sb.write(b"first").await.unwrap();
+ }
+ // Reopen: sequence resumes and the next write targets the staler slot.
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ assert_eq!(
+ present(sb.read_latest().await.unwrap()).as_deref(),
+ Some(&b"first"[..])
+ );
+ sb.write(b"second").await.unwrap();
+ assert_eq!(
+ present(sb.read_latest().await.unwrap()).as_deref(),
+ Some(&b"second"[..])
+ );
+ sb.write(b"third").await.unwrap();
+ assert_eq!(
+ present(sb.read_latest().await.unwrap()).as_deref(),
+ Some(&b"third"[..])
+ );
+
+ // Writes alternate slots, so both files exist.
+ assert!(dir.path().join(FILE_A).exists());
+ assert!(dir.path().join(FILE_B).exists());
+ }
+
+ #[compio::test]
+ async fn given_one_slot_corrupt_when_read_latest_should_refuse_boot() {
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ sb.write(b"first").await.unwrap(); // slot A, seq 1
+ sb.write(b"second").await.unwrap(); // slot B, seq 2 (newer)
+
+ // Corrupt the newer slot's payload byte so its checksum fails.
+ let path = dir.path().join(FILE_B);
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[HEADER_LEN] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+
+ // Falling back to slot A would serve `first`, a generation consensus already
+ // moved past. Nothing proves the corrupt slot was the older one, so refuse.
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable { version: None }
+ );
+ }
+
+ #[compio::test]
+ async fn given_corrupt_slot_when_reopen_should_refuse_write_and_keep_both_slots() {
+ // `open` aims the next write at the slot that looks stale, which is the
+ // corrupt one. Writing there would stamp `sequence = 2` over it, leaving two
+ // clean slots whose newest carries `first` and laundering the regression
+ // `read_latest` just refused. Writes stay blocked instead.
+ let dir = tempdir().unwrap();
+ {
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ sb.write(b"first").await.unwrap(); // slot A, seq 1
+ sb.write(b"second").await.unwrap(); // slot B, seq 2
+ }
+ let path_b = dir.path().join(FILE_B);
+ let mut bytes = std::fs::read(&path_b).unwrap();
+ bytes[HEADER_LEN] ^= 0xFF;
+ std::fs::write(&path_b, &bytes).unwrap();
+ let before_a = std::fs::read(dir.path().join(FILE_A)).unwrap();
+ let before_b = std::fs::read(&path_b).unwrap();
+
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable { version: None }
+ );
+ let error = sb.write(b"third").await.unwrap_err();
+ assert_eq!(error.kind(), io::ErrorKind::InvalidData);
+
+ assert_eq!(
+ std::fs::read(dir.path().join(FILE_A)).unwrap(),
+ before_a,
+ "the surviving record must be left untouched for out-of-band repair"
+ );
+ assert_eq!(
+ std::fs::read(&path_b).unwrap(),
+ before_b,
+ "the corrupt slot is the only evidence of the lost generation"
+ );
+ }
+
+ #[compio::test]
+ async fn given_both_slots_corrupt_when_read_latest_should_return_unreadable() {
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ sb.write(b"first").await.unwrap();
+ sb.write(b"second").await.unwrap();
+
+ for name in [FILE_A, FILE_B] {
+ let path = dir.path().join(name);
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[HEADER_LEN] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+ }
+
+ // Bytes on both slots, neither verifies: a torn superblock, NOT a fresh
+ // deployment. Recovery must refuse boot.
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable { version: None }
+ );
+ }
+
+ #[compio::test]
+ async fn given_unsupported_version_when_read_latest_should_report_version() {
+ // Magic matches but the format version is unknown: a downgrade, or a
+ // corrupt version field. Version is checked before the checksum, so
+ // patching it alone suffices.
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ sb.write(b"payload").await.unwrap(); // slot A, current version
+
+ let path = dir.path().join(FILE_A);
+ let mut bytes = std::fs::read(&path).unwrap();
+ let bogus = SUPERBLOCK_VERSION + 1;
+ bytes[4..6].copy_from_slice(&bogus.to_le_bytes());
+ std::fs::write(&path, &bytes).unwrap();
+
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable {
+ version: Some(bogus)
+ }
+ );
+
+ // A clean record beside it does not rescue the read: this build cannot
+ // sequence the unrecognized slot, so that slot may be the newer generation.
+ sb.write(b"newer").await.unwrap(); // slot B, current version
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable {
+ version: Some(bogus)
+ }
+ );
+ }
+
+ #[compio::test]
+ async fn given_two_writes_when_decode_slots_from_disk_should_match_read_latest() {
+ // The off-runtime decoder must reach the same verdict the async read path
+ // does: newer sequence wins across the two slots.
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ sb.write(b"first").await.unwrap();
+ sb.write(b"second").await.unwrap();
+
+ let read_slots = || {
+ let slot_a = std::fs::read(dir.path().join(SLOT_FILE_NAMES[0])).ok();
+ let slot_b = std::fs::read(dir.path().join(SLOT_FILE_NAMES[1])).ok();
+ decode_slots(slot_a.as_deref(), slot_b.as_deref())
+ };
+ assert_eq!(read_slots(), sb.read_latest().await.unwrap());
+ assert_eq!(
+ read_slots(),
+ SuperblockContents::Present(b"second".to_vec())
+ );
+
+ // And it must refuse where the server would, not read through a corrupt slot
+ // to the older record.
+ let path = dir.path().join(SLOT_FILE_NAMES[1]);
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[HEADER_LEN] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+
+ assert_eq!(read_slots(), sb.read_latest().await.unwrap());
+ assert_eq!(
+ read_slots(),
+ SuperblockContents::Unreadable { version: None }
+ );
+ }
+
+ // Two `write`s interleaving across the `atomic_replace` await read the same
+ // `(sequence, slot)` and would tear that slot while both report success.
+ // Production serializes writes behind the metadata superblock lock; this
+ // proves the debug tripwire catches a bypass instead of corrupting a slot.
+ // Compiled out of release builds.
+ #[cfg(debug_assertions)]
+ #[compio::test]
+ #[should_panic(expected = "called concurrently")]
+ async fn given_overlapping_writes_when_write_should_trip_single_writer_guard() {
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ // Polling both together parks the first on its `atomic_replace` await
+ // holding the writer flag, so the second trips the assert on the same poll.
+ let (first, second) = futures::future::join(sb.write(b"first"), sb.write(b"second")).await;
+ let _ = (first, second);
+ }
+
+ // A dropped write future must not latch the tripwire: the flag is a
+ // single-writer contract check, and leaving it set would make every later write
+ // panic on a violation that never happened.
+ #[cfg(debug_assertions)]
+ #[compio::test]
+ async fn given_dropped_write_future_when_writing_again_should_not_trip_guard() {
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ {
+ let mut write = Box::pin(sb.write(b"cancelled"));
+ // One poll parks it inside `atomic_replace` with the flag held, then the
+ // future is dropped without ever completing.
+ assert!(
+ futures::poll!(&mut write).is_pending(),
+ "the write must park on file I/O for this to model a cancellation"
+ );
+ }
+
+ sb.write(b"after").await.unwrap();
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Present(b"after".to_vec())
+ );
+ }
+
+ #[compio::test]
+ async fn given_oversized_payload_when_write_should_refuse() {
+ // Refused synchronously rather than written and then classified corrupt on the
+ // next boot, which would block writes and refuse boot over a payload the
+ // caller could have been told about here.
+ let dir = tempdir().unwrap();
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ let error = sb
+ .write(&vec![0u8; MAX_PAYLOAD_LEN + 1])
+ .await
+ .expect_err("a payload over the ceiling must be refused");
+ assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Empty,
+ "a refused write must leave the store untouched"
+ );
+ }
+
+ #[compio::test]
+ async fn given_oversized_slot_file_when_read_latest_should_refuse_without_reading_it() {
+ // A file longer than a well-formed record cannot be one (`classify` demands
+ // an exact length), so it is classified from its size alone. The verdict must
+ // still be the fail-closed one: a foreign file in a slot is not a fresh
+ // deployment.
+ let dir = tempdir().unwrap();
+ let path = dir.path().join(SLOT_FILE_NAMES[0]);
+ let mut bytes = Vec::with_capacity(MAX_RECORD_LEN + 1);
+ bytes.extend_from_slice(&SUPERBLOCK_MAGIC.to_le_bytes());
+ bytes.resize(MAX_RECORD_LEN + 1, 0);
+ std::fs::write(&path, &bytes).unwrap();
+
+ let sb = PingPongSuperblock::open(dir.path()).await.unwrap();
+ assert_eq!(
+ sb.read_latest().await.unwrap(),
+ SuperblockContents::Unreadable { version: None }
+ );
+ assert!(
+ sb.write(b"payload").await.is_err(),
+ "a slot whose generation is unknowable must block writes"
+ );
+ }
+}
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index cc2d263db..762e3c7e1 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -23,15 +23,15 @@ use crate::stm::stream::{Streams, TruncatePartitionRequest};
use crate::stm::user::{DeletePersonalAccessTokenRequest, Users};
use crate::stm::{ConsensusGroupAllocator, StateMachine};
use consensus::{
- CLIENTS_TABLE_MAX, Canceled, ClientTable, CommitLogEvent, CommitReply, Consensus,
- EvictionContext, Pipeline, PipelineEntry, Plane, PlaneIdentity, PlaneKind, PreflightOutcome,
- Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus,
- ack_preflight, ack_quorum_reached, apply_preflight_consensus_plane, build_eviction_message,
- build_reply_message, build_reply_message_with, build_result_rejection_reply, emit_sim_event,
- fence_old_prepare_by_commit, is_caught_up_primary,
- panic_if_hash_chain_would_break_in_same_view, peek_committable_head, pipeline_prepare_common,
- register_preflight, replicate_preflight, replicate_to_next_in_chain, request_preflight,
- send_eviction_to_client, send_prepare_ok as send_prepare_ok_common,
+ CLIENTS_TABLE_MAX, Canceled, ClientTable, ClientTableSnapshot, CommitLogEvent, CommitReply,
+ Consensus, EvictionContext, Pipeline, PipelineEntry, Plane, PlaneIdentity, PlaneKind,
+ PreflightOutcome, Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind,
+ VsrConsensus, ack_preflight, ack_quorum_reached, apply_preflight_consensus_plane,
+ build_eviction_message, build_reply_message, build_reply_message_with,
+ build_result_rejection_reply, emit_sim_event, fence_old_prepare_by_commit,
+ is_caught_up_primary, panic_if_hash_chain_would_break_in_same_view, peek_committable_head,
+ pipeline_prepare_common, register_preflight, replicate_preflight, replicate_to_next_in_chain,
+ request_preflight, send_eviction_to_client, send_prepare_ok as send_prepare_ok_common,
};
use iggy_binary_protocol::WireIdentifier;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
@@ -46,7 +46,9 @@ use iggy_binary_protocol::{
};
use iggy_common::IggyError;
use iggy_common::UserId;
+use iggy_common::calculate_checksum;
use iggy_common::variadic;
+use journal::superblock::DynSuperblockStore;
use journal::{Journal, JournalHandle};
use message_bus::MessageBus;
use server_common::Message;
@@ -100,24 +102,53 @@ impl IggySnapshot {
&self.snapshot
}
+ /// Mutable view for tests that need to fold state into a checkpoint by hand,
+ /// the way [`Self::persist_snapshot`] does on the live path.
+ #[cfg(test)]
+ pub(crate) const fn snapshot_mut(&mut self) -> &mut MetadataSnapshot {
+ &mut self.snapshot
+ }
+
/// Persist the snapshot to disk.
///
/// # Errors
/// Returns `SnapshotError` if serialization or I/O fails.
pub fn persist(&self, path: &Path) -> Result<(), SnapshotError> {
+ Self::write_durably(path, &self.encode()?)
+ }
+
+ /// Write already-encoded snapshot bytes to `path` durably: temp, fsync,
+ /// rename, parent-dir fsync, with the integrity trailer appended.
+ ///
+ /// Split from [`Self::persist`] so the checkpoint path can hash and write one
+ /// buffer instead of encoding the whole snapshot twice under the durability
+ /// lock (the client table is folded in, so a second encode is a full
+ /// re-serialization).
+ ///
+ /// # Errors
+ /// `SnapshotError::Persist` if any write, fsync, or rename fails.
+ fn write_durably(path: &Path, encoded: &[u8]) -> Result<(), SnapshotError> {
use crate::stm::snapshot::PersistStage;
use std::fs;
use std::io::Write;
- let encoded = self.encode()?;
-
let tmp_path = path.with_extension("bin.tmp");
let mut file = fs::File::create(&tmp_path).map_err(|e| SnapshotError::Persist {
stage: PersistStage::Write,
source: e,
})?;
- file.write_all(&encoded)
+ file.write_all(encoded)
+ .map_err(|e| SnapshotError::Persist {
+ stage: PersistStage::Write,
+ source: e,
+ })?;
+ // Self-verifying trailer. The superblock's checkpoint pairing cannot stand in:
+ // phase 1 of a checkpoint renames the new snapshot over `snapshot.bin`, so a
+ // crash before the pairing write is the NORMAL crash-inside-a-checkpoint
+ // outcome, and it recovers through the `checkpoint_op < snapshot_op` arm, which
+ // accepts the snapshot with nothing to check it against.
+ file.write_all(&snapshot_trailer(encoded))
.map_err(|e| SnapshotError::Persist {
stage: PersistStage::Write,
source: e,
@@ -148,18 +179,101 @@ impl IggySnapshot {
Ok(())
}
- /// Load a snapshot from disk.
+ /// Load a snapshot from disk, with the [`checkpoint_checksum`] of the exact bytes
+ /// read.
+ ///
+ /// The checksum comes from the file's bytes, never from re-encoding what was
+ /// decoded: the pairing must survive a schema change. Adding a trailing
+ /// `#[serde(default)]` field is the repo's forward-compatible migration, and an
+ /// older file re-encodes with one MORE msgpack array element after it, so a
+ /// re-encode checksum would diverge on the first boot of the new build and refuse
+ /// every checkpointed node with its WAL prefix already drained.
///
/// # Errors
- /// Returns `SnapshotError` if the file cannot be read or deserialization fails.
- pub fn load(path: &Path) -> Result<Self, SnapshotError> {
+ /// `SnapshotError::ChecksumMismatch` if the file carries an integrity trailer that
+ /// does not match its payload, or `SnapshotError` if the file cannot be read or
+ /// deserialized.
+ pub fn load(path: &Path) -> Result<(Self, u128), SnapshotError> {
let data = std::fs::read(path)?;
+ let (payload, checksum) = split_trailer(&data, path)?;
+ Ok((Self::decode(payload)?, checksum))
+ }
+}
- // TODO: when checksum is added we need to check
- // if data.len() is atleast the size of checksum
+/// First retry delay after a failed superblock write, doubling per consecutive
+/// failure. Starts near the 10 ms consensus tick, so a one-off failure costs no
+/// latency, and a persistent one stops re-running `atomic_replace` per tick.
+const SUPERBLOCK_RETRY_BACKOFF_BASE_MICROS: u64 = 10_000;
+/// Ceiling on the retry delay. A replica fenced this long is not coming back without
+/// operator action, but the retry must stay frequent enough to recover on its own the
+/// moment the disk does.
+const SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS: u64 = 1_000_000;
+/// Caps the doubling so the shift cannot overflow before the ceiling clamps it.
+const SUPERBLOCK_RETRY_BACKOFF_MAX_SHIFT: u64 = 8;
+
+/// Framing marker for the snapshot integrity trailer, "ISNP". Distinguishes a sealed
+/// snapshot from one written before the trailer existed, so a MISSING trailer can be
+/// accepted (unverified, loudly) while a PRESENT but mismatching one refuses boot. A
+/// bare checksum could not tell those apart, and guessing wrong in either direction is
+/// unacceptable: silently accepting corruption, or bricking a healthy node.
+const SNAPSHOT_TRAILER_MAGIC: u32 = 0x4953_4E50;
+
+/// `magic` + the payload's [`checkpoint_checksum`].
+const SNAPSHOT_TRAILER_LEN: usize = size_of::<u32>() + size_of::<u128>();
+
+/// The integrity trailer for an encoded snapshot.
+fn snapshot_trailer(encoded: &[u8]) -> [u8; SNAPSHOT_TRAILER_LEN] {
+ let mut trailer = [0u8; SNAPSHOT_TRAILER_LEN];
+ trailer[..4].copy_from_slice(&SNAPSHOT_TRAILER_MAGIC.to_le_bytes());
+ trailer[4..].copy_from_slice(&checkpoint_checksum(encoded).to_le_bytes());
+ trailer
+}
- Self::decode(data.as_slice())
+/// Split a snapshot file into `(payload, checkpoint_checksum)`, verifying the trailer
+/// when one is present.
+///
+/// A file without the trailer is a snapshot written before sealing: its whole contents
+/// are the payload and the checksum is computed over them, exactly as the pairing
+/// recorded it, so an upgrade boots. It replays unverified, which is the same trade the
+/// WAL makes for entries no producer sealed, so it is warned about rather than
+/// silently accepted.
+fn split_trailer<'a>(data: &'a [u8], path: &Path) -> Result<(&'a [u8], u128), SnapshotError> {
+ let sealed = data.len() >= SNAPSHOT_TRAILER_LEN
+ && data[data.len() - SNAPSHOT_TRAILER_LEN..][..4] == SNAPSHOT_TRAILER_MAGIC.to_le_bytes();
+ if !sealed {
+ tracing::warn!(
+ path = %path.display(),
+ "metadata snapshot carries no integrity trailer; restored unverified \
+ (written before snapshot sealing)"
+ );
+ return Ok((data, checkpoint_checksum(data)));
}
+
+ let (payload, trailer) = data.split_at(data.len() - SNAPSHOT_TRAILER_LEN);
+ let expected = u128::from_le_bytes(
+ trailer[4..]
+ .try_into()
+ .expect("a sealed trailer holds 16 checksum bytes"),
+ );
+ let actual = checkpoint_checksum(payload);
+ if actual != expected {
+ return Err(SnapshotError::ChecksumMismatch { expected, actual });
+ }
+ Ok((payload, expected))
+}
+
+/// The superblock's `checkpoint_checksum` over a snapshot's on-disk bytes: the same
+/// `XxHash3_64` the WAL and superblock use, widened to the `u128` the durable record
+/// reserves.
+///
+/// Both sides of the pairing hash BYTES rather than a state: the checkpoint hashes
+/// what it wrote (`SnapshotCoordinator::persist_snapshot`), recovery hashes what it
+/// read ([`IggySnapshot::load`]). Nothing re-serializes a decoded snapshot, so the
+/// cross-check does not depend on decode-then-encode being byte-identical across
+/// builds.
+#[must_use]
+pub fn checkpoint_checksum(encoded: &[u8]) -> u128 {
+ u128::from(calculate_checksum(encoded))
}
impl Snapshot for IggySnapshot {
@@ -200,7 +314,11 @@ impl Snapshot for IggySnapshot {
/// Coordinates snapshot creation, persistence, and WAL compaction.
///
-/// Owns the data directory path and the snapshot creation function.
+/// Owns the data directory path and the snapshot creation function. The
+/// three-phase checkpoint (persist snapshot, record the pairing durably, drain the
+/// WAL) is orchestrated one layer up in [`IggyMetadata::checkpoint_if_needed`], so
+/// the superblock write can sit between persist and drain; this type owns only the
+/// snapshot I/O and the last-checkpoint bookkeeping.
pub struct SnapshotCoordinator<M> {
data_dir: std::path::PathBuf,
create_snapshot: fn(&M, u64, u64) -> Result<IggySnapshot, SnapshotError>,
@@ -209,6 +327,10 @@ pub struct SnapshotCoordinator<M> {
/// least the configured prepare-queue depth (see the static assert and
/// [`Self::set_checkpoint_margin`]).
checkpoint_margin: Cell<usize>,
+ /// `(checkpoint_op, checkpoint_checksum)` of the last snapshot persisted or
+ /// recovered at boot, `(0, 0)` when none. A view-change superblock write reads
+ /// this so it records the current pairing instead of regressing it to zero.
+ last_checkpoint: Cell<(u64, u128)>,
}
impl<M> SnapshotCoordinator<M> {
@@ -227,6 +349,7 @@ impl<M> SnapshotCoordinator<M> {
data_dir,
create_snapshot,
checkpoint_margin: Cell::new(Self::CHECKPOINT_MARGIN),
+ last_checkpoint: Cell::new((0, 0)),
}
}
@@ -239,63 +362,71 @@ impl<M> SnapshotCoordinator<M> {
.set(margin.max(Self::CHECKPOINT_MARGIN));
}
- /// Create a snapshot, persist it, and drain snapshotted entries from the
- /// journal to reclaim WAL space.
- ///
- /// # Errors
- /// Returns `SnapshotError` if snapshotting, persistence, or drain fails.
- #[allow(clippy::future_not_send)]
- pub async fn checkpoint<J>(
+ /// The last persisted checkpoint's `(op, checksum)`, `(0, 0)` when none.
+ const fn last_checkpoint(&self) -> (u64, u128) {
+ self.last_checkpoint.get()
+ }
+
+ /// Seed the last-checkpoint pairing at boot from the recovered snapshot, so the
+ /// first post-boot view-change superblock write records the real pairing rather
+ /// than `(0, 0)`.
+ fn seed_last_checkpoint(&self, checkpoint_op: u64, checkpoint_checksum: u128) {
+ self.last_checkpoint
+ .set((checkpoint_op, checkpoint_checksum));
+ }
+
+ /// Whether the journal is low enough on capacity to force a checkpoint. Gates
+ /// on the configurable margin, which bootstrap raises to at least the
+ /// prepare-queue depth; default [`Self::CHECKPOINT_MARGIN`].
+ fn should_checkpoint<J: JournalHandle>(&self, journal: &J) -> bool {
+ journal
+ .handle()
+ .remaining_capacity()
+ .is_some_and(|c| c <= self.checkpoint_margin.get())
+ }
+
+ /// Create and durably persist a snapshot at `commit_op`, record the pairing, and
+ /// return its checksum. Does NOT drain the WAL: the caller must durably record
+ /// the pairing in the superblock first, so a crash between persist and drain
+ /// recovers a consistent checkpoint with the WAL intact. Synchronous, since
+ /// snapshot creation and `std::fs` persistence never await.
+ fn persist_snapshot(
&self,
stm: &M,
- journal: &J,
- last_op: u64,
+ commit_op: u64,
created_at: u64,
- ) -> Result<(), SnapshotError>
- where
- J: JournalHandle,
- {
- let snapshot = (self.create_snapshot)(stm, last_op, created_at)?;
+ client_table: Option<ClientTableSnapshot>,
+ ) -> Result<u128, SnapshotError> {
+ let mut snapshot = (self.create_snapshot)(stm, commit_op, created_at)?;
+ // Fold in the client table, which `create_snapshot` does not see since it is
+ // not a state machine. Recovery restores it as the reconstruction floor for
+ // the drained WAL prefix.
+ snapshot.snapshot.client_table = client_table;
+ // Encode once: the checksum and the on-disk record share one buffer, so the
+ // snapshot is not serialized twice while the checkpoint lock freezes this
+ // core, and the pairing is provably over the bytes that reach the file.
+ let encoded = snapshot.encode()?;
+ let checksum = checkpoint_checksum(&encoded);
let path = self.data_dir.join(super::METADATA_DIR).join("snapshot.bin");
- snapshot.persist(&path)?;
-
- let _ = journal
- .handle()
- .drain(0..=last_op)
- .await
- .map_err(SnapshotError::Io)?;
-
- Ok(())
+ IggySnapshot::write_durably(&path, &encoded)?;
+ self.last_checkpoint.set((commit_op, checksum));
+ Ok(checksum)
}
- /// Force a checkpoint if the journal is running low on capacity.
- ///
- /// Returns `Ok(true)` if a checkpoint was taken, `Ok(false)` if not needed.
- ///
- /// # Errors
- /// Returns `SnapshotError` if the checkpoint fails.
+ /// Drain the snapshotted prefix `0..=last_op` to reclaim WAL space. Runs only
+ /// after the pairing is durable (see [`Self::persist_snapshot`]).
#[allow(clippy::future_not_send)]
- pub async fn checkpoint_if_needed<J>(
+ async fn drain<J: JournalHandle>(
&self,
- stm: &M,
journal: &J,
- commit_op: u64,
- created_at: u64,
- ) -> Result<bool, SnapshotError>
- where
- J: JournalHandle,
- {
- let needs_checkpoint = journal
+ last_op: u64,
+ ) -> Result<(), SnapshotError> {
+ journal
.handle()
- .remaining_capacity()
- .is_some_and(|c| c <= self.checkpoint_margin.get());
-
- if needs_checkpoint {
- self.checkpoint(stm, journal, commit_op, created_at).await?;
- Ok(true)
- } else {
- Ok(false)
- }
+ .drain(0..=last_op)
+ .await
+ .map_err(SnapshotError::Io)?;
+ Ok(())
}
}
@@ -487,6 +618,72 @@ fn require_shard_zero<'a, T>(
slot
}
+/// Apply one committed prepare to the state machine and client table.
+///
+/// The backup commit walk's per-op logic, shared so the simulator's WAL
+/// reconstruction reaches identical state from the same log through one apply path.
+/// Register creates or rebinds a session (no state-machine op); Logout drops the
+/// session and rebalances consumer groups; every other op applies to the state
+/// machine and caches the reply for at-most-once dedup. `fire_notifier` runs the
+/// post-commit hook (a no-op during reconstruction, before it is wired). Does not
+/// advance `commit_min`; the caller owns that counter.
+///
+/// # Panics
+/// If a committed op fails to apply, which is a decode/corruption bug, since a
+/// business rejection commits as a no-op rather than erroring: the committed log
+/// must apply cleanly on every replica.
+pub fn apply_committed_prepare<M>(
+ mux_stm: &M,
+ client_table: &RefCell<ClientTable>,
+ fire_notifier: impl Fn(Operation),
+ prepare: Message<PrepareHeader>,
+) where
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = crate::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >,
+{
+ let header = *prepare.header();
+ if header.operation == Operation::Register {
+ // Register: commit_register creates the session, no state-machine op.
+ let reply = build_reply_message(&header, &bytes::Bytes::new());
+ client_table
+ .borrow_mut()
+ .commit_register(header.client, header.user_id, reply);
+ return;
+ }
+ if header.operation == Operation::Logout {
+ client_table.borrow_mut().remove_client(header.client);
+ // Drop the disconnected client from every consumer group it joined and
+ // rebalance, Logout's only state-machine effect.
+ mux_stm.streams().remove_consumer_group_member(
+ header.client,
+ iggy_common::IggyTimestamp::from(header.timestamp),
+ );
+ return;
+ }
+ // Normal op: apply, build the reply. `Err` is decode/corruption only; a
+ // business rejection commits as a deterministic no-op whose code rides
+ // the reply body, replayed on retry.
+ let apply = gated_apply(mux_stm, prepare).unwrap_or_else(|err| {
+ panic!(
+ "apply_committed_prepare: committed metadata op={} failed to apply: {err}",
+ header.op
+ );
+ });
+ fire_notifier(header.operation);
+ let reply = build_reply_message_with(&header, apply.reply_body_len(), |dst| {
+ apply.write_reply_body(dst);
+ });
+ // Best-effort cache; a WAL replay may carry a reply for a later-evicted
+ // client, and replica-local eviction makes a stale-request replay
+ // reachable. Both are skips, not faults.
+ let outcome = client_table.borrow_mut().commit_reply(header.client, reply);
+ log_commit_reply_outcome(outcome, header.client, header.op);
+}
+
/// Late-bound callback invoked after every committed op on shard 0's metadata
/// commit path (via `gated_apply`, including a gated no-op).
///
@@ -514,6 +711,46 @@ pub struct IggyMetadata<C, J, S, M> {
pub journal: Option<J>,
/// `Some` on shard 0, `None` on other shards.
pub snapshot: Option<S>,
+ /// Durable VSR-state record (`view`/`log_view`/`commit`). `Some` only on the
+ /// shard owning metadata consensus (shard 0); `None` on peer shards and in the
+ /// partition plane, which is not yet durable. Behind `Rc<dyn ...>` so the
+ /// simulator harness can keep a clone outliving a replica across a restart.
+ pub superblock: Option<Rc<dyn DynSuperblockStore>>,
+ /// Serializes superblock writes on shard 0 so at most one is in flight.
+ /// View-change persists ([`Self::persist_superblock_if_needed`]) and checkpoints
+ /// ([`Self::checkpoint_if_needed`]) share the one ping-pong superblock, and
+ /// in-process metadata submits each run on their own spawned task, so both can
+ /// reach a write concurrently. `PingPongSuperblock::write` picks its slot before
+ /// it awaits, so two overlapping writers would target the same slot and could
+ /// tear it.
+ ///
+ /// Scoped to the write itself, NOT to a whole checkpoint. A pending view persist
+ /// blocks every gated send behind it, including the ack path's
+ /// `send_prepare_ok`, so it must not also wait out a checkpoint's snapshot
+ /// encode, two `std::fs` fsyncs and an async WAL drain. Whoever writes builds
+ /// its `VsrState` inside this section with no await in between, so the last
+ /// writer carries the freshest view and the durable view cannot regress.
+ ///
+ /// Held across the write `.await`, so it uses the same single-threaded,
+ /// cancel-safe [`LocalGate`] as `journal_gate`, a `Cell` flag with no atomics:
+ /// this shard is never `Sync`, so a `tokio::sync::Mutex` would only add an
+ /// atomic RMW per gated send for exclusion the `Cell` already provides.
+ superblock_lock: LocalGate,
+ /// Serializes whole checkpoints against each other: `persist_snapshot` renames
+ /// over the single `snapshot.bin` and `drain` rewrites the WAL through a shared
+ /// `wal.tmp`, so two concurrent checkpoints would race both. Distinct from
+ /// [`Self::superblock_lock`], which a checkpoint takes only for its own pairing
+ /// write. A checkpoint holds this one and then acquires that one; nothing takes
+ /// them in the other order.
+ checkpoint_lock: LocalGate,
+ /// Consecutive failed superblock writes, and the clock reading after which the
+ /// next attempt may run. A persistent `ENOSPC` / `EIO` would otherwise re-run a
+ /// full `atomic_replace` (create, write, fsync, rename, dir fsync) on every 10 ms
+ /// consensus tick, on the executor that also serves partition traffic. Reset on
+ /// the first success. See [`Self::persist_superblock_if_needed`] for the terminal
+ /// policy.
+ superblock_write_failures: Cell<u64>,
+ superblock_retry_after_micros: Cell<u64>,
/// State machine - lives on all shards
pub mux_stm: M,
pub allocator: ConsensusGroupAllocator,
@@ -558,6 +795,7 @@ where
consensus: Option<C>,
journal: Option<J>,
snapshot: Option<S>,
+ superblock: Option<Rc<dyn DynSuperblockStore>>,
mux_stm: M,
data_dir: Option<std::path::PathBuf>,
) -> Self {
@@ -568,6 +806,11 @@ where
consensus,
journal,
snapshot,
+ superblock,
+ superblock_lock: LocalGate::new(),
+ checkpoint_lock: LocalGate::new(),
+ superblock_write_failures: Cell::new(0),
+ superblock_retry_after_micros: Cell::new(0),
mux_stm,
allocator,
coordinator,
@@ -588,6 +831,17 @@ impl<C, J, S, M> IggyMetadata<C, J, S, M> {
*self.commit_notifier.borrow_mut() = notifier;
}
+ /// Seed the coordinator's last-checkpoint pairing at boot from the recovered
+ /// snapshot, so the first post-boot view-change superblock write records the real
+ /// `(checkpoint_op, checksum)` instead of `(0, 0)`. No-op without a coordinator
+ /// (peer shards, the simulator). Server-ng bootstrap calls this on shard 0 after
+ /// cross-checking the pairing.
+ pub fn seed_checkpoint_ref(&self, checkpoint_op: u64, checkpoint_checksum: u128) {
+ if let Some(coordinator) = &self.coordinator {
+ coordinator.seed_last_checkpoint(checkpoint_op, checkpoint_checksum);
+ }
+ }
+
/// Install the client table rebuilt by WAL-replay recovery
/// ([`crate::impls::recovery::recover`]). Boot-time only, on the owning
/// shard, before it serves traffic - replacing a live table would drop
@@ -2202,6 +2456,143 @@ where
}
}
+impl<B, P, J, S, M> IggyMetadata<VsrConsensus<B, P>, J, S, M>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ /// Persist the current VSR state to the superblock when the view changed since
+ /// the last write. The split-brain gate: callers MUST invoke this before
+ /// dispatching any view-scoped VSR message, so a replica that acted in a view can
+ /// never recover an older one after a crash.
+ ///
+ /// It fences the SEND, not the ACT. By the time a caller reaches here the handler
+ /// has already moved `view`, `log_view`, `status`, the sequencer and the pipeline,
+ /// and `commit_journal` runs outside the gate, so a failed persist still applies
+ /// committed ops locally. That is the VSR fence and it is sufficient: local state a
+ /// crash forgets is state no peer ever saw, whereas an externalized view must be
+ /// recoverable. Do not read this as "nothing changed until the write lands".
+ ///
+ /// `true` when the send may proceed, either because the state is now durable or
+ /// because there was nothing to persist (peer shard, partition plane, or an
+ /// unchanged view). `false` only when a write was attempted and failed, and the
+ /// caller must withhold the send. The in-memory view stays ahead of the durable
+ /// one, which a crash safely rolls back, and the next tick retries.
+ ///
+ /// Kept on a `B`/`P`-only impl, with no journal/snapshot/state-machine bounds, so
+ /// every VSR dispatch site can gate on it regardless of its own bounds.
+ #[allow(clippy::future_not_send)]
+ pub async fn persist_superblock_if_needed(&self, consensus: &VsrConsensus<B, P>) -> bool {
+ let Some(superblock) = self.superblock.as_ref() else {
+ return true;
+ };
+ // Lock-free fast path: the steady state is an unchanged view with nothing to
+ // write, and skipping the lock keeps every gated send off it, notably
+ // `send_prepare_ok`, which runs this per metadata prepare. Safe because
+ // `view`/`log_view` advance only on this single-threaded executor and no
+ // `.await` sits between the `Cell` read and the return, so the value cannot
+ // change under us; a concurrent advance is caught by the re-check below.
+ if !consensus.needs_superblock_persist() {
+ return true;
+ }
+ // A write that keeps failing must not re-run a full `atomic_replace` on every
+ // 10 ms tick. Back off first, while still reporting `false` so the send stays
+ // withheld: fail-closed is the point of this gate, and the backoff only bounds
+ // what the retry costs.
+ if consensus.clock_realtime_micros() < self.superblock_retry_after_micros.get() {
+ return false;
+ }
+ // Serialize superblock writes on this shard: view-change persists here and
+ // checkpoints share the one ping-pong superblock, whose `write` picks its slot
+ // before it awaits, so two overlapping writers would target the same slot and
+ // could tear it while both report success. Re-check needs-persist AFTER
+ // acquiring the lock so check and write are atomic and a redundant caller
+ // coalesces, finding the state already made durable by the writer it queued
+ // behind.
+ let _superblock = self.superblock_lock.acquire().await;
+ if !consensus.needs_superblock_persist() {
+ return true;
+ }
+ self.write_superblock(consensus, superblock.as_ref()).await
+ }
+
+ /// Write the current VSR state, paired with the last durable checkpoint, under
+ /// [`Self::superblock_lock`].
+ ///
+ /// The caller must hold that lock. The state is captured HERE rather than passed
+ /// in: with writes serialized and no await between the capture and the write, the
+ /// last writer carries the freshest view, so the durable view cannot regress even
+ /// when a checkpoint and a view change interleave. See `mark_superblock_durable`
+ /// for why the written values, not a re-read, mark durability.
+ ///
+ /// # Terminal policy
+ /// There is none beyond staying fenced: a replica that cannot record the view it
+ /// is in must not act in it, so it withholds every view-scoped send, goes quiet,
+ /// and its peers elect around it. Failures are counted and the retry interval backs
+ /// off to [`SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS`], with the error logged on the
+ /// first failure and then at each backoff step rather than per tick.
+ ///
+ /// TODO(fail-stop): a replica wedged here is dead weight an operator has to notice
+ /// from logs. Fail-stopping the process is the TigerBeetle-style answer, but this
+ /// layer holds no process-lifecycle handle; wire it where the shard owns shutdown.
+ #[allow(clippy::future_not_send)]
+ async fn write_superblock(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ superblock: &dyn DynSuperblockStore,
+ ) -> bool {
+ // Carry the last durable pairing forward so a view-change write never
+ // regresses the `(checkpoint_op, checksum)` a checkpoint recorded. `(0, 0)`
+ // with no checkpoint taken, or no coordinator (peer shards, the simulator).
+ let (checkpoint_op, checkpoint_checksum) = self
+ .coordinator
+ .as_ref()
+ .map_or((0, 0), SnapshotCoordinator::last_checkpoint);
+ let state = consensus.vsr_state(checkpoint_op, checkpoint_checksum);
+ match superblock.dyn_write(&state.to_bytes()).await {
+ Ok(()) => {
+ consensus.mark_superblock_durable(state.view, state.log_view);
+ self.superblock_write_failures.set(0);
+ self.superblock_retry_after_micros.set(0);
+ true
+ }
+ Err(error) => {
+ let failures = self.superblock_write_failures.get() + 1;
+ self.superblock_write_failures.set(failures);
+ let backoff = SUPERBLOCK_RETRY_BACKOFF_BASE_MICROS
+ .saturating_mul(1 << failures.min(SUPERBLOCK_RETRY_BACKOFF_MAX_SHIFT))
+ .min(SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS);
+ self.superblock_retry_after_micros
+ .set(consensus.clock_realtime_micros() + backoff);
+ // Rate-limited to the backoff steps: the tick would otherwise emit this
+ // every 10 ms for as long as the disk stays broken.
+ if failures.is_power_of_two() {
+ tracing::error!(
+ target: "iggy.metadata.diag",
+ plane = "metadata",
+ replica_id = consensus.replica(),
+ view = state.view,
+ log_view = state.log_view,
+ superblock_write_failures = failures,
+ retry_in_micros = backoff,
+ %error,
+ "superblock persist failed; withholding every view-scoped send \
+ until it succeeds, so this replica stays quorum-invisible"
+ );
+ }
+ false
+ }
+ }
+ }
+
+ /// Consecutive failed superblock writes, `0` when the last one succeeded. Read by
+ /// diagnostics: a non-zero value means this replica is fenced out of view changes.
+ #[must_use]
+ pub const fn superblock_write_failures(&self) -> u64 {
+ self.superblock_write_failures.get()
+ }
+}
+
impl<B, P, J, S, M> IggyMetadata<VsrConsensus<B, P>, J, S, M>
where
B: MessageBus,
@@ -2227,6 +2618,22 @@ where
let Some(coordinator) = &self.coordinator else {
return;
};
+ // Serialize whole checkpoints against each other. In-process metadata submits
+ // each run on their own spawned task (`bus.spawn` in server-ng's metadata submit
+ // handler), so at the checkpoint margin two can enter here concurrently; without
+ // this lock they would run concurrent `persist_snapshot`s over the single
+ // `snapshot.bin` and concurrently `drain` the WAL, which rewrites through a
+ // shared `wal.tmp`. Acquire BEFORE `should_checkpoint` so check and sequence are
+ // atomic: the second caller re-checks under the lock, finds the margin restored
+ // by the first's drain, and coalesces away the redundant work.
+ //
+ // The superblock write below takes `superblock_lock` for itself, so a
+ // concurrent view persist (and with it every gated send, including the ack
+ // path) waits only for that write and not for this whole sequence.
+ let _checkpoint = self.checkpoint_lock.acquire().await;
+ if !coordinator.should_checkpoint(journal) {
+ return;
+ }
// Use commit_min (locally executed), not commit_max. WAL entries
// between commit_min+1 and commit_max haven't been applied to the
@@ -2236,31 +2643,82 @@ where
// under the simulator), not the wall clock, so replayed snapshots are
// byte-identical.
let created_at = consensus.clock_realtime_micros();
- match coordinator
- .checkpoint_if_needed(&self.mux_stm, journal, snap_op, created_at)
- .await
- {
- Ok(true) => {
- debug!(
+
+ // Durability ordering, must not be reordered: persist the snapshot, durably
+ // record the (checkpoint_op, checksum, commit_max) pairing in the superblock,
+ // THEN drain the snapshotted prefix from the WAL. A crash before the
+ // superblock write recovers the prior checkpoint with the WAL intact; a crash
+ // after it recovers the new one. Draining before the superblock points at the
+ // new snapshot could strand committed ops on a crash. Each fallible step
+ // withholds the rest and returns early, leaving the WAL undrained, so
+ // `should_checkpoint` stays true and the next tick retries from the top at the
+ // then-current commit_min. The prepare being replicated appends regardless
+ // (see the phantom-op comment at the call site).
+ let client_table = self.client_table.borrow().to_snapshot();
+ let checksum = match coordinator.persist_snapshot(
+ &self.mux_stm,
+ snap_op,
+ created_at,
+ Some(client_table),
+ ) {
+ Ok(checksum) => checksum,
+ Err(e) => {
+ error!(
target: "iggy.metadata.diag",
plane = "metadata",
replica_id = consensus.replica(),
checkpoint_op = snap_op,
- "forced checkpoint completed"
+ error = %e,
+ "checkpoint snapshot persist failed"
);
+ return;
}
- Ok(false) => {}
- Err(e) => {
+ };
+
+ if let Some(superblock) = self.superblock.as_ref() {
+ // `persist_snapshot` already recorded the new pairing on the coordinator, so
+ // `write_superblock` picks it up from there. A view persist that interleaves
+ // between those two steps writes the same new pairing, which only makes it
+ // durable sooner.
+ debug_assert_eq!(
+ self.coordinator
+ .as_ref()
+ .map(SnapshotCoordinator::last_checkpoint),
+ Some((snap_op, checksum)),
+ "the checkpoint's pairing must be what the superblock write records"
+ );
+ let _superblock = self.superblock_lock.acquire().await;
+ if !self.write_superblock(consensus, superblock.as_ref()).await {
error!(
target: "iggy.metadata.diag",
plane = "metadata",
replica_id = consensus.replica(),
checkpoint_op = snap_op,
- error = %e,
- "forced checkpoint failed; continuing without WAL reclamation"
+ "checkpoint superblock write failed; withholding WAL drain"
);
+ return;
}
}
+
+ if let Err(e) = coordinator.drain(journal, snap_op).await {
+ error!(
+ target: "iggy.metadata.diag",
+ plane = "metadata",
+ replica_id = consensus.replica(),
+ checkpoint_op = snap_op,
+ error = %e,
+ "checkpoint WAL drain failed"
+ );
+ return;
+ }
+
+ debug!(
+ target: "iggy.metadata.diag",
+ plane = "metadata",
+ replica_id = consensus.replica(),
+ checkpoint_op = snap_op,
+ "forced checkpoint completed"
+ );
}
#[allow(clippy::too_many_lines)]
@@ -2491,47 +2949,16 @@ where
// SM apply + client_table mutation BEFORE `advance_commit_min`
// (see `on_ack` for matching invariant). No await between table
- // mutation and counter bump.
- if header.operation == Operation::Register {
- // Register: commit_register creates session, no SM.
- let reply = build_reply_message(&header, &bytes::Bytes::new());
- self.client_table.borrow_mut().commit_register(
- header.client,
- header.user_id,
- reply,
- );
- } else if header.operation == Operation::Logout {
- self.client_table.borrow_mut().remove_client(header.client);
- // Mirror the on_ack path: drop the disconnected client from
- // every consumer group it joined and rebalance.
- self.mux_stm.streams().remove_consumer_group_member(
- header.client,
- iggy_common::IggyTimestamp::from(header.timestamp),
- );
- } else {
- // Normal op: apply SM, commit_reply. `Err` is decode/corruption
- // only; a business rejection commits as a deterministic no-op
- // whose `code` rides the reply body, replayed on retry.
- let apply = gated_apply(&self.mux_stm, prepare).unwrap_or_else(|err| {
- panic!("commit_journal: committed metadata op={op} failed to apply: {err}");
- });
- // Post-commit notifier (e.g. partition reconciler
- // wake-up). Same hook fires on backups so reconcilers
- // converge after replicated commits, not only quorum-acked
- // ones reached via `on_ack` on the primary.
- self.fire_commit_notifier(header.operation);
- let reply = build_reply_message_with(&header, apply.reply_body_len(), |dst| {
- apply.write_reply_body(dst);
- });
- // Best-effort cache; WAL replay may carry a reply for a
- // later-evicted client, and replica-local eviction makes a
- // stale-request replay reachable. Both are skips, not faults.
- let outcome = self
- .client_table
- .borrow_mut()
- .commit_reply(header.client, reply);
- log_commit_reply_outcome(outcome, header.client, op);
- }
+ // mutation and counter bump. The post-commit notifier (e.g. partition
+ // reconciler wake-up) fires on backups too, so reconcilers converge
+ // after replicated commits, not only quorum-acked ones reached via
+ // `on_ack` on the primary.
+ apply_committed_prepare(
+ &self.mux_stm,
+ &self.client_table,
+ |operation| self.fire_commit_notifier(operation),
+ prepare,
+ );
consensus.advance_commit_min(op);
debug!("commit_journal: committed op={op}");
}
@@ -2575,6 +3002,14 @@ where
#[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
async fn send_prepare_ok(&self, header: &PrepareHeader) {
let consensus = self.consensus.as_ref().unwrap();
+ // Durable-before-send: a PrepareOk implies this replica's (view, log_view), so
+ // it must not leave until they are durable, or a crash could recover an older
+ // view than the one this ack helped commit in, losing a committed op. Mirrors
+ // the view-change dispatch gate; withhold on persist failure and let the
+ // primary's prepare retransmit re-drive the ack once the next tick persists.
+ if !self.persist_superblock_if_needed(consensus).await {
+ return;
+ }
let journal = self.journal.as_ref().unwrap();
let persisted = journal.handle().header(header.op as usize).is_some();
send_prepare_ok_common(consensus, header, Some(persisted)).await;
@@ -2844,6 +3279,12 @@ where
// UpdateTopic default-size rewrite, and the PAT-cleaner delete), which
// would otherwise reset it to 0 via `..Default::default()`.
user_id: request.user_id,
+ // Seal the body integrity field over the rewritten body, exactly as
+ // `Project::project` does for wire-projected prepares. This helper builds
+ // a NEW body, so the wire header's stamp does not describe it; leaving it
+ // zero makes the journal scan read every rewritten entry as corrupt and
+ // refuse boot on the next restart.
+ checksum_body: u128::from(iggy_common::calculate_checksum(body)),
..Default::default()
};
@@ -3000,6 +3441,95 @@ mod tests {
);
}
+ #[test]
+ fn populated_snapshot_reencode_and_checksum_are_stable() {
+ // Two replicas holding identical state must serialize it identically, which
+ // holds only while every serialized collection keeps a deterministic order.
+ // Fold a multi-slot client table (the envelope's own Vec) and assert the raw
+ // re-encode survives the round-trip. The streams sub-tree's map-derived Vecs are
+ // covered by
+ // `crate::stm::stream::tests::populated_streams_snapshot_reencode_is_byte_stable`.
+ let mut snapshot = IggySnapshot::new(7);
+ snapshot.snapshot.client_table = Some(consensus::ClientTableSnapshot {
+ slots: vec![
+ (
+ 0,
+ consensus::ClientEntrySnapshot {
+ client_id: 1,
+ epoch: 10,
+ user_id: 1,
+ watermark: 3,
+ watermark_checksum: 0xabc,
+ reply: vec![1, 2, 3],
+ },
+ ),
+ (
+ 2,
+ consensus::ClientEntrySnapshot {
+ client_id: 2,
+ epoch: 20,
+ user_id: 2,
+ watermark: 0,
+ watermark_checksum: 0,
+ reply: vec![4, 5],
+ },
+ ),
+ ],
+ });
+
+ let encoded = snapshot.encode().unwrap();
+ let decoded = IggySnapshot::decode(&encoded).unwrap();
+ assert_eq!(
+ encoded,
+ decoded.encode().unwrap(),
+ "a populated snapshot must re-encode byte-identically after a decode; an \
+ unordered collection in the serialized form would make two replicas with \
+ identical state serialize differently"
+ );
+ assert_ne!(
+ checkpoint_checksum(&encoded),
+ checkpoint_checksum(&IggySnapshot::new(7).encode().unwrap()),
+ "the checksum must track content, else it could not detect a torn snapshot"
+ );
+ }
+
+ #[test]
+ fn client_table_reply_bytes_encode_as_a_msgpack_blob() {
+ // A `Vec<u8>` serialized through serde's sequence path spends 2 bytes on every
+ // byte >= 0x80, so a checkpoint's reply payload runs up to roughly double.
+ // Reply bytes are wire messages, mostly high bytes, so pin the `bin` encoding:
+ // the format freezes at release and this is much cheaper to fix now.
+ const REPLY_LEN: usize = 512;
+ let mut snapshot = IggySnapshot::new(1);
+ snapshot.snapshot.client_table = Some(consensus::ClientTableSnapshot {
+ slots: vec![(
+ 0,
+ consensus::ClientEntrySnapshot {
+ client_id: 1,
+ epoch: 1,
+ user_id: 1,
+ watermark: 0,
+ watermark_checksum: 0,
+ reply: vec![0xFF; REPLY_LEN],
+ },
+ )],
+ });
+
+ let encoded = snapshot.encode().unwrap();
+ let baseline = IggySnapshot::new(1).encode().unwrap().len();
+ let reply_cost = encoded.len() - baseline;
+ assert!(
+ reply_cost < REPLY_LEN * 2,
+ "a {REPLY_LEN}-byte reply of high bytes cost {reply_cost} bytes, so it is \
+ still encoding as an integer array rather than a msgpack blob"
+ );
+
+ // And it must decode back to the same bytes.
+ let decoded = IggySnapshot::decode(&encoded).unwrap();
+ let table = decoded.snapshot().client_table.as_ref().unwrap();
+ assert_eq!(table.slots[0].1.reply, vec![0xFF; REPLY_LEN]);
+ }
+
type TestMux = MuxStateMachine<variadic!(Users, Streams)>;
/// Build a peer-shard-style `IggyMetadata` with `consensus`,
@@ -3008,7 +3538,7 @@ mod tests {
/// the test picks `()` for `C` / `J` / `S` since no notifier code path
/// touches their methods.
fn peer_metadata() -> IggyMetadata<(), (), (), TestMux> {
- IggyMetadata::new(None, None, None, TestMux::default(), None)
+ IggyMetadata::new(None, None, None, None, TestMux::default(), None)
}
#[test]
@@ -3148,6 +3678,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3249,7 +3780,7 @@ mod tests {
LocalPipeline::new(),
);
consensus.init();
- IggyMetadata::new(Some(consensus), None, None, TestMux::default(), None)
+ IggyMetadata::new(Some(consensus), None, None, None, TestMux::default(), None)
}
fn create_topic_request(client: u128, wire_user_id: u32) -> Message<RequestHeader> {
@@ -3417,6 +3948,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3593,6 +4125,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3743,6 +4276,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
Some(dir.path().to_path_buf()),
);
@@ -3888,6 +4422,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -4012,6 +4547,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -4147,6 +4683,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 35d26669f..46aba04a6 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -19,13 +19,19 @@ use crate::impls::metadata::IggySnapshot;
use crate::stm::StateMachine;
use crate::stm::authz::GatedApply;
use crate::stm::snapshot::{MetadataSnapshot, RestoreSnapshot, Snapshot, SnapshotError};
-use consensus::{ClientTable, build_reply_message, build_reply_message_with};
+use consensus::{
+ ClientTable, ClientTableDecodeError, VsrState, VsrStateError, build_reply_message,
+ build_reply_message_with,
+};
use iggy_binary_protocol::consensus::{Operation, PrepareHeader};
use iggy_common::IggyError;
use journal::prepare_journal::{JournalError, PrepareJournal};
+use journal::superblock::{
+ PingPongSuperblock, SLOT_FILE_NAMES, SuperblockContents, SuperblockStore,
+};
use server_common::Message;
use std::fmt;
-use std::path::Path;
+use std::path::{Path, PathBuf};
/// Error type for metadata recovery.
#[derive(Debug)]
@@ -34,6 +40,103 @@ pub enum RecoveryError {
Journal(JournalError),
StateMachine(IggyError),
Io(std::io::Error),
+ /// Opening or reading the durable superblock failed. `dir` is the metadata
+ /// directory holding [`journal::superblock::SLOT_FILE_NAMES`]: `compio` attaches no
+ /// path to its I/O errors, so without this the operator's chain ends in a bare
+ /// `ENOENT`.
+ Superblock {
+ dir: PathBuf,
+ source: std::io::Error,
+ },
+ /// A checkpoint client-table slot held bytes that are not a valid reply
+ /// message (a corrupt or torn snapshot).
+ ClientTableDecode(ClientTableDecodeError),
+ /// The superblock references a checkpoint newer than the on-disk snapshot, a
+ /// lost or reverted snapshot write: recovering from the older snapshot while
+ /// trusting the superblock's commit point could skip committed state.
+ CheckpointAheadOfSnapshot {
+ dir: PathBuf,
+ checkpoint_op: u64,
+ snapshot_op: u64,
+ },
+ /// The paired snapshot's checksum does not match the superblock's record (a
+ /// corrupt or torn snapshot).
+ CheckpointChecksumMismatch {
+ dir: PathBuf,
+ op: u64,
+ expected: u128,
+ actual: u128,
+ },
+ /// A superblock record was present and checksum-clean but its payload did not
+ /// decode into a [`VsrState`]: a layout change that skipped a version bump, or
+ /// corruption inside the checksummed region. The durable consensus state is
+ /// unrecoverable, so refuse boot rather than infer a stale view from the WAL and
+ /// risk re-entering a superseded view.
+ SuperblockUndecodable {
+ dir: PathBuf,
+ source: VsrStateError,
+ },
+ /// The slots yield no record this build can trust: bytes that fail verification
+ /// on every copy, or one clean record beside a copy that does not verify, whose
+ /// lost `sequence` may have been the newer (`version` names an unrecognized
+ /// format version when that was the cause). A superblock was written and its
+ /// latest generation cannot be established, so refuse boot rather than treat the
+ /// node as fresh or read through to a superseded view. An empty superblock is a
+ /// different case, a genuinely fresh or not-yet-checkpointed node, and is NOT an
+ /// error.
+ SuperblockUnreadable {
+ dir: PathBuf,
+ version: Option<u16>,
+ },
+ /// The superblock was written by a different cluster or a different replica
+ /// index: a copied, restored-to-the-wrong-host, or otherwise misplaced data
+ /// directory. Recovering `(view, log_view)` from another replica's record would
+ /// have this node act on consensus numbers it never reached.
+ SuperblockIdentityMismatch {
+ dir: PathBuf,
+ field: IdentityField,
+ expected: u128,
+ found: u128,
+ },
+}
+
+/// Which identity field of a durable [`VsrState`] failed to match this replica.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum IdentityField {
+ Cluster,
+ ReplicaId,
+ ReplicaCount,
+}
+
+impl fmt::Display for IdentityField {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::Cluster => write!(f, "cluster"),
+ Self::ReplicaId => write!(f, "replica_id"),
+ Self::ReplicaCount => write!(f, "replica_count"),
+ }
+ }
+}
+
+/// The identity a replica expects its own durable superblock to carry.
+///
+/// Bundled rather than passed as three positional scalars, since `cluster`,
+/// `replica_id` and `replica_count` are exactly the values a silent misbinding would
+/// make unrecoverable.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ReplicaIdentity {
+ pub cluster: u128,
+ pub replica_id: u8,
+ pub replica_count: u8,
+}
+
+impl ReplicaIdentity {
+ /// A single-replica cluster: quorum is 1/1, so every journaled op committed the
+ /// moment it was written.
+ #[must_use]
+ pub const fn is_solo(&self) -> bool {
+ self.replica_count == 1
+ }
}
impl fmt::Display for RecoveryError {
@@ -43,6 +146,74 @@ impl fmt::Display for RecoveryError {
Self::Journal(e) => write!(f, "recovery journal error: {e}"),
Self::StateMachine(e) => write!(f, "recovery state machine error: {e}"),
Self::Io(e) => write!(f, "recovery I/O error: {e}"),
+ Self::Superblock { dir, source } => write!(
+ f,
+ "recovery superblock error in {} ({} / {}): {source}",
+ dir.display(),
+ SLOT_FILE_NAMES[0],
+ SLOT_FILE_NAMES[1]
+ ),
+ Self::ClientTableDecode(e) => write!(f, "recovery client-table decode error: {e}"),
+ Self::CheckpointAheadOfSnapshot {
+ dir,
+ checkpoint_op,
+ snapshot_op,
+ } => write!(
+ f,
+ "superblock in {} references checkpoint op {checkpoint_op} but the on-disk \
+ snapshot is only at op {snapshot_op}: a lost or reverted snapshot write, \
+ refusing boot",
+ dir.display()
+ ),
+ Self::CheckpointChecksumMismatch {
+ dir,
+ op,
+ expected,
+ actual,
+ } => write!(
+ f,
+ "snapshot at op {op} in {} does not match the superblock's checkpoint checksum \
+ (superblock {expected:#034x}, snapshot {actual:#034x}): corrupt or torn snapshot, \
+ refusing boot",
+ dir.display()
+ ),
+ Self::SuperblockUndecodable { dir, source } => write!(
+ f,
+ "superblock record in {} is present and checksum-clean but its payload did not \
+ decode into VSR state ({source}): unrecoverable durable consensus state, \
+ refusing boot",
+ dir.display()
+ ),
+ Self::SuperblockUnreadable {
+ dir,
+ version: Some(version),
+ } => write!(
+ f,
+ "superblock in {} present but its format version {version} is unrecognized by \
+ this build (a downgrade, or a corrupt version field): refusing boot",
+ dir.display()
+ ),
+ Self::SuperblockUnreadable { dir, version: None } => write!(
+ f,
+ "superblock present but a copy holds bytes that do not verify (bit-rot or a \
+ checksum failure), so its latest generation cannot be established: refusing \
+ boot; inspect {} and {} in {}",
+ SLOT_FILE_NAMES[0],
+ SLOT_FILE_NAMES[1],
+ dir.display()
+ ),
+ Self::SuperblockIdentityMismatch {
+ dir,
+ field,
+ expected,
+ found,
+ } => write!(
+ f,
+ "superblock {field} in {} is {found} but this replica is configured with \
+ {expected}: the metadata directory belongs to another cluster or replica, or the \
+ cluster was resized without reconfiguration, refusing boot",
+ dir.display()
+ ),
}
}
}
@@ -53,11 +224,23 @@ impl std::error::Error for RecoveryError {
Self::Snapshot(e) => Some(e),
Self::Journal(e) => Some(e),
Self::StateMachine(e) => Some(e),
- Self::Io(e) => Some(e),
+ Self::Io(e) | Self::Superblock { source: e, .. } => Some(e),
+ Self::ClientTableDecode(e) => Some(e),
+ Self::SuperblockUndecodable { source, .. } => Some(source),
+ Self::CheckpointAheadOfSnapshot { .. }
+ | Self::CheckpointChecksumMismatch { .. }
+ | Self::SuperblockUnreadable { .. }
+ | Self::SuperblockIdentityMismatch { .. } => None,
}
}
}
+impl From<ClientTableDecodeError> for RecoveryError {
+ fn from(e: ClientTableDecodeError) -> Self {
+ Self::ClientTableDecode(e)
+ }
+}
+
impl From<SnapshotError> for RecoveryError {
fn from(e: SnapshotError) -> Self {
Self::Snapshot(e)
@@ -86,13 +269,31 @@ impl From<std::io::Error> for RecoveryError {
pub struct RecoveredMetadata<M> {
pub journal: PrepareJournal,
pub snapshot: Option<IggySnapshot>,
+ /// The durable superblock, opened by recovery so it could cross-check the
+ /// snapshot's op and checksum against the checkpoint pairing before restoring the
+ /// state machine from it. Handed to the consensus/metadata layer to reuse;
+ /// re-opening would fork the ping-pong sequence counter.
+ pub superblock: PingPongSuperblock,
+ /// The durable VSR state read from the superblock, or `None` when none has been
+ /// written yet (a fresh deployment, or a node that took writes but never
+ /// checkpointed or changed view). A present but unreadable or undecodable
+ /// superblock refuses boot instead ([`RecoveryError::SuperblockUnreadable`] /
+ /// [`RecoveryError::SuperblockUndecodable`]). Consensus restores
+ /// `(view, log_view)` from it; recovery has already verified the paired snapshot
+ /// against it.
+ pub recovered_state: Option<VsrState>,
+ /// `(snapshot_op, snapshot_checksum)` of the verified on-disk snapshot, `(0, 0)`
+ /// when none. Seeds the coordinator's last-checkpoint pairing so the first
+ /// post-boot view-change superblock write records the real pairing.
+ pub snapshot_checkpoint: (u64, u128),
pub mux_stm: M,
- /// Client table rebuilt from the replayed committed prefix: registers
- /// re-mint epochs in apply order, replies re-cache byte-identically
- /// (`build_reply_message*` reads only the prepare header + deterministic
- /// apply output). Sessions whose register fell below the snapshot floor
- /// are NOT recovered - the table has no checkpoint artifact yet
- /// (IGGY-137); those clients re-register and their epoch restarts at 1.
+ /// Client table restored from the checkpoint and advanced by the replayed
+ /// committed suffix: registers carry their own commit op as the epoch, so
+ /// replay reads fences out of the log rather than deriving them from replay
+ /// order, and replies re-cache byte-identically (`build_reply_message*` reads
+ /// only the prepare header + deterministic apply output). Sessions whose
+ /// register fell below the snapshot floor come back from the checkpoint's
+ /// folded table, watermarks included.
pub client_table: ClientTable,
/// `None` means no snapshot existed and no journal entries were replayed.
/// `Some(op)` is the highest op applied, either from the snapshot or journal replay.
@@ -134,16 +335,18 @@ pub struct RecoveredMetadata<M> {
/// caller built, so reading the compile-time default here would make the knob
/// inert on every restart.
///
-/// `solo` marks a single-replica cluster: the quorum is 1/1, so every
-/// journaled op was committed the moment it was written and replay runs to
+/// `identity` is what this replica expects its own durable superblock to carry, and
+/// a mismatch refuses boot ([`RecoveryError::SuperblockIdentityMismatch`]). Its
+/// `replica_count == 1` also marks a single-replica cluster: the quorum is 1/1, so
+/// every journaled op was committed the moment it was written and replay runs to
/// the journal head. The embedded `commit` stamps cannot be used there: each
/// stamp is the primary's commit point when the prepare was SENT, so a
/// pipelined burst (e.g. create-stream + create-topic on one connection) is
/// stamped entirely below its own ops and would replay as nothing.
-#[allow(clippy::future_not_send)]
+#[allow(clippy::future_not_send, clippy::too_many_lines)]
pub async fn recover<M>(
data_dir: &Path,
- solo: bool,
+ identity: ReplicaIdentity,
journal_slots: usize,
clients_table_max: usize,
seed_baseline: impl FnOnce(&M),
@@ -158,11 +361,90 @@ where
std::fs::create_dir_all(&metadata_dir)?;
let snapshot_path = metadata_dir.join("snapshot.bin");
- let snapshot = if snapshot_path.exists() {
- Some(IggySnapshot::load(&snapshot_path)?)
+ // The checksum is over the bytes on disk, so the pairing cross-check below holds
+ // across a schema change: see `IggySnapshot::load`.
+ let (snapshot, snapshot_checksum) = if snapshot_path.exists() {
+ let (snapshot, checksum) = IggySnapshot::load(&snapshot_path)?;
+ (Some(snapshot), checksum)
} else {
- None
+ (None, 0)
};
+
+ // Open the durable superblock and read the last VSR state. Recovery owns this so
+ // it can verify the snapshot against the superblock's checkpoint pairing BEFORE
+ // trusting (decoding) it into the state machine and client table. The consensus
+ // layer reuses the returned superblock, since re-opening would fork the ping-pong
+ // sequence counter, and restores `(view, log_view)` from `recovered_state`.
+ let superblock = PingPongSuperblock::open(&metadata_dir)
+ .await
+ .map_err(|source| RecoveryError::Superblock {
+ dir: metadata_dir.clone(),
+ source,
+ })?;
+ // Distinguish the three read outcomes so a lost or corrupt superblock cannot
+ // masquerade as a fresh deployment: a present but unreadable or
+ // unrecognized-version record refuses boot with a typed error. An EMPTY
+ // superblock is genuinely fresh, or a node that took writes but never
+ // checkpointed or changed view, and is safe to treat as no durable state: the
+ // persist-before-send gate guarantees this replica never externalized a view it
+ // cannot re-derive by re-probing, so it recovers `(view, log_view)` from the
+ // WAL/init path with no split-brain risk.
+ let recovered_state =
+ match superblock
+ .read_latest()
+ .await
+ .map_err(|source| RecoveryError::Superblock {
+ dir: metadata_dir.clone(),
+ source,
+ })? {
+ SuperblockContents::Present(bytes) => {
+ // A checksum-clean record that does not decode is a durability violation,
+ // not absent state: refuse boot rather than infer a stale view.
+ Some(VsrState::try_from(bytes.as_slice()).map_err(|source| {
+ RecoveryError::SuperblockUndecodable {
+ dir: metadata_dir.clone(),
+ source,
+ }
+ })?)
+ }
+ SuperblockContents::Unreadable { version } => {
+ return Err(RecoveryError::SuperblockUnreadable {
+ dir: metadata_dir.clone(),
+ version,
+ });
+ }
+ SuperblockContents::Empty => None,
+ };
+
+ // The superblock is the only on-disk identity in the metadata directory, and the
+ // wire handshake cannot stand in for it: both peers derive `cluster` from the
+ // configured cluster name, so it never sees a foreign disk.
+ if let Some(state) = recovered_state.as_ref() {
+ verify_identity(state, identity, &metadata_dir)?;
+ }
+
+ // Second of the snapshot's two integrity checks, and the one the superblock owns.
+ // Ordering, precisely: `IggySnapshot::load` verified the snapshot's own trailer
+ // against its payload before decoding it, and this cross-checks the decoded
+ // snapshot's op against the superblock's pairing. The pairing check cannot run
+ // first, since it needs `sequence_number` out of the decoded snapshot.
+ //
+ // A superblock pointing past the snapshot (a lost snapshot write) or disagreeing on
+ // its checksum (corrupt or torn) is a durability violation, not a recoverable
+ // state. One that merely lags a newer, atomically-complete snapshot is accepted,
+ // since `commit_max` is a recovery lower bound and the WAL suffix re-commits.
+ //
+ // TODO(state-transfer): there is no metadata state transfer yet, so refusing boot
+ // is the only sound response. Once it lands, fetch the checkpoint from a healthy
+ // replica here instead of erroring.
+ let snapshot_op = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number);
+ verify_checkpoint_pairing(
+ recovered_state.as_ref(),
+ snapshot_op,
+ snapshot_checksum,
+ &metadata_dir,
+ )?;
+
let replay_from = snapshot
.as_ref()
.map_or(0, |snapshot| snapshot.sequence_number() + 1);
@@ -179,6 +461,19 @@ where
let watermark = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number);
let journal = PrepareJournal::open_with_slots(&journal_path, watermark, journal_slots).await?;
+ // The scan replays entries no producer sealed rather than rejecting them, so
+ // report the count: an operator gets no other signal that these bodies were
+ // replayed on trust. Goes away once every node in the cluster seals.
+ let unsealed_entries = journal.unsealed_entry_count();
+ if unsealed_entries > 0 {
+ tracing::warn!(
+ unsealed_entries,
+ path = %journal_path.display(),
+ "metadata WAL holds entries without a body checksum; replayed unverified \
+ (written before body sealing, or replicated from a primary that predates it)"
+ );
+ }
+
// Intentional fail-fast: a bad entry aborts recovery and the operator
// must repair or truncate the WAL before the node can boot again.
let headers_to_replay = journal.iter_headers_from(replay_from);
@@ -192,7 +487,7 @@ where
// future non-monotone stamping change cannot silently under-apply the
// committed prefix.
let snapshot_floor = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number);
- let commit_watermark = if solo {
+ let commit_watermark = if identity.is_solo() {
headers_to_replay
.iter()
.map(|header| header.op)
@@ -204,7 +499,41 @@ where
.fold(snapshot_floor, u64::max)
};
- let mut client_table = ClientTable::new(clients_table_max);
+ // The superblock's `commit_max` is the highest op this replica knew the cluster had
+ // committed when it last wrote. Report a gap against what the WAL can prove, since
+ // nothing else surfaces "committed ops this node cannot see".
+ //
+ // A warning and not a refusal, deliberately: each journaled prepare stamps the
+ // primary's commit point as of its SEND, so the derived watermark is a lower bound
+ // and trails `commit_max` on a perfectly healthy node. Only state transfer can turn
+ // this into a decision, which is why the field has no other reader (see the
+ // TODO(state-transfer) above).
+ if let Some(state) = recovered_state.as_ref()
+ && state.commit_max > commit_watermark
+ {
+ tracing::warn!(
+ durable_commit_max = state.commit_max,
+ wal_commit_watermark = commit_watermark,
+ snapshot_op,
+ "superblock records a commit point above what this replica's WAL can prove; \
+ the gap re-commits from the cluster, or needs state transfer if it does not"
+ );
+ }
+
+ // Restore the client table from the checkpoint's folded copy, then replay the
+ // committed suffix on top, mirroring how the state machine is restored from the
+ // snapshot and advanced by the WAL. This is what lets a session registered below
+ // the snapshot floor come back with its watermark intact.
+ let mut client_table = match snapshot
+ .as_ref()
+ .and_then(|snapshot| snapshot.snapshot().client_table.clone())
+ {
+ // Integrity is verified above, so a decode failure here means genuinely
+ // corrupt bytes: refuse boot with a typed error rather than panicking.
+ Some(table) => ClientTable::from_snapshot(table, clients_table_max)?,
+ None => ClientTable::new(clients_table_max),
+ };
+
let mut last_applied_op: Option<u64> = None;
let mut last_journaled_op: Option<u64> = None;
for header in &headers_to_replay {
@@ -224,24 +553,21 @@ where
// own commit op, so replay reads them out of the log rather than
// deriving them from replay order.
//
- // Watermarks do NOT, and this is a known gap rather than an invariant.
- // Replay starts at THIS node's snapshot floor, and checkpointing is
- // node-local (`checkpoint_if_needed` fires on local journal occupancy),
- // so replicas cross the floor at different ops. If a client's earlier
- // register fell below this node's floor while a later one survived,
- // replay takes `commit_register`'s fresh-entry branch and the entry
- // returns with `watermark = 0`, where a peer that replayed both took
- // the rebind branch and kept it. The fence still passes, so nothing is
- // evicted, and the same request id a peer answers `Duplicate` gets
- // answered `New` here and re-executed -- exactly-once degrading to
- // at-least-once, silently.
+ // Watermarks do not come out of replay alone: replay starts at THIS
+ // node's snapshot floor, and checkpointing is node-local
+ // (`checkpoint_if_needed` fires on local journal occupancy), so replicas
+ // cross the floor at different ops. A client whose earlier register fell
+ // below this node's floor would take `commit_register`'s fresh-entry
+ // branch and come back with `watermark = 0`, where a peer that replayed
+ // both took the rebind branch and kept it -- the same request id one node
+ // answers `Duplicate` the other re-executes, exactly-once degrading to
+ // at-least-once. That is why the table is folded into the checkpoint
+ // above: the pre-floor watermark is restored from the snapshot, and
+ // replay only advances it.
//
- // Unobservable today (no shipping client re-presents a recovered
- // `client_id`), and closed by putting the table in the snapshot so
- // replay no longer has to reconstruct it. Until then a caught-up
- // primary is authoritative for its OWN table only, which is why
- // `request_preflight`'s catch-up gate cannot bridge this (see its
- // rustdoc).
+ // Residual gap: a checkpoint written before the fold existed carries no
+ // table, so recovery from one still starts empty. It converges once the
+ // node takes its next checkpoint.
if header.operation == Operation::Register {
let reply = build_reply_message(header, &bytes::Bytes::new());
client_table.commit_register(header.client, header.user_id, reply);
@@ -270,7 +596,7 @@ where
// Re-cache the reply exactly like the commit paths: same prepare
// header + deterministic apply output = the original bytes. Skipped
// when the session is absent (server-originated ops, or the client
- // was evicted / registered below the snapshot floor).
+ // was evicted).
if client_table.get_epoch(header.client).is_some() {
let cached = build_reply_message_with(header, reply.reply_body_len(), |dst| {
reply.write_reply_body(dst);
@@ -286,7 +612,6 @@ where
op = header.op,
operation = ?header.operation,
user_id = header.user_id,
- reply = ?reply,
"recovery replayed op"
);
last_applied_op = Some(header.op);
@@ -295,6 +620,9 @@ where
Ok(RecoveredMetadata {
journal,
snapshot,
+ superblock,
+ recovered_state,
+ snapshot_checkpoint: (snapshot_op, snapshot_checksum),
mux_stm,
client_table,
last_applied_op,
@@ -302,10 +630,111 @@ where
})
}
+/// Refuse a durable record this replica did not write.
+///
+/// Scope it honestly: identity catches MISPLACEMENT, not staleness. An older backup
+/// of this replica's own directory carries the right `cluster`, `replica_id` and
+/// `replica_count` and passes here; only the checkpoint pairing and the WAL constrain
+/// how old it may be.
+///
+/// `replica_count` is checked alongside the identity pair because quorum size and the
+/// `view % replica_count` primary mapping both derive from it. There is no VSR
+/// reconfiguration yet, so a count that disagrees with the config means the cluster
+/// was resized underneath a node, and booting it would have replicas disagree on both
+/// quorum and who leads a view.
+///
+/// # Errors
+/// [`RecoveryError::SuperblockIdentityMismatch`] naming the first field that differs
+/// and the directory it was read from, since which directory a node was pointed at is
+/// the whole diagnosis here.
+fn verify_identity(
+ state: &VsrState,
+ expected: ReplicaIdentity,
+ dir: &Path,
+) -> Result<(), RecoveryError> {
+ let mismatch = |field, expected: u128, found: u128| RecoveryError::SuperblockIdentityMismatch {
+ dir: dir.to_path_buf(),
+ field,
+ expected,
+ found,
+ };
+ if state.cluster != expected.cluster {
+ return Err(mismatch(
+ IdentityField::Cluster,
+ expected.cluster,
+ state.cluster,
+ ));
+ }
+ if state.replica_id != expected.replica_id {
+ return Err(mismatch(
+ IdentityField::ReplicaId,
+ expected.replica_id.into(),
+ state.replica_id.into(),
+ ));
+ }
+ if state.replica_count != expected.replica_count {
+ return Err(mismatch(
+ IdentityField::ReplicaCount,
+ expected.replica_count.into(),
+ state.replica_count.into(),
+ ));
+ }
+ Ok(())
+}
+
+/// Cross-check the superblock's checkpoint pairing against the on-disk snapshot
+/// BEFORE the snapshot's decoded contents are trusted. The superblock's
+/// `(checkpoint_op, checkpoint_checksum)` identify the snapshot its `commit_max` was
+/// durable against.
+///
+/// - `checkpoint_op > snapshot_op`: the superblock references a checkpoint newer than
+/// the snapshot on disk, a lost or reverted snapshot write. Recovering from the
+/// older snapshot while trusting `commit_max` could skip committed state, so refuse
+/// boot.
+/// - `checkpoint_op == snapshot_op` with differing checksums: the paired snapshot is
+/// corrupt or torn. Refuse boot.
+/// - `checkpoint_op < snapshot_op`: the pairing lags a newer, atomically-complete
+/// snapshot, from a checkpoint whose superblock update had not yet landed. The
+/// snapshot subsumes it and `commit_max` is only a recovery lower bound, so accept.
+/// - No recovered state, meaning a fresh or not-yet-checkpointed node: the WAL and
+/// snapshot inference stands alone, nothing to cross-check. A present but
+/// unreadable superblock refuses boot upstream, so it never arrives here as `None`.
+///
+/// # Errors
+/// [`RecoveryError::CheckpointAheadOfSnapshot`] or
+/// [`RecoveryError::CheckpointChecksumMismatch`] on a durability violation.
+fn verify_checkpoint_pairing(
+ recovered: Option<&VsrState>,
+ snapshot_op: u64,
+ snapshot_checksum: u128,
+ dir: &Path,
+) -> Result<(), RecoveryError> {
+ let Some(state) = recovered else {
+ return Ok(());
+ };
+ if state.checkpoint_op > snapshot_op {
+ return Err(RecoveryError::CheckpointAheadOfSnapshot {
+ dir: dir.to_path_buf(),
+ checkpoint_op: state.checkpoint_op,
+ snapshot_op,
+ });
+ }
+ if state.checkpoint_op == snapshot_op && state.checkpoint_checksum != snapshot_checksum {
+ return Err(RecoveryError::CheckpointChecksumMismatch {
+ dir: dir.to_path_buf(),
+ op: state.checkpoint_op,
+ expected: state.checkpoint_checksum,
+ actual: snapshot_checksum,
+ });
+ }
+ Ok(())
+}
+
#[cfg(test)]
#[allow(clippy::cast_possible_truncation)]
mod tests {
use super::*;
+ use crate::impls::metadata::checkpoint_checksum;
use consensus::CLIENTS_TABLE_MAX;
use iggy_binary_protocol::consensus::{Command2, Operation};
use journal::Journal;
@@ -318,6 +747,90 @@ mod tests {
const HEADER_SIZE: usize = size_of::<PrepareHeader>();
+ /// Replica 0 of a 3-node cluster 1: quorum 2, so the WAL's embedded commit
+ /// stamps bound replay.
+ const CLUSTERED: ReplicaIdentity = ReplicaIdentity {
+ cluster: 1,
+ replica_id: 0,
+ replica_count: 3,
+ };
+
+ /// The same replica running solo: quorum 1/1, so replay runs to the journal head.
+ const SOLO: ReplicaIdentity = ReplicaIdentity {
+ cluster: 1,
+ replica_id: 0,
+ replica_count: 1,
+ };
+
+ fn vsr_state_with_checkpoint(checkpoint_op: u64, checkpoint_checksum: u128) -> VsrState {
+ VsrState {
+ cluster: 1,
+ replica_id: 0,
+ replica_count: 3,
+ view: 7,
+ log_view: 7,
+ commit_max: 100,
+ checkpoint_op,
+ checkpoint_checksum,
+ }
+ }
+
+ /// The full accept/reject matrix for the checkpoint pairing cross-check, which
+ /// decides whether a node boots at all. Grouped because each case is one call
+ /// on a pure function and the boundaries only mean something together.
+ #[test]
+ fn checkpoint_pairing_accepts_consistent_pairings_and_rejects_violations() {
+ // The directory only reaches the error message, so one stand-in serves every case.
+ let dir = Path::new("/metadata");
+
+ // No superblock yet (fresh or not-yet-checkpointed): nothing to cross-check.
+ assert!(verify_checkpoint_pairing(None, 42, 0xabc, dir).is_ok());
+
+ // Exact pairing.
+ let matching = vsr_state_with_checkpoint(42, 0xabc);
+ assert!(verify_checkpoint_pairing(Some(&matching), 42, 0xabc, dir).is_ok());
+
+ // A checkpoint whose superblock update had not yet landed leaves the
+ // superblock behind an atomically-complete newer snapshot: safe, and it must
+ // NOT refuse boot on an otherwise healthy node.
+ let lagging = vsr_state_with_checkpoint(40, 0xdead);
+ assert!(verify_checkpoint_pairing(Some(&lagging), 42, 0xbeef, dir).is_ok());
+
+ // Superblock points past the snapshot on disk: a lost snapshot write.
+ let ahead = vsr_state_with_checkpoint(50, 0xabc);
+ assert!(matches!(
+ verify_checkpoint_pairing(Some(&ahead), 42, 0xabc, dir),
+ Err(RecoveryError::CheckpointAheadOfSnapshot {
+ checkpoint_op: 50,
+ snapshot_op: 42,
+ ..
+ })
+ ));
+
+ // A checkpoint claim with no snapshot on disk is the same violation at op 0.
+ let claim_without_snapshot = vsr_state_with_checkpoint(5, 0xabc);
+ assert!(matches!(
+ verify_checkpoint_pairing(Some(&claim_without_snapshot), 0, 0, dir),
+ Err(RecoveryError::CheckpointAheadOfSnapshot {
+ checkpoint_op: 5,
+ snapshot_op: 0,
+ ..
+ })
+ ));
+
+ // Same op, different content: corrupt or torn snapshot.
+ let mismatched = vsr_state_with_checkpoint(42, 0xaaaa);
+ assert!(matches!(
+ verify_checkpoint_pairing(Some(&mismatched), 42, 0xbbbb, dir),
+ Err(RecoveryError::CheckpointChecksumMismatch {
+ op: 42,
+ expected: 0xaaaa,
+ actual: 0xbbbb,
+ ..
+ })
+ ));
+ }
+
fn make_prepare(op: u64, body_size: usize) -> Message<PrepareHeader> {
make_prepare_with_commit(op, op.saturating_sub(1), body_size)
}
@@ -327,6 +840,11 @@ mod tests {
fn make_prepare_with_commit(op: u64, commit: u64, body_size: usize) -> Message<PrepareHeader> {
let total_size = HEADER_SIZE + body_size;
let mut buffer = Owned::<4096>::zeroed(total_size);
+ // Seal the body checksum so the WAL scan's integrity check accepts it,
+ // matching what the primary seals in `project`.
+ let checksum_body = u128::from(iggy_common::calculate_checksum(
+ &buffer.as_slice()[HEADER_SIZE..],
+ ));
let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
&mut buffer.as_mut_slice()[..HEADER_SIZE],
);
@@ -335,6 +853,7 @@ mod tests {
header.op = op;
header.commit = commit;
header.operation = Operation::CreateStream;
+ header.checksum_body = checksum_body;
Message::try_from(buffer).unwrap()
}
@@ -349,6 +868,9 @@ mod tests {
) -> Message<PrepareHeader> {
let total_size = HEADER_SIZE;
let mut buffer = Owned::<4096>::zeroed(total_size);
+ let checksum_body = u128::from(iggy_common::calculate_checksum(
+ &buffer.as_slice()[HEADER_SIZE..],
+ ));
let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
&mut buffer.as_mut_slice()[..HEADER_SIZE],
);
@@ -360,6 +882,7 @@ mod tests {
header.client = client;
header.user_id = user_id;
header.request = request;
+ header.checksum_body = checksum_body;
Message::try_from(buffer).unwrap()
}
@@ -368,7 +891,7 @@ mod tests {
let dir = tempdir().unwrap();
let recovered = recover::<TestStm>(
dir.path(),
- false,
+ CLUSTERED,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -393,7 +916,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- false,
+ CLUSTERED,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -428,7 +951,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- false,
+ CLUSTERED,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -470,7 +993,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- false,
+ CLUSTERED,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -507,7 +1030,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- false,
+ CLUSTERED,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -526,23 +1049,25 @@ mod tests {
);
}
- // The watermark half of table recovery does NOT survive a checkpoint, and
- // this pins that as a fact rather than a comment. Shape: a client registers,
- // commits request 1, a checkpoint lands past that register, then the client
- // rebinds. Replay starts above the floor, so it never sees the first
- // register: `commit_register` takes its fresh-entry branch and the entry
- // comes back with watermark 0, while a peer whose floor sat lower replayed
- // both registers, took the rebind branch, and kept watermark 1.
+ // A checkpoint that carries no folded client table still loses the watermark
+ // half of table recovery, and this pins that residual as a fact rather than a
+ // comment: it is the shape of a snapshot written before the fold existed.
+ // Shape: a client registers, commits request 1, a checkpoint lands past that
+ // register, then the client rebinds. Replay starts above the floor, so it
+ // never sees the first register: `commit_register` takes its fresh-entry
+ // branch and the entry comes back with watermark 0, while a peer whose floor
+ // sat lower replayed both registers, took the rebind branch, and kept
+ // watermark 1.
//
// Consequence, once a client re-presents a recovered id: the same request
// that peer answers `Duplicate` is `New` here and gets re-executed. The
// fence is unaffected -- epochs are op-derived, so this node still returns
// the second register's op.
//
- // Red/green for the table-in-snapshot work: when the table ships in the
- // checkpoint, the watermark assertion below flips to `Some(1)`.
+ // The green counterpart, where the checkpoint does carry the table, is
+ // `recover_restores_the_watermark_from_a_folded_checkpoint_table`.
#[compio::test]
- async fn recover_loses_the_watermark_when_a_checkpoint_hides_the_first_register() {
+ async fn recover_loses_the_watermark_when_a_tableless_checkpoint_hides_the_first_register() {
const CLIENT: u128 = 0x1337;
const USER: u32 = 7;
const FLOOR: u64 = 2;
@@ -574,7 +1099,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- true,
+ SOLO,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -591,8 +1116,87 @@ mod tests {
assert_eq!(
table.get_watermark(CLIENT),
Some(0),
- "KNOWN GAP: the pre-floor watermark is lost, so request 1 reads as New \
- here while a lower-floor peer answers it as a duplicate"
+ "a tableless checkpoint loses the pre-floor watermark, so request 1 reads \
+ as New here while a lower-floor peer answers it as a duplicate"
+ );
+ }
+
+ // The fold closes the gap above: same WAL and same floor, but the checkpoint
+ // carries the client table as it stood at the floor, so the pre-floor
+ // watermark is restored and only advanced by replay. Without this, replicas
+ // that checkpoint at different ops disagree on which requests are duplicates
+ // and exactly-once silently degrades to at-least-once.
+ #[compio::test]
+ async fn recover_restores_the_watermark_from_a_folded_checkpoint_table() {
+ const CLIENT: u128 = 0x1337;
+ const USER: u32 = 7;
+ const FLOOR: u64 = 2;
+
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+
+ // The table as the live commit path left it at the floor: registered at
+ // op 1, request 1 committed at op 2.
+ let mut at_checkpoint = ClientTable::new(CLIENTS_TABLE_MAX);
+ let register = make_client_prepare(1, Operation::Register, CLIENT, USER, 0);
+ at_checkpoint.commit_register(
+ CLIENT,
+ USER,
+ build_reply_message(register.header(), &bytes::Bytes::new()),
+ );
+ let app = make_client_prepare(2, Operation::CreateStream, CLIENT, USER, 1);
+ at_checkpoint.commit_reply(
+ CLIENT,
+ build_reply_message(app.header(), &bytes::Bytes::new()),
+ );
+
+ let mut snapshot = IggySnapshot::new(FLOOR);
+ snapshot.snapshot_mut().client_table = Some(at_checkpoint.to_snapshot());
+ snapshot
+ .persist(&metadata_dir.join("snapshot.bin"))
+ .unwrap();
+
+ {
+ let journal = PrepareJournal::open(&metadata_dir.join("journal.wal"), 0)
+ .await
+ .unwrap();
+ for entry in [
+ register,
+ app,
+ // The rebind, above the floor, so replay does see this one.
+ make_client_prepare(3, Operation::Register, CLIENT, USER, 0),
+ ] {
+ journal.append(entry).await.unwrap();
+ }
+ journal.storage_ref().fsync().await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ SOLO,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await
+ .unwrap();
+
+ let table = &recovered.client_table;
+ assert_eq!(
+ table.get_epoch(CLIENT),
+ Some(3),
+ "replay still advances the fence to the rebind's op"
+ );
+ assert_eq!(
+ table.get_watermark(CLIENT),
+ Some(1),
+ "the pre-floor watermark comes back from the checkpoint's folded table"
+ );
+ assert_eq!(
+ table.get_user_id(CLIENT),
+ Some(USER),
+ "the folded entry carries the acting user, so no metadata lookup is needed"
);
}
@@ -635,7 +1239,7 @@ mod tests {
// Solo: every journaled op is committed.
let recovered = recover::<TestStm>(
dir.path(),
- true,
+ SOLO,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -690,7 +1294,7 @@ mod tests {
let recovered = recover::<TestStm>(
dir.path(),
- true,
+ SOLO,
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
@@ -713,7 +1317,263 @@ mod tests {
let snapshot = IggySnapshot::new(99);
snapshot.persist(&path).unwrap();
- let loaded = IggySnapshot::load(&path).unwrap();
+ let (loaded, checksum) = IggySnapshot::load(&path).unwrap();
assert_eq!(loaded.sequence_number(), 99);
+ assert_eq!(
+ checksum,
+ checkpoint_checksum(&snapshot.encode().unwrap()),
+ "load must report the checksum of the payload bytes on disk, which is what \
+ the superblock pairing recorded"
+ );
+
+ // The trailer makes the file self-verifying, so a torn or bit-rotted snapshot is
+ // caught even when the superblock's pairing cannot judge it (a crash inside a
+ // checkpoint leaves the pairing behind the snapshot on disk).
+ let mut bytes = std::fs::read(&path).unwrap();
+ bytes[0] ^= 0xFF;
+ std::fs::write(&path, &bytes).unwrap();
+ assert!(matches!(
+ IggySnapshot::load(&path),
+ Err(SnapshotError::ChecksumMismatch { .. })
+ ));
+ }
+
+ #[compio::test]
+ async fn recover_refuses_boot_on_undecodable_superblock() {
+ // A checksum-clean record whose payload is not a valid VsrState: here a short
+ // payload, in the field a layout change that skipped a version bump or
+ // corruption inside the checksummed region. Refuse boot rather than infer a
+ // stale view from the WAL.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ {
+ let sb = PingPongSuperblock::open(&metadata_dir).await.unwrap();
+ sb.write(b"short").await.unwrap();
+ }
+
+ // `matches!` rather than `unwrap_err`: the Ok type `RecoveredMetadata` holds
+ // non-Debug handles.
+ let result = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await;
+ assert!(matches!(
+ result,
+ Err(RecoveryError::SuperblockUndecodable { .. })
+ ));
+ }
+
+ #[compio::test]
+ async fn recover_refuses_boot_on_corrupt_superblock() {
+ // Bytes on disk but no copy verifies: a superblock was written and is now
+ // torn, NOT a fresh deployment. Refuse boot rather than degrade to WAL-view
+ // inference.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ {
+ let sb = PingPongSuperblock::open(&metadata_dir).await.unwrap();
+ sb.write(b"durable").await.unwrap();
+ }
+ let path = metadata_dir.join(journal::superblock::SLOT_FILE_NAMES[0]);
+ let mut bytes = std::fs::read(&path).unwrap();
+ let last = bytes.len() - 1;
+ bytes[last] ^= 0xFF; // break the trailing checksum
+ std::fs::write(&path, &bytes).unwrap();
+
+ let result = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await;
+ assert!(matches!(
+ result,
+ Err(RecoveryError::SuperblockUnreadable { version: None, .. })
+ ));
+ }
+
+ #[compio::test]
+ async fn recover_treats_absent_superblock_with_wal_as_fresh() {
+ // A node that took writes but never checkpointed or changed view has a
+ // non-empty WAL and NO superblock. It must recover as fresh, NOT refuse boot,
+ // which would brick a healthy node. The persist-before-send gate makes this
+ // safe: it never externalized a view it cannot re-derive by re-probing.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ {
+ let journal = PrepareJournal::open(&metadata_dir.join("journal.wal"), 0)
+ .await
+ .unwrap();
+ journal.append(make_prepare(1, 32)).await.unwrap();
+ journal.storage_ref().fsync().await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await
+ .unwrap();
+ assert!(recovered.recovered_state.is_none());
+ }
+
+ #[compio::test]
+ async fn recover_refuses_boot_on_foreign_superblock_identity() {
+ // The superblock is the only on-disk identity in the metadata directory, and
+ // the wire handshake cannot substitute: both peers derive `cluster` from the
+ // configured cluster name, so it never sees a foreign DISK. Each field is
+ // checked on its own, since a copied directory, a replica index swap, and a
+ // cluster resized without reconfiguration are distinct operator mistakes.
+ let cases = [
+ (
+ VsrState {
+ cluster: 99,
+ ..vsr_state_with_checkpoint(0, 0)
+ },
+ IdentityField::Cluster,
+ ),
+ (
+ VsrState {
+ replica_id: 2,
+ ..vsr_state_with_checkpoint(0, 0)
+ },
+ IdentityField::ReplicaId,
+ ),
+ (
+ VsrState {
+ replica_count: 5,
+ ..vsr_state_with_checkpoint(0, 0)
+ },
+ IdentityField::ReplicaCount,
+ ),
+ ];
+
+ for (state, expected_field) in cases {
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ {
+ let superblock = PingPongSuperblock::open(&metadata_dir).await.unwrap();
+ superblock.write(&state.to_bytes()).await.unwrap();
+ }
+
+ let result = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await;
+ match result {
+ Err(RecoveryError::SuperblockIdentityMismatch { field, .. }) => {
+ assert_eq!(field, expected_field);
+ }
+ Err(other) => panic!("expected an identity mismatch, got {other}"),
+ Ok(_) => panic!("expected {expected_field} mismatch to refuse boot"),
+ }
+ }
+ }
+
+ #[compio::test]
+ async fn recover_accepts_snapshot_written_before_a_trailing_default_field() {
+ // Appending a `#[serde(default)]` field is this repo's forward-compatible
+ // snapshot migration, and msgpack encodes structs positionally: a file the
+ // previous build wrote decodes fine (the default fills the missing element)
+ // but re-encodes with one MORE element. A pairing checksum recomputed by
+ // re-encoding the decoded snapshot would therefore diverge on the FIRST boot
+ // of the new build and refuse every checkpointed node, with the WAL prefix
+ // already drained. Hashing the bytes on disk is what makes that upgrade boot.
+ //
+ // Emulated in the direction the migration runs: strip the trailing element off
+ // a current-shape file, which is what the pre-`client_table` build wrote.
+ const CHECKPOINT_OP: u64 = 42;
+ let encoded = IggySnapshot::new(CHECKPOINT_OP).encode().unwrap();
+ assert_eq!(
+ encoded[0] & 0xf0,
+ 0x90,
+ "snapshot must encode as a msgpack fixarray for this emulation"
+ );
+ assert_eq!(
+ *encoded.last().unwrap(),
+ 0xC0,
+ "the trailing snapshot field must encode as nil here; adjust the emulation \
+ if the last field stops being an Option"
+ );
+ let mut legacy = vec![0x90 | ((encoded[0] & 0x0f) - 1)];
+ legacy.extend_from_slice(&encoded[1..encoded.len() - 1]);
+
+ let decoded = IggySnapshot::decode(&legacy).unwrap();
+ assert_eq!(decoded.sequence_number(), CHECKPOINT_OP);
+ assert_ne!(
+ decoded.encode().unwrap(),
+ legacy,
+ "the emulated legacy file must NOT round-trip byte-identically, else this \
+ test cannot distinguish the two checksum sources"
+ );
+
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ std::fs::write(metadata_dir.join("snapshot.bin"), &legacy).unwrap();
+ let state = vsr_state_with_checkpoint(CHECKPOINT_OP, checkpoint_checksum(&legacy));
+ {
+ let superblock = PingPongSuperblock::open(&metadata_dir).await.unwrap();
+ superblock.write(&state.to_bytes()).await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await
+ .unwrap();
+ assert_eq!(recovered.snapshot_checkpoint.0, CHECKPOINT_OP);
+ assert_eq!(
+ recovered.snapshot_checkpoint.1,
+ checkpoint_checksum(&legacy),
+ "the verified pairing must be the checksum of the bytes on disk"
+ );
+ }
+
+ #[compio::test]
+ async fn recover_accepts_matching_superblock_identity() {
+ // The fence must not brick the healthy case it guards: the replica's own
+ // record boots, so the mismatch arm above is proving identity and not just
+ // that any record refuses.
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ let state = vsr_state_with_checkpoint(0, 0);
+ {
+ let superblock = PingPongSuperblock::open(&metadata_dir).await.unwrap();
+ superblock.write(&state.to_bytes()).await.unwrap();
+ }
+
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ CLUSTERED,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ CLIENTS_TABLE_MAX,
+ |_| {},
+ )
+ .await
+ .unwrap();
+ assert_eq!(recovered.recovered_state, Some(state));
}
}
diff --git a/core/metadata/src/lib.rs b/core/metadata/src/lib.rs
index 9f7af7df5..6533f7756 100644
--- a/core/metadata/src/lib.rs
+++ b/core/metadata/src/lib.rs
@@ -22,7 +22,9 @@ pub mod permissioner;
pub mod stm;
// Re-export IggyMetadata for use in other modules
-pub use impls::metadata::{BoundSession, CommitNotifier, IggyMetadata, MetadataSubmitError};
+pub use impls::metadata::{
+ BoundSession, CommitNotifier, IggyMetadata, MetadataSubmitError, apply_committed_prepare,
+};
// Re-export MuxStateMachine for use in other modules
pub use stm::mux::MuxStateMachine;
diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs
index afe9a7081..287244278 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -36,8 +36,10 @@ pub enum SnapshotError {
stage: PersistStage,
source: std::io::Error,
},
- /// Checksum mismatch on snapshot load.
- ChecksumMismatch { expected: u32, actual: u32 },
+ /// The snapshot's integrity trailer does not match its payload: a torn or
+ /// bit-rotted checkpoint. Refuse to restore from it rather than feed corrupt state
+ /// into the state machine.
+ ChecksumMismatch { expected: u128, actual: u128 },
/// Snapshot file is too short to contain a valid checksum.
Truncated { size: u64 },
}
@@ -71,7 +73,7 @@ impl fmt::Display for SnapshotError {
Self::ChecksumMismatch { expected, actual } => {
write!(
f,
- "snapshot checksum mismatch: expected {expected:#010x}, actual {actual:#010x}"
+ "snapshot checksum mismatch: expected {expected:#034x}, actual {actual:#034x}"
)
}
Self::Truncated { size } => {
@@ -103,6 +105,15 @@ impl From<std::io::Error> for SnapshotError {
/// The snapshot container for all metadata state machines.
/// Each field corresponds to one state machine's serialized state.
+///
+/// Serialized-form invariant: every collection here and in the nested `*Snapshot`
+/// types must keep a deterministic order (`Vec` or `BTreeMap`, never an unordered
+/// `HashMap`/`HashSet`/`AHashMap`). The checkpoint pairing hashes bytes on both sides
+/// (`impls::metadata::checkpoint_checksum`), so it no longer depends on this, but
+/// cross-replica state comparison and any future content-addressed transfer do: two
+/// replicas with identical state must serialize identically. Regression guards:
+/// `stream::tests::populated_streams_snapshot_reencode_is_byte_stable` and
+/// `impls::metadata::tests::populated_snapshot_reencode_and_checksum_are_stable`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetadataSnapshot {
/// Snapshot format version for forward/backward compatibility.
@@ -117,6 +128,12 @@ pub struct MetadataSnapshot {
/// Streams state machine snapshot data (consumer groups are nested inside
/// each topic).
pub streams: Option<StreamsSnapshot>,
+ /// Client-table state (sessions + at-most-once dedup replies) at the checkpoint
+ /// op. Not a state machine, but folded in so a restart that drained the WAL
+ /// prefix still recovers pre-checkpoint sessions. `#[serde(default)]` so a
+ /// snapshot predating this field decodes as `None`.
+ #[serde(default)]
+ pub client_table: Option<consensus::ClientTableSnapshot>,
}
impl Default for MetadataSnapshot {
@@ -140,6 +157,7 @@ impl MetadataSnapshot {
sequence_number,
users: None,
streams: None,
+ client_table: None,
}
}
@@ -317,6 +335,7 @@ mod tests {
assert_eq!(decoded.sequence_number, 42);
assert!(decoded.users.is_none());
assert!(decoded.streams.is_none());
+ assert!(decoded.client_table.is_none());
}
#[test]
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 5996bcff9..fa5f4d48f 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -1850,6 +1850,12 @@ impl StateHandler for DeletePartitionsRequest {
}
/// Snapshot representation for the Streams state machine.
+///
+/// Serialized-form invariant (see [`crate::stm::snapshot::MetadataSnapshot`]):
+/// `items` and the nested `topics` / `consumer_groups` / `partitions` stay ordered
+/// `Vec`s even though the runtime holds them in `AHashMap`s and a `Slab`. Swapping
+/// any back to an unordered map reorders on a decode and re-encode, breaking the
+/// checkpoint checksum cross-check recovery relies on.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamsSnapshot {
pub items: Vec<(usize, StreamSnapshot)>,
@@ -2048,6 +2054,7 @@ impl_fill_restore!(Streams, streams);
#[cfg(test)]
mod tests {
use super::*;
+ use crate::stm::snapshot::MetadataSnapshot;
use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireDecode;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
@@ -2100,6 +2107,85 @@ mod tests {
}
}
+ /// Regression guard for the [`StreamsSnapshot`] serialized-form invariant: a
+ /// populated snapshot must re-encode byte-identically after a decode, or the
+ /// checkpoint checksum cross-check (`recovery::verify_checkpoint_pairing`) would
+ /// diverge and refuse boot on a healthy node. Populates the three map-derived
+ /// `Vec`s (`items`, `topics`, `consumer_groups`) with two entries each, since one
+ /// entry cannot reorder and only >= 2 makes a regression observable.
+ #[test]
+ fn populated_streams_snapshot_reencode_is_byte_stable() {
+ let mut inner = StreamsInner::new();
+ for name in ["alpha", "beta"] {
+ create_stream(&mut inner, name);
+ }
+ // Streams are assigned ids 0, 1 in creation order; two topics per stream,
+ // two consumer groups per topic.
+ for stream_id in 0..2u32 {
+ for topic_name in ["logs", "events"] {
+ let create_topic = CreateTopicWithAssignmentsRequest {
+ request: make_topic_request(stream_id, 2, topic_name),
+ partitions: vec![
+ CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ },
+ CreatedPartitionAssignment {
+ partition_id: 1,
+ consensus_group_id: 2,
+ },
+ ],
+ };
+ let _ = StateHandler::apply(&create_topic, &mut inner, IggyTimestamp::now());
+ }
+ }
+ for stream_id in 0..2u32 {
+ for topic_id in 0..2u32 {
+ for group_name in ["cg-a", "cg-b"] {
+ let request = CreateConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream_id),
+ topic_id: WireIdentifier::numeric(topic_id),
+ name: WireName::new(group_name).unwrap(),
+ };
+ let _ = StateHandler::apply(&request, &mut inner, IggyTimestamp::now());
+ }
+ }
+ }
+ let streams: Streams = inner.into();
+
+ let mut snapshot = MetadataSnapshot::new(7);
+ snapshot.streams = Some(streams.to_snapshot());
+
+ // The tree really is populated, else a byte-stable empty snapshot would pass
+ // vacuously.
+ let streams_snapshot = snapshot.streams.as_ref().unwrap();
+ assert_eq!(streams_snapshot.items.len(), 2, "two streams");
+ let (_, first_stream) = &streams_snapshot.items[0];
+ assert_eq!(
+ first_stream.topics.len(),
+ 2,
+ "two topics in the first stream"
+ );
+ let (_, first_topic) = &first_stream.topics[0];
+ assert_eq!(
+ first_topic.consumer_groups.len(),
+ 2,
+ "two consumer groups in the first topic"
+ );
+
+ let encoded = snapshot.encode().unwrap();
+ let reencoded = MetadataSnapshot::decode(&encoded)
+ .unwrap()
+ .encode()
+ .unwrap();
+ assert_eq!(
+ encoded, reencoded,
+ "a populated snapshot must re-encode byte-identically after a decode; an \
+ unordered collection would reorder and break the checkpoint checksum \
+ cross-check, refusing boot on a healthy node"
+ );
+ }
+
#[test]
fn current_partition_count_scans_existing_topic_state() {
let mut inner = StreamsInner::new();
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index c8ea95f9d..3725696b4 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -731,6 +731,11 @@ pub struct PermissionerSnapshot {
}
/// Snapshot representation for the Users state machine.
+///
+/// Serialized-form invariant (see [`crate::stm::snapshot::MetadataSnapshot`]):
+/// `items`, `personal_access_tokens`, and the permissioner's maps stay ordered
+/// (`Vec` / `BTreeMap`) even though the runtime holds them in `AHashMap`s. Swapping
+/// any to an unordered map breaks the checkpoint checksum cross-check.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UsersSnapshot {
pub items: Vec<(usize, UserSnapshot)>,
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index d99815414..45572ef55 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -35,11 +35,13 @@ use configs::ng_sharding::{
};
use configs::server_ng::{NgSystemConfig, ServerNgConfig};
use consensus::{
- LocalPipeline, MetadataHandle, PartitionsHandle, PipelineEntry, Sequencer, VsrConsensus,
+ ClientTable, LocalPipeline, MetadataHandle, PartitionsHandle, PipelineEntry, Sequencer,
+ VsrConsensus,
};
// `try_send` / `try_recv` resolve through these traits on `MAsyncTx` /
// `MAsyncRx`; the metadata-handoff loops below depend on the
// non-blocking variants for cancel-safe shutdown polling.
+use consensus::VsrState;
use crossfire::{AsyncRxTrait, AsyncTxTrait};
use iggy_binary_protocol::{Operation, PrepareHeader};
use iggy_common::defaults::{
@@ -48,6 +50,7 @@ use iggy_common::defaults::{
};
use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyByteSize, PartitionStats, variadic};
use journal::prepare_journal::PrepareJournal;
+use journal::superblock::{DynSuperblockStore, PingPongSuperblock};
use journal::{Journal, JournalHandle};
use message_bus::client_listener::{self, RequestHandler};
use message_bus::installer;
@@ -69,7 +72,7 @@ use message_bus::{
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
-use metadata::impls::recovery::recover;
+use metadata::impls::recovery::{ReplicaIdentity, recover};
use metadata::stm::mux::WithFactory;
use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
@@ -937,7 +940,11 @@ async fn shard_main(
// shifts one slab id and root is lost after the first restart.
let recovered = recover::<ServerNgMuxStateMachine>(
data_dir,
- topology.replica_count == 1,
+ ReplicaIdentity {
+ cluster: topology.cluster_id,
+ replica_id: topology.self_replica_id,
+ replica_count: topology.replica_count,
+ },
config.metadata.journal_slots,
config.metadata.clients_table_max,
|mux_stm| {
@@ -975,13 +982,16 @@ async fn shard_main(
.await?;
(
recovered.mux_stm,
- Some((
- recovered.journal,
- recovered.snapshot,
- recovered.last_applied_op,
- recovered.last_journaled_op,
- recovered.client_table,
- )),
+ Some(RecoveredOwnerState {
+ journal: recovered.journal,
+ snapshot: recovered.snapshot,
+ last_applied_op: recovered.last_applied_op,
+ last_journaled_op: recovered.last_journaled_op,
+ client_table: recovered.client_table,
+ superblock: recovered.superblock,
+ recovered_state: recovered.recovered_state,
+ snapshot_checkpoint: recovered.snapshot_checkpoint,
+ }),
)
}
MetadataHandoff::Waiter { bundle_rx } => {
@@ -999,42 +1009,37 @@ async fn shard_main(
// Metadata consensus + journal + snapshot live only on shard 0.
// `IggyShard::tick_metadata` short-circuits when `consensus.is_none()`,
// so peer shards have no caller that reads `journal` or `snapshot`.
- let (metadata_consensus, journal_for_metadata, snapshot_for_metadata, recovered_client_table) =
- if let Some((journal, snapshot, last_applied_op, last_journaled_op, client_table)) =
- owner_state
- {
- let snapshot_floor = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number);
- let commit_watermark = last_applied_op.unwrap_or(snapshot_floor);
- let restored_op = last_journaled_op.unwrap_or(snapshot_floor);
- let consensus = restore_metadata_consensus(
- &journal,
- restored_op,
- commit_watermark,
- topology.cluster_id,
- topology.self_replica_id,
- topology.replica_count,
- Rc::clone(&bus),
- config.metadata.prepare_queue_depth,
- cluster_heartbeat_ticks(config),
- commit_broadcast_ticks(config),
- prepare_retransmit_ticks(config),
- view_change_retransmit_ticks(config),
- view_change_status_ticks(config),
- request_start_view_ticks(config),
- config.cluster.view_probe_attempts_max,
- recovery_barrier_deadline(
- config.cluster.heartbeat_timeout.get_duration(),
- config.cluster.view_change_status_timeout.get_duration(),
- ),
- );
- (Some(consensus), Some(journal), snapshot, Some(client_table))
- } else {
- (None, None, None, None)
- };
+ let (
+ metadata_consensus,
+ journal_for_metadata,
+ snapshot_for_metadata,
+ superblock_for_metadata,
+ checkpoint_seed,
+ recovered_client_table,
+ ) = if let Some(owner) = owner_state {
+ // `recover()` already opened the superblock, read `recovered_state`, and
+ // verified the on-disk snapshot against its checkpoint pairing BEFORE decoding
+ // it. Reuse that superblock rather than re-opening it, which would fork the
+ // ping-pong sequence counter. Consensus recovers its true (view, log_view)
+ // from `recovered_state` instead of inferring a stale view from the WAL.
+ let consensus = restore_metadata_consensus(&owner, &topology, config, Rc::clone(&bus));
+ let superblock: Rc<dyn DynSuperblockStore> = Rc::new(owner.superblock);
+ (
+ Some(consensus),
+ Some(owner.journal),
+ owner.snapshot,
+ Some(superblock),
+ owner.snapshot_checkpoint,
+ Some(owner.client_table),
+ )
+ } else {
+ (None, None, None, None, (0, 0), None)
+ };
let metadata = ServerNgMetadata::new(
metadata_consensus,
journal_for_metadata,
snapshot_for_metadata,
+ superblock_for_metadata,
mux_stm,
Some(PathBuf::from(&config.system.path)),
);
@@ -1043,15 +1048,20 @@ async fn shard_main(
// table from scratch, so running it afterwards would drop every resumed
// session (and trip its empty-table assert).
metadata.set_clients_table_max(config.metadata.clients_table_max);
- // Reinstall the sessions the WAL replay rebuilt, so a rebooted node
- // dedups retries and admits continuations from clients that kept their
- // identity across the restart (IGGY-137). Recovery sized this table from
- // the same config value, so the install preserves the configured cap.
+ // Reinstall the sessions recovery restored from the checkpoint and the WAL
+ // suffix, so a rebooted node dedups retries and admits continuations from
+ // clients that kept their identity across the restart (IGGY-137). Recovery
+ // sized this table from the same config value, so the install preserves the
+ // configured cap.
if let Some(client_table) = recovered_client_table {
// Refusal (a client registered before this ran) keeps the live table
// and is logged by the callee; boot continues either way.
let _ = metadata.install_client_table(client_table);
}
+ // Seed the coordinator's last-checkpoint pairing so the first post-boot
+ // view-change superblock write records the real (checkpoint_op, checksum)
+ // instead of (0, 0). No-op on peer shards, which have no coordinator.
+ metadata.seed_checkpoint_ref(checkpoint_seed.0, checkpoint_seed.1);
// Shard 0's copy resolves the `ServerDefault` sentinels (max topic size and
// message expiry) at admission; every shard's copy backs the same resolution in responses.
metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64());
@@ -1908,28 +1918,52 @@ pub(crate) fn repair_retry_ticks(config: &ServerNgConfig) -> u32 {
.unwrap_or(u32::MAX)
}
-#[allow(clippy::too_many_arguments)]
+/// Shard 0's half of a metadata recovery: everything [`recover`] produced except the
+/// state machine, which every shard receives through the factory bundle.
+///
+/// Named rather than a positional tuple: the fields are same-typed `Option<u64>`s and
+/// `(u64, u128)` pairs that a reorder would silently rebind, and one of them decides
+/// what view the replica boots into.
+struct RecoveredOwnerState {
+ journal: PrepareJournal,
+ snapshot: Option<IggySnapshot>,
+ last_applied_op: Option<u64>,
+ last_journaled_op: Option<u64>,
+ client_table: ClientTable,
+ superblock: PingPongSuperblock,
+ recovered_state: Option<VsrState>,
+ snapshot_checkpoint: (u64, u128),
+}
+
+/// Rebuild metadata consensus from what recovery read off this replica's own disk.
+///
+/// Takes the recovery result, topology and config whole rather than the dozen-plus
+/// scalars it needs from them: most were `u64` tick counts, where a misordered
+/// argument type-checks and mistunes a timeout silently.
fn restore_metadata_consensus(
- journal: &PrepareJournal,
- restored_op: u64,
- commit_watermark: u64,
- cluster_id: u128,
- self_replica_id: u8,
- replica_count: u8,
+ owner: &RecoveredOwnerState,
+ topology: &TcpTopology,
+ config: &ServerNgConfig,
bus: Rc<IggyMessageBus>,
- prepare_queue_depth: usize,
- normal_heartbeat_ticks: u64,
- commit_message_ticks: u64,
- prepare_ticks: u64,
- view_change_retransmit_ticks: u64,
- view_change_status_ticks: u64,
- request_start_view_ticks: u64,
- probe_attempts_max: u32,
- recovery_deadline: Duration,
) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let journal = &owner.journal;
+ let replica_count = topology.replica_count;
+ let recovered_state = owner.recovered_state;
+ let snapshot_floor = owner
+ .snapshot
+ .as_ref()
+ .map_or(0, IggySnapshot::sequence_number);
+ let commit_watermark = owner.last_applied_op.unwrap_or(snapshot_floor);
+ let restored_op = owner.last_journaled_op.unwrap_or(snapshot_floor);
+ let recovery_deadline = recovery_barrier_deadline(
+ config.cluster.heartbeat_timeout.get_duration(),
+ config.cluster.view_change_status_timeout.get_duration(),
+ );
+ let prepare_queue_depth = config.metadata.prepare_queue_depth;
+
let mut consensus = VsrConsensus::new(
- cluster_id,
- self_replica_id,
+ topology.cluster_id,
+ topology.self_replica_id,
replica_count,
server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
bus,
@@ -1938,34 +1972,56 @@ fn restore_metadata_consensus(
// in-flight prepares and drain as prepares commit.
LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2),
);
- consensus.set_normal_heartbeat_ticks(normal_heartbeat_ticks);
- consensus.set_commit_message_ticks(commit_message_ticks);
- consensus.set_prepare_ticks(prepare_ticks);
- consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks);
- consensus.set_view_change_status_ticks(view_change_status_ticks);
- consensus.set_request_start_view_ticks(request_start_view_ticks);
- consensus.set_probe_attempts_max(probe_attempts_max);
+ consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config));
+ consensus.set_commit_message_ticks(commit_broadcast_ticks(config));
+ consensus.set_prepare_ticks(prepare_retransmit_ticks(config));
+ consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks(config));
+ consensus.set_view_change_status_ticks(view_change_status_ticks(config));
+ consensus.set_request_start_view_ticks(request_start_view_ticks(config));
+ consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max);
+ // Fresh random incarnation each boot, so a StartView addressed to a previous
+ // incarnation still in flight is ignored (`handle_start_view` guard). `| 1`
+ // guarantees the non-zero the guard treats as set. The deterministic simulator
+ // overrides this with a seed-derived value bumped per restart.
+ consensus.set_incarnation(rand::random::<u128>() | 1);
let last_header = journal
.last_op()
.and_then(|op| usize::try_from(op).ok())
.and_then(|op| journal.header(op).map(|header| *header));
- if let Some(header) = last_header {
+ // View and log_view come from the durable superblock when present. A present but
+ // unreadable superblock already refused boot in `recover()`, so reaching the
+ // `else` means it is genuinely absent: a fresh node, or one that took writes but
+ // never checkpointed or changed view. There, inferring the view from the last WAL
+ // prepare is safe, since the persist-before-send gate guarantees this replica
+ // never externalized a view beyond what a re-probe re-derives, and it re-probes
+ // as a backup below. log_view cannot be inferred and stays 0 until the next
+ // superblock write.
+ if let Some(state) = recovered_state {
+ consensus.set_view(state.view);
+ consensus.set_log_view(state.log_view);
+ consensus.mark_superblock_durable(state.view, state.log_view);
+ } else if let Some(header) = last_header {
consensus.set_view(header.view);
}
- // On a RESTART in a cluster (a non-empty WAL proves a prior life),
- // rejoin as a quorum-invisible backup and probe for the current view
- // (`RequestStartView`): the view's primary answers with a `StartView`,
- // the replica adopts it as a backup, and journal repair fills any WAL
- // gap. A probing replica never resumes primaryship -- if this replica
- // IS the current primary-by-index, its probe makes the backups elect
- // past it.
+ // On a RESTART in a cluster, rejoin as a quorum-invisible backup and
+ // probe for the current view (`RequestStartView`): the view's primary
+ // answers with a `StartView`, the replica adopts it as a backup, and
+ // journal repair fills any WAL gap. A probing replica never resumes
+ // primaryship -- if this replica IS the current primary-by-index, its
+ // probe makes the backups elect past it.
// The probe re-broadcasts on its timeout, so it needs no live mesh at
- // boot. A FRESH boot (empty WAL) keeps the plain init: the cluster
- // needs its view-0 primary to exist, and a single-replica cluster has
- // no peer to ask.
- if replica_count > 1 && restored_op > 0 {
+ // boot. A FRESH boot keeps the plain init: the cluster needs its view-0
+ // primary to exist, and a single-replica cluster has no peer to ask.
+ //
+ // Prior life is EITHER a non-empty WAL or a recovered superblock. A view
+ // change persists without touching the WAL, so a replica that changed
+ // view before its first metadata write comes back with a non-zero view
+ // and an empty journal; gating on the WAL alone would `init()` it into
+ // `Status::Normal` as primary for a view the cluster may have moved past,
+ // with `ceded_primaryship` false and no probe to correct it.
+ if replica_count > 1 && (restored_op > 0 || recovered_state.is_some()) {
consensus.init_as_backup();
consensus.begin_view_probe();
} else {
@@ -2039,7 +2095,7 @@ fn restore_metadata_consensus(
break;
};
let mut entry = PipelineEntry::new(*header);
- entry.add_ack(self_replica_id);
+ entry.add_ack(topology.self_replica_id);
pipeline.push(entry);
}
}
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 3d0179b05..7209e8a6b 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -3036,6 +3036,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3156,7 +3157,7 @@ mod tests {
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
- let metadata = IggyMetadata::new(None, None, None, TestMux::default(), None);
+ let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
@@ -3279,7 +3280,7 @@ mod tests {
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
- let metadata = IggyMetadata::new(None, None, None, TestMux::default(), None);
+ let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
@@ -3341,7 +3342,7 @@ mod tests {
const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
let bus = SpyBus::default();
- let metadata = IggyMetadata::new(None, None, None, TestMux::default(), None);
+ let metadata = IggyMetadata::new(None, None, None, None, TestMux::default(), None);
let partitions = IggyPartitions::new(
ShardId::new(0),
PartitionsConfig {
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index da549bb1b..2ca7b9baf 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -1110,7 +1110,7 @@ mod tests {
journal::prepare_journal::PrepareJournal,
IggySnapshot,
_,
- > = IggyMetadata::new(None, None, None, mux, None);
+ > = IggyMetadata::new(None, None, None, None, mux, None);
let partitions = IggyPartitions::new(
ShardId::new(shard_id),
PartitionsConfig {
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 835308e6d..1189ff897 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -55,6 +55,11 @@ use metadata::stm::StateMachine;
use metadata::{BoundSession, MetadataSubmitError};
use partitions::{IggyPartition, IggyPartitions, PollFragments, PollingArgs, PollingConsumer};
use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+// Read only by the durable-before-send tripwire, which is `debug_assertions`-only, so
+// an unconditional import warns in release builds. CI's `-D warnings` rides clippy,
+// which builds debug, so that warning goes unobserved there.
+#[cfg(debug_assertions)]
+use server_common::sharding::METADATA_CONSENSUS_NAMESPACE;
use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen};
use shards_table::ShardsTable;
use std::cell::{Cell, RefCell};
@@ -1924,7 +1929,9 @@ where
&& consensus.namespace() == header.namespace
{
let actions = consensus.handle_start_view_change(PlaneKind::Metadata, &header);
- dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ if planes.0.persist_superblock_if_needed(consensus).await {
+ dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ }
return;
}
@@ -1967,7 +1974,9 @@ where
&& consensus.namespace() == header.namespace
{
let actions = consensus.handle_do_view_change(PlaneKind::Metadata, &header);
- dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ if planes.0.persist_superblock_if_needed(consensus).await {
+ dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ }
if actions
.iter()
.any(|action| matches!(action, VsrAction::CommitJournal))
@@ -2023,7 +2032,9 @@ where
&& consensus.namespace() == header.namespace
{
let actions = consensus.handle_start_view(PlaneKind::Metadata, &header);
- dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ if planes.0.persist_superblock_if_needed(consensus).await {
+ dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ }
// `dispatch_vsr_actions` deliberately no-ops `CommitJournal` (it
// needs the plane); without this the ops a StartView marks
// committed stay journaled-but-unapplied forever, because the
@@ -2152,7 +2163,13 @@ where
match consensus.handle_commit(&header) {
CommitOutcome::Advanced => planes.0.commit_journal().await,
CommitOutcome::RespondStartView => {
- respond_start_view::<B, _, MJ>(consensus).await;
+ // Durable-before-send: the StartView advertises this replica's
+ // current view, so persist before answering, as the view-change
+ // dispatch gate does. Withhold on failure; the stale peer keeps
+ // heartbeating, so it re-triggers once the tick persists.
+ if planes.0.persist_superblock_if_needed(consensus).await {
+ respond_start_view::<B, _, MJ>(consensus).await;
+ }
}
CommitOutcome::Accepted => {}
}
@@ -2173,6 +2190,9 @@ where
match consensus.handle_commit(&header) {
CommitOutcome::Advanced => partition.commit_journal(config).await,
CommitOutcome::RespondStartView => {
+ // Partition consensus is not superblock-durable yet, so there is no
+ // view to persist before answering here; the metadata arm above
+ // gates its StartView on the durable view.
respond_start_view::<B, _, MJ>(consensus).await;
}
CommitOutcome::Accepted => {}
@@ -2200,7 +2220,9 @@ where
&& consensus.namespace() == header.namespace
{
let actions = consensus.handle_request_start_view(PlaneKind::Metadata, &header);
- dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ if planes.0.persist_superblock_if_needed(consensus).await {
+ dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await;
+ }
return;
}
let Some(partition) = planes
@@ -2874,7 +2896,9 @@ where
let actions = consensus.tick(PlaneKind::Metadata);
- dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &actions).await;
+ if metadata.persist_superblock_if_needed(consensus).await {
+ dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &actions).await;
+ }
// Repair a lost primary self-ack: `RetransmitPrepares` to self is a
// no-op, so the timer-driven retransmit above cannot recover the
@@ -2954,10 +2978,16 @@ where
namespace = consensus.namespace(),
"answering stale-view heartbeat with StartView"
);
+ // Unsolicited, answering a stale-view heartbeat rather than a probe, so there is
+ // no incarnation to echo; freshness comes from the receiver's view checks. Sent to
+ // every backup, since a replica heartbeating an older view has peers that missed
+ // the view change with it.
let action = VsrAction::SendStartView {
view: consensus.view(),
op: consensus.sequencer().current_sequence(),
commit: consensus.commit_max(),
+ incarnation: 0,
+ target: None,
namespace: consensus.namespace(),
};
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
@@ -3024,6 +3054,41 @@ async fn dispatch_vsr_actions<B, P, J>(
}
};
+ // Centralized durable-before-send tripwire: a view-scoped message must never
+ // advertise a (view, log_view) the superblock has not recorded, or a crash could
+ // recover an older view than one a peer already saw, splitting the brain or
+ // losing a commit. Every metadata caller persists first (the view-change dispatch
+ // sites and the on_replicate / on_commit send gates), so this asserts they did
+ // rather than letting a future bypass through silently. Metadata plane only:
+ // partition consensus has no superblock to record a view in, so it is exempt and
+ // the namespace test below is what does the work (its `needs_superblock_persist`
+ // is not a stand-in: that predicate reads clean at view 0, which is where a
+ // partition group spends most of its life). `RequestStartView` is exempt too,
+ // being a probe that asks to LEARN the view rather than advertise it.
+ #[cfg(debug_assertions)]
+ if consensus.namespace() == METADATA_CONSENSUS_NAMESPACE {
+ for action in actions {
+ let advertises_view = matches!(
+ action,
+ VsrAction::SendStartViewChange { .. }
+ | VsrAction::SendDoViewChange { .. }
+ | VsrAction::SendStartView { .. }
+ | VsrAction::SendPrepareOk { .. }
+ // A backup drops a Commit whose view differs from its own, and a
+ // primary answers an older-view one with a StartView, so the
+ // heartbeat advertises a view like the rest. Gated today only
+ // because its sole emitter rides the tick, which persists first.
+ | VsrAction::SendCommit { .. }
+ );
+ debug_assert!(
+ !advertises_view || !consensus.needs_superblock_persist(),
+ "durable-before-send violated: dispatching a view-scoped metadata action \
+ while the superblock is behind the in-memory view {}",
+ consensus.view(),
+ );
+ }
+ }
+
for action in actions {
match action {
VsrAction::SendStartViewChange { view, namespace } => {
@@ -3061,6 +3126,9 @@ async fn dispatch_vsr_actions<B, P, J>(
send(*target, msg.into_generic().into_frozen()).await;
}
VsrAction::SendRequestStartView { view, namespace } => {
+ // Stamp this replica's incarnation so the answering StartView can
+ // echo it, proving to us the reply post-dates our restart.
+ let incarnation = consensus.incarnation();
let msg =
Message::<RequestStartViewHeader>::new(size_of::<RequestStartViewHeader>())
.transmute_header(|_, h: &mut RequestStartViewHeader| {
@@ -3068,6 +3136,7 @@ async fn dispatch_vsr_actions<B, P, J>(
h.cluster = cluster;
h.replica = self_id;
h.view = *view;
+ h.incarnation = incarnation;
h.namespace = *namespace;
h.size = size_of::<RequestStartViewHeader>() as u32;
});
@@ -3077,6 +3146,8 @@ async fn dispatch_vsr_actions<B, P, J>(
view,
op,
commit,
+ incarnation,
+ target,
namespace,
} => {
let msg = Message::<StartViewHeader>::new(size_of::<StartViewHeader>())
@@ -3087,10 +3158,19 @@ async fn dispatch_vsr_actions<B, P, J>(
h.view = *view;
h.op = *op;
h.commit = *commit;
+ h.incarnation = *incarnation;
h.namespace = *namespace;
h.size = size_of::<StartViewHeader>() as u32;
});
- broadcast(msg.into_generic().into_frozen()).await;
+ let frozen = msg.into_generic().into_frozen();
+ // A probe echo is addressed to its requester: the incarnation it
+ // carries is that replica's freshness proof, and a peer recovering
+ // at the same time would read it as foreign and reject a current
+ // StartView.
+ match target {
+ Some(replica) => send(*replica, frozen).await,
+ None => broadcast(frozen).await,
+ }
}
VsrAction::SendPrepareOk {
view,
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index b0035d49f..b557d8bdc 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -19,6 +19,7 @@ use crate::executor::TimerHandle;
use clock::Clock;
use iggy_binary_protocol::PrepareHeader;
use iggy_common::{IggyTimestamp, variadic};
+use journal::superblock::{SuperblockContents, SuperblockStore};
use journal::{Journal, JournalHandle, Storage};
use metadata::MuxStateMachine;
use metadata::stm::stream::Streams;
@@ -99,6 +100,10 @@ pub struct SimJournal<S: Storage> {
headers: UnsafeCell<HashMap<u64, PrepareHeader>>,
offsets: UnsafeCell<HashMap<u64, usize>>,
write_offset: Cell<usize>,
+ /// Highest op appended, `None` when empty. Tracked so a restart reads the
+ /// retained head in O(1) without scanning `headers` (see
+ /// [`SimJournal::last_op`]).
+ last_op: Cell<Option<u64>>,
/// Debug-only single-accessor tripwire. `entry` / `append` hold a
/// [`JournalAccessGuard`] across their whole body, including the storage
/// `.await`, so if a suspending storage tier ever let a second task touch
@@ -115,6 +120,7 @@ impl<S: Storage + Default> Default for SimJournal<S> {
headers: UnsafeCell::new(HashMap::new()),
offsets: UnsafeCell::new(HashMap::new()),
write_offset: Cell::new(0),
+ last_op: Cell::new(None),
#[cfg(debug_assertions)]
accessing: Cell::new(false),
}
@@ -224,6 +230,8 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for SimJournal<S> {
unsafe { &mut *self.headers.get() }.insert(header.op, header);
unsafe { &mut *self.offsets.get() }.insert(header.op, offset);
self.write_offset.set(offset + bytes_written);
+ let head = self.last_op.get().map_or(header.op, |op| op.max(header.op));
+ self.last_op.set(Some(head));
Ok(())
}
@@ -242,8 +250,121 @@ impl JournalHandle for SimJournal<MemStorage> {
}
}
+impl SimJournal<MemStorage> {
+ /// Highest op appended, `None` when empty. Survives a restart because the
+ /// harness retains the whole journal behind an `Rc`.
+ #[must_use]
+ pub const fn last_op(&self) -> Option<u64> {
+ self.last_op.get()
+ }
+
+ /// The committed watermark to restore after a restart, mirroring
+ /// `metadata::recover`. On a solo cluster every appended op commits the instant
+ /// it is durable, so the head IS the commit point; otherwise the highest
+ /// `commit` any journaled prepare stamped, a lower bound, since a prepare
+ /// records the primary's commit point at send time, so the true point may be one
+ /// op higher and re-commits on rejoin.
+ #[must_use]
+ pub fn recovery_commit_watermark(&self, solo: bool) -> u64 {
+ if solo {
+ return self.last_op.get().unwrap_or(0);
+ }
+ let headers = unsafe { &*self.headers.get() };
+ headers
+ .values()
+ .map(|header| header.commit)
+ .max()
+ .unwrap_or(0)
+ }
+
+ /// The head prepare's header, `None` when empty. Restores the last-prepare
+ /// checksum (hash-chain continuity) and the prepare-timestamp floor at restart,
+ /// mirroring `restore_metadata_consensus`.
+ #[must_use]
+ pub fn last_header(&self) -> Option<PrepareHeader> {
+ let head = self.last_op.get()?;
+ let headers = unsafe { &*self.headers.get() };
+ headers.get(&head).copied()
+ }
+
+ /// Read the entry at `op` synchronously, for off-executor WAL replay at restart.
+ /// Mirrors [`Journal::entry`] without the never-suspending `MemStorage` await, so
+ /// it needs no [`JournalAccessGuard`]: a synchronous read offers no suspension
+ /// point for another task to interleave on.
+ ///
+ /// # Panics
+ /// If the bytes at `op` do not decode as a prepare message, meaning the retained
+ /// WAL is corrupt, which is a harness bug since the sim has no torn writes.
+ #[must_use]
+ pub fn entry_sync(&self, op: u64) -> Option<Message<PrepareHeader>> {
+ let headers = unsafe { &*self.headers.get() };
+ let offsets = unsafe { &*self.offsets.get() };
+ let header = headers.get(&op)?;
+ let offset = *offsets.get(&op)?;
+ let data = self.storage.data.borrow();
+ let end = offset.checked_add(header.size as usize)?;
+ let buffer = data.get(offset..end)?.to_vec();
+ let message = Message::try_from(Owned::<4096>::copy_from_slice(&buffer))
+ .expect("prepare buffer must contain a valid prepare message");
+ Some(message)
+ }
+}
+
#[derive(Debug, Default)]
pub struct SimSnapshot {}
+/// In-memory superblock for the simulator.
+///
+/// Held by the harness behind an `Rc` so its bytes survive a replica being dropped
+/// and rebuilt across a restart. RAM has no torn writes, so it holds only the latest
+/// payload, with none of the framing or checksum `PingPongSuperblock` adds.
+#[derive(Debug, Default)]
+pub struct SimSuperblock {
+ latest: RefCell<Option<Vec<u8>>>,
+ /// Fault injection: when set, every [`SuperblockStore::write`] errors and
+ /// persists nothing, so `persist_superblock_if_needed` returns `false` and the
+ /// shard withholds the view-scoped send. Proves the split-brain gate, that a
+ /// replica never sends in a view it has not durably recorded.
+ fail_writes: Cell<bool>,
+}
+
+impl SimSuperblock {
+ /// Latest persisted payload, read synchronously. The harness reads this off the
+ /// executor at restart, so no `block_on` is needed.
+ #[must_use]
+ pub fn read_latest_sync(&self) -> Option<Vec<u8>> {
+ self.latest.borrow().clone()
+ }
+
+ /// Make every subsequent write fail, a persistent write fault, so the durability
+ /// gate withholds view-scoped sends. See [`Self::fail_writes`].
+ pub fn set_fail_writes(&self) {
+ self.fail_writes.set(true);
+ }
+}
+
+#[allow(clippy::future_not_send)]
+impl SuperblockStore for SimSuperblock {
+ async fn write(&self, payload: &[u8]) -> std::io::Result<()> {
+ if self.fail_writes.get() {
+ return Err(std::io::Error::other("sim superblock write fault"));
+ }
+ *self.latest.borrow_mut() = Some(payload.to_vec());
+ Ok(())
+ }
+
+ async fn read_latest(&self) -> std::io::Result<SuperblockContents> {
+ // RAM has no torn writes and the sim never injects an unreadable record, so
+ // the payload is either present or was never written.
+ Ok(self
+ .latest
+ .borrow()
+ .as_ref()
+ .map_or(SuperblockContents::Empty, |payload| {
+ SuperblockContents::Present(payload.clone())
+ }))
+ }
+}
+
/// Type alias for simulator state machine
pub type SimMuxStateMachine = MuxStateMachine<variadic!(Users, Streams)>;
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 8301a1ecb..115ea7bdd 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -27,11 +27,14 @@ pub mod workload;
use bus::SimOutbox;
use client::SimClient;
-use consensus::{ConsensusClock, MetadataHandle, PartitionsHandle};
+use consensus::{ConsensusClock, MetadataHandle, PartitionsHandle, VsrState};
use deps::SimClock;
+use deps::SimSuperblock;
+use deps::{MemStorage, SimJournal};
use executor::{DetExecutor, RunOutcome, TaskId};
use iggy_binary_protocol::{GenericHeader, ReplyHeader};
use iggy_common::IggyError;
+use journal::superblock::DynSuperblockStore;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use metadata::impls::metadata::StreamsFrontend;
use network::Network;
@@ -70,6 +73,19 @@ pub const ENTRY_SHARD_SEED_SALT: u64 = 0x5A1A_F0E5_FACE_0003;
pub struct SimReplica {
/// Shards of this replica, indexed by shard id.
pub shards: Vec<Rc<Replica>>,
+ /// Shard 0's durable superblock, held here rather than inside the shard so its
+ /// bytes survive the shard being dropped and rebuilt across a restart.
+ pub superblock: Rc<SimSuperblock>,
+ /// Shard 0's metadata WAL, held here for the same reason as the superblock: the
+ /// bytes and index survive a restart, so a rebuilt replica recovers its
+ /// op/commit and committed metadata from its own disk. Only shard 0 owns
+ /// metadata consensus, so this is the single retained journal.
+ pub metadata_journal: Rc<SimJournal<MemStorage>>,
+ /// Shard 0's metadata consensus incarnation nonce. Harness-owned: a seed-derived
+ /// value bumped by one on each restart, so successive incarnations are distinct
+ /// yet the run stays byte-identical on replay. See
+ /// `VsrConsensus::set_incarnation`.
+ pub metadata_incarnation: u128,
/// Keeps each pump's stop channel alive; dropping one would end that
/// pump gracefully, which is reserved for future shutdown/restart
/// tests (crash uses `DetExecutor::abort` instead).
@@ -183,7 +199,7 @@ impl Simulator {
)
}
- #[allow(clippy::cast_possible_truncation)]
+ #[allow(clippy::cast_possible_truncation, clippy::too_many_lines)]
fn build(
replica_count: usize,
shards_per_replica: u16,
@@ -234,6 +250,15 @@ impl Simulator {
bus.add_replica(j);
}
let outbox = Rc::new(bus);
+ // Harness-owned so they outlive a replica restart, which drops and
+ // rebuilds the shards: the superblock's VSR state and the metadata WAL
+ // both survive.
+ let superblock = Rc::new(SimSuperblock::default());
+ let metadata_journal = Rc::new(SimJournal::<MemStorage>::default());
+ // Initial incarnation, spread across the 128-bit space per replica so the
+ // restart increments in `replica_restart` never collide across replicas.
+ // Non-zero.
+ let metadata_incarnation = 1 + (u128::from(id) << 64);
// One crossfire mesh per replica; every shard gets a clone of
// the canonical senders vec and exclusively takes its inbox.
@@ -253,6 +278,15 @@ impl Simulator {
let inbox = inboxes[usize::from(shard_idx)]
.take()
.expect("mesh yields exactly one inbox per shard");
+ // Only shard 0 owns metadata consensus, so only it carries the
+ // superblock. Peer shards persist nothing.
+ let shard_superblock: Option<Rc<dyn DynSuperblockStore>> = if shard_idx == 0 {
+ let sb: Rc<dyn DynSuperblockStore> = superblock.clone();
+ Some(sb)
+ } else {
+ None
+ };
+ let shard_journal = (shard_idx == 0).then(|| Rc::clone(&metadata_journal));
let (shard, writer_bundle) = new_shard(
id,
shard_idx,
@@ -264,6 +298,10 @@ impl Simulator {
consensus_clock.clone(),
shell,
metadata_bundle.clone(),
+ shard_superblock,
+ shard_journal,
+ None, // fresh boot: no recovered VSR state
+ metadata_incarnation,
);
if shard_idx == 0 {
metadata_bundle = Some(
@@ -286,6 +324,9 @@ impl Simulator {
replicas.push(SimReplica {
shards,
+ superblock,
+ metadata_journal,
+ metadata_incarnation,
_stop_txs: stop_txs,
pump_tasks,
});
@@ -642,10 +683,10 @@ impl Simulator {
// Sessions/dedup/eviction live on metadata only (IggyMetadata).
}
- /// Crash a replica: abort its pump tasks, disable its network links,
- /// and discard its outbox. The replica object stays alive but receives
- /// no messages until a `replica_restart` (not yet implemented; needs
- /// consensus durability).
+ /// Crash a replica: abort its pump tasks, disable its network links, and discard
+ /// its outbox. The replica object stays alive but receives no messages; a
+ /// following [`Self::replica_restart`] drops and rebuilds it from the durable
+ /// superblock, which is when volatile state is actually lost and recovered.
///
/// # Panics
/// If the replica is already crashed.
@@ -683,6 +724,106 @@ impl Simulator {
self.crashed.contains(&replica_index)
}
+ /// Restart a crashed replica: drop its shards, losing all volatile consensus
+ /// state as a real restart does, and rebuild them against the retained
+ /// superblock, recovering `(view, log_view)` from disk exactly as production's
+ /// `restore_metadata_consensus` does. The superblock and outbox are
+ /// harness-owned, so they survive the drop; a fresh inter-shard mesh and pump
+ /// tasks are wired, and the network is re-enabled.
+ ///
+ /// # Panics
+ /// If the replica is not crashed, or if its shard count does not fit `u16`,
+ /// impossible since mesh construction caps it.
+ pub fn replica_restart(&mut self, replica_index: u8) {
+ assert!(
+ self.crashed.contains(&replica_index),
+ "cannot restart replica {replica_index}: not crashed"
+ );
+ let idx = replica_index as usize;
+ let shards_per_replica =
+ u16::try_from(self.replicas[idx].shards.len()).expect("shard count fits u16");
+ let superblock = Rc::clone(&self.replicas[idx].superblock);
+ // The metadata WAL is harness-owned too, so its bytes and index survive the
+ // drop: the rebuilt shard 0 recovers op/commit and committed state from it,
+ // not from an empty journal.
+ let metadata_journal = Rc::clone(&self.replicas[idx].metadata_journal);
+ // Bump the incarnation on restart, so a StartView addressed to the previous
+ // incarnation and still in flight is ignored. Deterministic, so replay stays
+ // byte-identical.
+ let metadata_incarnation = self.replicas[idx].metadata_incarnation + 1;
+
+ // Recover the durable VSR state from the retained superblock before the
+ // rebuild, as production reads it in restore_metadata_consensus.
+ let recovered_state = superblock
+ .read_latest_sync()
+ .and_then(|bytes| VsrState::try_from(bytes.as_slice()).ok());
+
+ let consensus_clock = ConsensusClock::new(Rc::new(SimClock::new(self.executor.timer())));
+ let outbox = Rc::clone(&self.outboxes[idx]);
+ let (senders, mut inboxes) =
+ shard::shard_mesh_channels(shards_per_replica, SIM_INBOX_CAPACITY);
+
+ let mut shards = Vec::with_capacity(usize::from(shards_per_replica));
+ let mut stop_txs = Vec::with_capacity(usize::from(shards_per_replica));
+ let mut pump_tasks = Vec::with_capacity(usize::from(shards_per_replica));
+ let mut metadata_bundle: Option<replica::SimMetadataBundle> = None;
+ for shard_idx in 0..shards_per_replica {
+ let inbox = inboxes[usize::from(shard_idx)]
+ .take()
+ .expect("mesh yields exactly one inbox per shard");
+ let shard_superblock: Option<Rc<dyn DynSuperblockStore>> = if shard_idx == 0 {
+ let sb: Rc<dyn DynSuperblockStore> = superblock.clone();
+ Some(sb)
+ } else {
+ None
+ };
+ let shard_journal = (shard_idx == 0).then(|| Rc::clone(&metadata_journal));
+ let (shard, writer_bundle) = new_shard(
+ replica_index,
+ shard_idx,
+ format!("replica-{replica_index}-shard-{shard_idx}"),
+ &outbox,
+ self.replica_count,
+ senders.clone(),
+ inbox,
+ consensus_clock.clone(),
+ self.shell,
+ metadata_bundle.clone(),
+ shard_superblock,
+ shard_journal,
+ recovered_state,
+ metadata_incarnation,
+ );
+ if shard_idx == 0 {
+ metadata_bundle =
+ Some(writer_bundle.expect("shard 0 returns the metadata factory bundle"));
+ }
+ let (stop_tx, stop_rx) = shard::channel::<()>(1);
+ let pump_shard = Rc::clone(&shard);
+ pump_tasks.push(self.executor.spawn(async move {
+ pump_shard.run_message_pump(stop_rx).await;
+ }));
+ stop_txs.push(stop_tx);
+ shards.push(shard);
+ }
+
+ // Replacing the replica drops the old shards, losing all volatile consensus
+ // state as a real restart does. The harness-owned superblock carries forward.
+ self.replicas[idx] = SimReplica {
+ shards,
+ superblock,
+ metadata_journal,
+ metadata_incarnation,
+ _stop_txs: stop_txs,
+ pump_tasks,
+ };
+
+ // Reconnect to the network and mark the replica live again.
+ self.network
+ .process_enable(ProcessId::Replica(replica_index));
+ self.crashed.remove(&replica_index);
+ }
+
/// Advance consensus timeouts on every live replica without a full
/// step cycle: fires the pumps' virtual tick timers and runs the
/// executor to quiescence.
@@ -891,6 +1032,370 @@ mod tests {
);
}
+ /// A metadata replica that advanced its view, persisted it through the superblock
+ /// gate, then crashed recovers that same view from its own disk on restart, not a
+ /// fresh 0. The split-brain guarantee: a replica never forgets a view it acted in.
+ /// Impossible before the superblock, since a rebuilt consensus starts at view 0.
+ #[test]
+ fn given_advanced_view_when_metadata_replica_restarts_should_recover_view_from_superblock() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 5;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(client_id),
+ network_opts,
+ );
+
+ // Metadata consensus view of a replica's shard 0, if it owns one.
+ let metadata_view = |sim: &Simulator, replica: u8| -> Option<u32> {
+ let consensus = sim.replicas[replica as usize].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()?;
+ Some(consensus.view())
+ };
+
+ // Crash the metadata primary (replica 0) to force a view change on the
+ // survivors; each persists the new view through the gate.
+ sim.replica_crash(0);
+ for _ in 0..800 {
+ sim.step();
+ }
+
+ // Find the new metadata primary, elected at view >= 1.
+ let primary = (1..replica_count)
+ .find(|&replica| {
+ sim.replicas[replica as usize].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .is_some_and(|consensus| {
+ consensus.view() > 0
+ && consensus.status() == Status::Normal
+ && consensus.is_primary()
+ })
+ })
+ .expect("a metadata primary elected at view >= 1 after crashing replica 0");
+ let view_before = metadata_view(&sim, primary).expect("primary owns metadata consensus");
+ assert!(
+ view_before >= 1,
+ "expected an advanced view, got {view_before}"
+ );
+
+ // Crash and restart the new primary. Restart drops its shards, losing the
+ // in-memory view, and rebuilds from the retained superblock.
+ sim.replica_crash(primary);
+ sim.replica_restart(primary);
+
+ // It recovered its persisted view from the superblock, not a fresh 0.
+ let view_after = metadata_view(&sim, primary).expect("restarted primary owns consensus");
+ assert_eq!(
+ view_after, view_before,
+ "restarted replica must recover its persisted view from the superblock, not reset to 0"
+ );
+
+ // This run takes zero traffic, so every WAL is empty: the recovered view came
+ // from the superblock alone. Pin that, since it is what makes the restart gate
+ // below load-bearing.
+ assert!(
+ sim.replicas[primary as usize]
+ .metadata_journal
+ .last_op()
+ .is_none(),
+ "no metadata traffic in this test, so the restarted replica's WAL must be empty"
+ );
+
+ // A recovered view is a prior life even with an empty WAL, so the replica must
+ // rejoin as a probing backup. It is still primary-by-index for the recovered
+ // view, so resuming primaryship here would have it act as primary in a view the
+ // survivors may already have left, with no probe to correct it.
+ let restarted = sim.replicas[primary as usize].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .expect("restarted primary owns metadata consensus");
+ assert!(
+ restarted.is_primary(),
+ "the restarted replica is still primary-by-index for the recovered view, \
+ which is what makes ceding it necessary"
+ );
+ assert_eq!(
+ restarted.status(),
+ Status::Recovering,
+ "a replica with a recovered view must probe for the current view, not \
+ resume primaryship"
+ );
+ assert!(
+ restarted.has_ceded_primaryship(),
+ "a probing replica must be quorum-invisible until a StartView brings it \
+ forward"
+ );
+ }
+
+ #[test]
+ fn given_committed_metadata_when_solo_replica_restarts_should_recover_from_own_wal() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: 1,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ // Solo cluster: 1-of-1 quorum commits every metadata op the instant it is
+ // journaled, giving a fully-committed WAL with no uncommitted suffix to
+ // reconcile on restart.
+ let mut sim = Simulator::new(1, std::iter::once(client_id), network_opts);
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+
+ // Resolves the namespace of the stream and topic created below; `Some` exactly
+ // when the Streams STM holds them.
+ let resolve = |sim: &Simulator| {
+ sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .namespace_from_partition(
+ &iggy_binary_protocol::WireIdentifier::named("events").unwrap(),
+ &iggy_binary_protocol::WireIdentifier::named("logs").unwrap(),
+ 0,
+ )
+ };
+
+ // Drive committed metadata through consensus: a stream, then a topic with one
+ // partition under it. Each appends a prepare to shard 0's WAL and mutates the
+ // Streams STM. The topic references the stream, so the stream commits first.
+ for msg in [
+ client.create_stream("events"),
+ client.create_topic("events", "logs", 1),
+ ] {
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..50 {
+ sim.step();
+ }
+ }
+
+ let namespace_before = resolve(&sim).expect("stream + topic must resolve after creation");
+ let head_before = sim.replicas[0]
+ .metadata_journal
+ .last_op()
+ .expect("metadata ops must have been appended to the WAL");
+ let commit_before = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .expect("solo shard 0 owns metadata consensus")
+ .commit_min();
+ assert_eq!(
+ commit_before, head_before,
+ "a solo replica commits every durable op, so commit tracks the WAL head"
+ );
+
+ // Crash and restart: the shards are dropped, losing all volatile consensus and
+ // state-machine state, and rebuilt against the RETAINED WAL and superblock, so
+ // recovery is from this replica's own disk with no peer.
+ sim.replica_crash(0);
+ sim.replica_restart(0);
+
+ // The WAL bytes and index survived the restart.
+ assert_eq!(
+ sim.replicas[0].metadata_journal.last_op(),
+ Some(head_before),
+ "the metadata WAL head must survive a restart, bytes and index retained"
+ );
+ // Consensus recovered its op/commit from its own disk, not a fresh 0.
+ let consensus_ref = sim.replicas[0].shards[0].plane.metadata();
+ let consensus = consensus_ref
+ .consensus
+ .as_ref()
+ .expect("restarted solo shard 0 owns metadata consensus");
+ assert_eq!(
+ consensus.commit_min(),
+ head_before,
+ "commit must be recovered from the retained WAL, not reset to 0"
+ );
+ // Replaying the retained WAL reconstructed the committed Streams STM, so the
+ // stream and topic survive the restart from this replica's own disk.
+ assert_eq!(
+ resolve(&sim),
+ Some(namespace_before),
+ "the created stream/topic must survive the restart via WAL replay"
+ );
+ }
+
+ #[test]
+ fn given_registered_client_when_solo_replica_restarts_should_recover_session_from_own_wal() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: 1,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(1, std::iter::once(client_id), network_opts);
+ let client = SimClient::new(client_id);
+
+ // Register creates the client-table session; one committed metadata op caches
+ // a reply for at-most-once dedup. Both live in the client table, which is not
+ // part of the state machine and would otherwise reset to empty on restart.
+ sim.register_client_with_primary(&client);
+ sim.submit_request(client_id, 0, client.create_stream("events").into_generic());
+ for _ in 0..50 {
+ sim.step();
+ }
+
+ let epoch_before = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow()
+ .get_epoch(client_id);
+ assert!(
+ epoch_before.is_some(),
+ "client must hold a session before the crash"
+ );
+
+ // Crash and restart: the client table drops with the shard and is rebuilt by
+ // replaying the retained WAL through the same commit apply path the live
+ // cluster uses.
+ sim.replica_crash(0);
+ sim.replica_restart(0);
+
+ let epoch_after = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow()
+ .get_epoch(client_id);
+ assert_eq!(
+ epoch_after, epoch_before,
+ "the client session must survive a restart, reconstructed from the retained WAL, \
+ so a returning client is recognized instead of hitting NoSession"
+ );
+ }
+
+ #[test]
+ fn given_superblock_write_fails_when_primary_crashes_should_withhold_votes_and_not_elect() {
+ // The split-brain gate under test: a view-scoped send (SVC, DVC, StartView)
+ // must not go out until the new view is durable. Fail one survivor's
+ // superblock writes so its persist gate returns false, advancing its view
+ // in-memory while withholding every view-scoped send. In a 3-replica cluster,
+ // quorum 2, crashing the primary leaves one working survivor whose lone vote
+ // cannot reach quorum, so NO new primary is elected. Were the gate to send
+ // before persisting, the withheld votes would reach the peer and elect a
+ // primary that has no durable record of the new view, reintroducing
+ // split-brain on its restart. The positive control is
+ // `given_advanced_view_when_metadata_replica_restarts_...`, which elects a new
+ // primary from the same crash with healthy superblocks.
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(client_id),
+ network_opts,
+ );
+
+ // Replica 1 survives the crash and is the primary-by-index for view 1.
+ // Failing its superblock keeps it from durably taking any new view, so it
+ // never emits a view-scoped vote.
+ sim.replicas[1].superblock.set_fail_writes();
+
+ sim.replica_crash(0);
+ for _ in 0..800 {
+ sim.step();
+ }
+
+ for replica in 1..replica_count {
+ let elected = sim.replicas[replica as usize].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .is_some_and(|consensus| {
+ consensus.view() > 0
+ && consensus.status() == Status::Normal
+ && consensus.is_primary()
+ });
+ assert!(
+ !elected,
+ "replica {replica} must not be elected primary while a survivor's \
+ superblock write fails: the persist gate withholds its view-scoped \
+ votes, so quorum cannot form"
+ );
+ }
+
+ // The faulted replica never durably recorded a new view, so a restart recovers
+ // its old view, never one it might have voted in.
+ let faulted = sim.replicas[1].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .expect("replica 1 owns metadata consensus");
+ assert!(
+ faulted.view() > 0,
+ "the gate must withhold the SEND, not the view advance: with the view still \
+ at 0 there was nothing to persist and this test would pass vacuously"
+ );
+ assert_eq!(
+ sim.replicas[1].superblock.read_latest_sync(),
+ None,
+ "a failing superblock persists nothing, so no new view is durable"
+ );
+
+ // The retry is bounded. Without a backoff the 10 ms consensus tick would run a
+ // full `atomic_replace` (create, write, fsync, rename, dir fsync) on every tick
+ // for as long as the disk stays broken, on the executor that also serves
+ // partition traffic.
+ let attempts = sim.replicas[1].shards[0]
+ .plane
+ .metadata()
+ .superblock_write_failures();
+ assert!(attempts > 0, "the gate must have attempted a persist");
+ assert!(
+ attempts < 40,
+ "superblock writes must back off, not retry every tick (attempted \
+ {attempts} times)"
+ );
+ }
+
/// At-least-once failover: `SendMessages` retry on a new primary
/// re-executes. Retry reply carries a HIGHER `commit` op (re-execution
/// proof, not dedup). Duplicate payload lives at two offsets; consumers
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 050a8fa71..640ee6c30 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -18,13 +18,14 @@
use crate::bus::{SharedSimOutbox, SimOutbox};
use crate::deps::{MemStorage, SimJournal, SimMuxStateMachine, SimSnapshot};
use configs::server_ng::NgSystemConfig;
-use consensus::{ConsensusClock, LocalPipeline, VsrConsensus};
+use consensus::{ConsensusClock, LocalPipeline, Sequencer, VsrConsensus, VsrState};
use iggy_common::IggyByteSize;
use iggy_common::variadic;
-use metadata::IggyMetadata;
+use journal::superblock::DynSuperblockStore;
use metadata::stm::mux::WithFactory;
use metadata::stm::stream::{Streams, StreamsInner};
use metadata::stm::user::{Users, UsersInner};
+use metadata::{IggyMetadata, apply_committed_prepare};
use partitions::{IggyPartitions, PartitionsConfig};
use server_common::crypto;
use server_common::sharding::{METADATA_CONSENSUS_NAMESPACE, ShardId};
@@ -55,7 +56,9 @@ pub const SHELL_ROOT_PASSWORD: &str = "iggy";
// shard per replica always resolves to shard 0.
pub type Replica = shard::IggyShard<
SharedSimOutbox,
- SimJournal<MemStorage>, // MJ: metadata journal
+ // MJ: metadata journal, behind an `Rc` so the harness retains it across a
+ // replica restart; the WAL bytes and index survive the drop.
+ Rc<SimJournal<MemStorage>>,
SimSnapshot,
SimMuxStateMachine,
PapayaShardsTable,
@@ -104,7 +107,7 @@ pub const SIM_INBOX_CAPACITY: usize = 8192;
/// Panics if `senders` is not in canonical order (a bug in the caller's
/// mesh construction), or if `reader_bundle` disagrees with `shard_idx`
/// (peer without a bundle, or shard 0 with one).
-#[allow(clippy::too_many_arguments)]
+#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub fn new_shard(
replica_id: u8,
shard_idx: u16,
@@ -116,6 +119,10 @@ pub fn new_shard(
clock: ConsensusClock,
shell: bool,
reader_bundle: Option<SimMetadataBundle>,
+ superblock: Option<Rc<dyn DynSuperblockStore>>,
+ metadata_journal: Option<Rc<SimJournal<MemStorage>>>,
+ recovered_state: Option<VsrState>,
+ incarnation: u128,
) -> (Rc<Replica>, Option<SimMetadataBundle>) {
// Metadata is single-writer, mirroring server-ng bootstrap. Shard 0 owns
// the only writable STM; every peer shard rebuilds a reader-mode mirror from
@@ -128,7 +135,16 @@ pub fn new_shard(
reader_bundle.is_none(),
"shard 0 is the metadata writer (no bundle); peers are readers (bundle)"
);
- let (mux, metadata_bundle) = reader_bundle.map_or_else(
+ // Head of the retained metadata WAL, shard 0 only and `None` on a peer shard or a
+ // fresh boot. Its committed prefix drives both the STM replay below and the
+ // consensus op/commit restore, so a restart recovers committed metadata from its
+ // own disk instead of an empty state.
+ let solo = replica_count == 1;
+ let restored_op = metadata_journal
+ .as_ref()
+ .and_then(|journal| journal.last_op());
+
+ let mux = reader_bundle.map_or_else(
// Writer shard (shard 0). Seed the root user at slab id 0, matching
// production bootstrap (`ensure_root_user`); it is undeletable in-apply,
// so it stays in slot 0 and the workload's users land at slab id 1+, out
@@ -137,7 +153,9 @@ pub fn new_shard(
// user (the login gate Argon2-checks the stored hash), so seed a real
// hash of `SHELL_ROOT_PASSWORD`; otherwise a deterministic placeholder,
// since login never runs and a random per-run argon2 salt would break
- // state equality.
+ // state equality. Committed metadata replays into it from the retained WAL
+ // after `IggyMetadata` is built (see below), so the factory bundle is minted
+ // only then.
|| {
let users: Users = UsersInner::new().into();
let root_password_hash = if shell {
@@ -147,12 +165,10 @@ pub fn new_shard(
};
users.ensure_root_user(SHELL_ROOT_USERNAME, &root_password_hash);
let streams: Streams = StreamsInner::new().into();
- let mux = SimMuxStateMachine::new(variadic!(users, streams));
- let bundle = mux.factory_bundle();
- (mux, Some(bundle))
+ SimMuxStateMachine::new(variadic!(users, streams))
},
// Peer shard: reader-mode mirror over shard 0's shared read handle.
- |bundle| (SimMuxStateMachine::from_factory_bundle(bundle), None),
+ SimMuxStateMachine::from_factory_bundle,
);
// The production metadata consensus namespace, not 0: the router
@@ -160,7 +176,7 @@ pub fn new_shard(
// by comparing against this exact value; anything else hashes to an
// arbitrary shard with no metadata consensus.
let metadata_consensus = (shard_idx == 0).then(|| {
- let consensus = VsrConsensus::with_clock(
+ let mut consensus = VsrConsensus::with_clock(
CLUSTER_ID,
replica_id,
replica_count,
@@ -169,20 +185,90 @@ pub fn new_shard(
LocalPipeline::new(),
clock.clone(),
);
- consensus.init();
+ // Seed-derived incarnation, bumped per restart by the harness, so a StartView
+ // from a previous incarnation is ignored while replay stays deterministic; an
+ // OS-random nonce would break byte-identical replay.
+ consensus.set_incarnation(incarnation);
+ // View/log_view come from the durable superblock; op, commit, and the
+ // last-prepare markers come from the retained WAL. Independent inputs,
+ // mirroring server-ng's restore_metadata_consensus.
+ let last_header = metadata_journal
+ .as_ref()
+ .and_then(|journal| journal.last_header());
+ if let Some(state) = recovered_state {
+ consensus.set_view(state.view);
+ consensus.set_log_view(state.log_view);
+ consensus.mark_superblock_durable(state.view, state.log_view);
+ } else if let Some(header) = last_header {
+ // No superblock: a fresh node, one that never changed view, or the
+ // first boot after an upgrade. Production infers the view from the
+ // last WAL prepare here, so mirror it -- otherwise the branch every
+ // existing deployment takes has no sim coverage.
+ consensus.set_view(header.view);
+ }
+ // Prior life is EITHER a non-empty WAL or a recovered superblock: a view
+ // change persists without touching the WAL, so an empty journal no longer
+ // proves a fresh boot. A clustered replica with a prior life rejoins as a
+ // quorum-invisible probing backup, since the live primary re-forms its view
+ // and re-commits any uncommitted suffix; a solo replica has no peer to ask,
+ // so it inits and resumes as its own primary. Matches
+ // restore_metadata_consensus exactly. Production's primary-only re-pipeline
+ // of an uncommitted suffix is skipped, since a rejoining replica is always a
+ // backup.
+ let restored_head = restored_op.unwrap_or(0);
+ if !solo && (restored_head > 0 || recovered_state.is_some()) {
+ consensus.init_as_backup();
+ consensus.begin_view_probe();
+ } else {
+ consensus.init();
+ }
+ if let (Some(journal), Some(head)) = (metadata_journal.as_ref(), restored_op) {
+ let commit_watermark = journal.recovery_commit_watermark(solo);
+ consensus.sequencer().set_sequence(head);
+ consensus.restore_commit_state(commit_watermark, commit_watermark);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ consensus.observe_prepare_timestamp(header.timestamp);
+ }
+ // The suffix past the committed watermark is prepared but not proven
+ // committed, so gate reads until the cluster re-commits it, as production
+ // does. Solo has no such suffix: the head IS the commit point.
+ if commit_watermark < head {
+ consensus.set_recovery_barrier(head);
+ }
+ }
consensus
});
- let metadata_journal = (shard_idx == 0).then(SimJournal::<MemStorage>::default);
let metadata_snapshot = (shard_idx == 0).then(SimSnapshot::default);
let metadata = IggyMetadata::new(
metadata_consensus,
metadata_journal,
metadata_snapshot,
+ superblock,
mux,
None,
);
+ // Reconstruct shard 0's committed metadata from the retained WAL, the sim analog
+ // of production's snapshot + WAL replay. `apply_committed_prepare` is the SAME
+ // apply path the commit walk uses, so it rebuilds BOTH the state machine and the
+ // client table (sessions + at-most-once dedup), which a restart would otherwise
+ // reset. The consensus setup above already restored `commit_min` to the committed
+ // watermark; this only rebuilds derived state, and does nothing on a fresh boot.
+ // The commit notifier is unwired during recovery, hence the no-op hook.
+ if let Some(journal) = metadata.journal.as_ref() {
+ let commit_watermark = journal.recovery_commit_watermark(solo);
+ for op in 1..=commit_watermark {
+ if let Some(entry) = journal.entry_sync(op) {
+ apply_committed_prepare(&metadata.mux_stm, &metadata.client_table, |_| {}, entry);
+ }
+ }
+ }
+ // Mint the peers' read-side bundle AFTER reconstruction so it reflects the
+ // recovered state. Shard 0 only; peers pass it back in as `reader_bundle`.
+ let metadata_bundle = (shard_idx == 0).then(|| metadata.mux_stm.factory_bundle());
+
let partitions_config = PartitionsConfig {
messages_required_to_save: 1000,
size_of_messages_required_to_save: IggyByteSize::from(4 * 1024 * 1024),
@@ -200,7 +286,7 @@ pub fn new_shard(
// The deferred handlers upgrade this weak self-reference per frame; it
// stays `None` until the shard is built and downgraded into it below.
- let shard_handle: ShellShardHandle<SharedSimOutbox, SimJournal<MemStorage>, SimSnapshot> =
+ let shard_handle: ShellShardHandle<SharedSimOutbox, Rc<SimJournal<MemStorage>>, SimSnapshot> =
Rc::new(RefCell::new(None));
let ShellHandlers {
on_replica_message,