You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/07/28 11:48:24 UTC
(iggy) 01/01: feat(consensus): persist metadata VSR state in a durable superblock
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch vsrconsensus-persist
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ec7aa1841ed53eeb5f0fda9d90b5d7bee085e852
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Jul 28 17:16:17 2026 +0530
feat(consensus): persist metadata VSR state in a durable superblock
---
Cargo.lock | 4 +
core/binary_protocol/src/consensus/header.rs | 40 +-
core/consensus/Cargo.toml | 1 +
core/consensus/src/client_table.rs | 208 +++++++-
core/consensus/src/impls.rs | 266 +++++++++-
core/consensus/src/lib.rs | 6 +-
core/consensus/src/vsr_state.rs | 165 ++++++
core/integration/Cargo.toml | 8 +-
.../tests/server/cluster_view_durability_vsr.rs | 255 +++++++++
.../server/metadata_checkpoint_recovery_vsr.rs | 138 +++++
core/integration/tests/server/mod.rs | 8 +
core/journal/Cargo.toml | 1 +
core/journal/src/lib.rs | 15 +
core/journal/src/prepare_journal.rs | 158 +++++-
core/journal/src/superblock.rs | 579 +++++++++++++++++++++
core/metadata/src/impls/metadata.rs | 562 +++++++++++++++-----
core/metadata/src/impls/recovery.rs | 466 ++++++++++++++++-
core/metadata/src/lib.rs | 4 +-
core/metadata/src/stm/snapshot.rs | 18 +
core/metadata/src/stm/stream.rs | 86 +++
core/metadata/src/stm/user.rs | 5 +
core/server-ng/src/bootstrap.rs | 107 +++-
core/server-ng/src/dispatch.rs | 1 +
core/server-ng/src/partition_reconciler.rs | 2 +-
core/shard/src/lib.rs | 72 ++-
core/simulator/src/deps.rs | 121 +++++
core/simulator/src/lib.rs | 453 +++++++++++++++-
core/simulator/src/replica.rs | 111 +++-
28 files changed, 3651 insertions(+), 209 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 7591cdbf3..ce0fa5932 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,7 @@ dependencies = [
"iggy_binary_protocol",
"server_common",
"tempfile",
+ "twox-hash",
]
[[package]]
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index 7c852e150..d82edfa53 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1018,15 +1018,29 @@ 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`, inert
+ /// per the `handle_start_view` guard, so a mixed-version rolling upgrade is
+ /// wire-compatible.
+ 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 {
@@ -1067,8 +1081,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 {
@@ -1083,15 +1100,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 6b0a8526f..d020d4013 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -16,10 +16,13 @@
// 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;
+use std::fmt;
use std::mem::size_of;
-use tracing::trace;
+use tracing::{error, trace};
/// Refcounted wrapper around a committed reply.
///
@@ -65,6 +68,12 @@ 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).
@@ -86,6 +95,85 @@ pub struct ClientEntry {
pub reply: CachedReply,
}
+/// Serializable form of one occupied slot.
+///
+/// Folded into the metadata checkpoint (`MetadataSnapshot`) so sessions and dedup
+/// replies survive a restart that drained the WAL prefix they committed in.
+/// Carries `client_id`, the live index key that [`ClientEntry`] does not store,
+/// so the index rebuilds on decode. The reply is raw bytes, round-tripped
+/// through [`CachedReply::from_message`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ClientEntrySnapshot {
+ pub client_id: u128,
+ pub session: u64,
+ pub user_id: u32,
+ pub reply: Vec<u8>,
+}
+
+/// Serializable [`ClientTable`].
+///
+/// Slots keep their position, preserving deterministic eviction order across a
+/// checkpoint; [`ClientTable::from_snapshot`] rebuilds the index from the
+/// occupied slots' `client_id`s.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ClientTableSnapshot {
+ pub slots: Vec<Option<ClientEntrySnapshot>>,
+}
+
+/// 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,
+ },
+}
+
+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}"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for ClientTableDecodeError {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ Self::InvalidReply { source, .. } => Some(source),
+ Self::DuplicateClientId { .. } => None,
+ }
+ }
+}
+
/// Result of checking a request against the client table.
///
/// In-progress dedup is the caller's job, preflights consult
@@ -132,10 +220,6 @@ pub enum RequestStatus {
/// Committed state only, latest reply per client. In-flight state
/// (acks, subscribers, in-progress dedup) lives on [`crate::PipelineEntry`].
/// Updated by `commit_reply` / `commit_register`.
-///
-/// ## Known gaps
-///
-/// - **Checkpoint serialization**: slot layout deterministic, encode/decode TODO.
#[derive(Debug)]
pub struct ClientTable {
/// `None` = free slot. Deterministic iteration for eviction + serialization.
@@ -156,6 +240,92 @@ impl ClientTable {
}
}
+ /// Snapshot the table for the metadata checkpoint, preserving slot positions
+ /// (deterministic eviction order) and carrying each occupied slot's
+ /// `client_id` so the index rebuilds on decode.
+ ///
+ /// An occupied slot with no index entry is an index/slots desync, already
+ /// unreachable through the live table since every lookup goes via the index.
+ /// It is logged and dropped rather than panicking: this runs on shard 0's
+ /// checkpoint task under the durability lock, where an unwind would take down
+ /// the owner shard mid-checkpoint.
+ #[must_use]
+ pub fn to_snapshot(&self) -> ClientTableSnapshot {
+ // Invert the index once so each occupied slot can carry its client_id.
+ let mut client_of_slot: Vec<Option<u128>> = vec![None; self.slots.len()];
+ for (&client_id, &slot_idx) in &self.index {
+ client_of_slot[slot_idx] = Some(client_id);
+ }
+ let slots = self
+ .slots
+ .iter()
+ .zip(client_of_slot)
+ .enumerate()
+ .map(|(slot_idx, (slot, client_id))| {
+ let entry = slot.as_ref()?;
+ let Some(client_id) = client_id else {
+ error!(
+ slot = slot_idx,
+ "client-table checkpoint: occupied slot has no index entry; dropping it \
+ from the snapshot (index/slots desync)"
+ );
+ return None;
+ };
+ Some(ClientEntrySnapshot {
+ client_id,
+ session: entry.session,
+ user_id: entry.user_id,
+ reply: entry.reply.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.
+ ///
+ /// # Errors
+ /// [`ClientTableDecodeError`] if a slot's reply bytes are not a valid reply
+ /// message, or if two slots share a `client_id` (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) -> Result<Self, ClientTableDecodeError> {
+ let mut index = HashMap::with_capacity(snapshot.slots.len());
+ let mut slots = Vec::with_capacity(snapshot.slots.len());
+ for (slot_idx, slot) in snapshot.slots.into_iter().enumerate() {
+ let Some(entry) = slot else {
+ slots.push(None);
+ continue;
+ };
+ 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. `to_snapshot` iterates the index,
+ // whose keys are unique, 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,
+ });
+ }
+ slots.push(Some(ClientEntry {
+ session: entry.session,
+ user_id: entry.user_id,
+ reply: CachedReply::from_message(reply),
+ }));
+ }
+ Ok(Self { slots, index })
+ }
+
/// Check request against table. Session first, then request progression.
/// For Register, use [`check_register`].
///
@@ -485,6 +655,7 @@ impl ClientTable {
}
#[cfg(test)]
+#[allow(clippy::cast_possible_truncation)]
mod tests {
use super::*;
use iggy_binary_protocol::{Command2, Operation};
@@ -506,6 +677,7 @@ mod tests {
commit,
command: Command2::Reply,
operation: Operation::Register,
+ size: header_size as u32,
..ReplyHeader::default()
};
msg
@@ -524,6 +696,7 @@ mod tests {
commit,
command: Command2::Reply,
operation: Operation::SendMessages,
+ size: header_size as u32,
..ReplyHeader::default()
};
msg
@@ -534,6 +707,29 @@ mod tests {
|_| false
}
+ #[test]
+ fn to_from_snapshot_round_trips_sessions_and_dedup() {
+ let mut table = ClientTable::new(8);
+ table.commit_register(1, 11, make_register_reply(1, 10), no_in_flight());
+ table.commit_register(2, 22, make_register_reply(2, 20), no_in_flight());
+ // Client 1 committed request 5; its reply is cached for dedup.
+ table.commit_reply(1, 10, make_reply_for(1, 5, 30));
+
+ let restored = ClientTable::from_snapshot(table.to_snapshot()).unwrap();
+
+ // Sessions survive and the index is rebuilt (get_session reads it).
+ assert_eq!(restored.get_session(1), Some(10));
+ assert_eq!(restored.get_session(2), Some(20));
+ // Replaying request 5 is a dedup hit, not a re-execution: at-most-once
+ // holds across a restart.
+ assert!(matches!(
+ restored.check_request(1, 10, 5),
+ RequestStatus::Duplicate(_)
+ ));
+ // A client that never registered is still unknown.
+ assert_eq!(restored.get_session(3), None);
+ }
+
/// Register client 1 at commit 10. Returns (table, session=10).
fn table_with_client() -> (ClientTable, u64) {
let mut table = ClientTable::new(10);
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 8869f870f..d1b0e8f7e 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;
@@ -689,10 +691,15 @@ pub enum VsrAction {
/// out of the probed-primary election path.
SendRequestStartView { view: u32, namespace: u64 },
/// Send `StartView` to all backups (as new 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).
SendStartView {
view: u32,
op: u64,
commit: u64,
+ incarnation: u128,
namespace: u64,
},
/// Send `PrepareOK` for each op in `[from_op, to_op]` that is present in the WAL.
@@ -781,6 +788,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>,
@@ -903,6 +925,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),
ceded_primaryship: Cell::new(false),
status: Cell::new(Status::Recovering),
@@ -968,6 +993,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.
@@ -1316,6 +1355,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
+ /// durability 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
@@ -2125,10 +2216,13 @@ 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).
vec![VsrAction::SendStartView {
view: self.view.get(),
op: self.sequencer.current_sequence(),
commit: self.commit_max.get(),
+ incarnation: header.incarnation,
namespace: self.namespace,
}]
}
@@ -2193,6 +2287,31 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
return Vec::new();
}
+ // 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).
+ let self_incarnation = self.incarnation.get();
+ if self.status.get() == Status::Recovering
+ && self_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 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
@@ -2413,10 +2532,13 @@ 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).
let action = VsrAction::SendStartView {
view: self.view.get(),
op: new_op,
commit: max_commit,
+ incarnation: 0,
namespace: self.namespace,
};
emit_sim_event(
@@ -2613,6 +2735,18 @@ 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.
+ let checksum_body = u128::from(calculate_checksum(
+ &self.as_slice()[size_of::<PrepareHeader>()..],
+ ));
+
self.transmute_header(|old, new| {
*new = PrepareHeader {
cluster: consensus.cluster,
@@ -2630,6 +2764,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.
@@ -3050,4 +3185,131 @@ mod timestamp_clamp_tests {
"a clock ahead of the log must stamp real time, not floor + 1"
);
}
+
+ #[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.
+ #[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
+ }
+
+ 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"
+ );
+ }
+
+ /// 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 9c68f17ac..e67df2959 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -143,7 +143,9 @@ where
}
pub mod client_table;
-pub use client_table::{CachedReply, ClientTable};
+pub use client_table::{
+ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableSnapshot,
+};
// One-shot per `PipelineEntry` for in-process commit awaiters.
pub(crate) mod oneshot;
pub use oneshot::{Canceled, Receiver};
@@ -161,5 +163,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..f101be2ef
--- /dev/null
+++ b/core/consensus/src/vsr_state.rs
@@ -0,0 +1,165 @@
+// 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).
+ pub cluster: u128,
+ /// Replica index this superblock belongs to. Recovery rejects a mismatch.
+ pub replica_id: u8,
+ /// Cluster size at write time.
+ 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.
+ 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(),
+ })?;
+ Ok(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)),
+ })
+ }
+}
+
+/// 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 },
+}
+
+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}")
+ }
+ }
+ }
+}
+
+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,
+ view: 3,
+ log_view: 4,
+ 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], 3, "view low byte");
+ assert_eq!(bytes[22], 4, "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());
+ }
+}
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index d0dc17dba..5daf81c0f 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_latest_payload`).
+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..d4cf0d4de
--- /dev/null
+++ b/core/integration/tests/server/cluster_view_durability_vsr.rs
@@ -0,0 +1,255 @@
+// 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, decode_latest_payload};
+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 valid 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.
+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();
+ let payload = decode_latest_payload(slot_a.as_deref(), slot_b.as_deref())?;
+ VsrState::try_from(payload.as_slice()).ok()
+}
+
+/// 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..edb08a930 100644
--- a/core/journal/Cargo.toml
+++ b/core/journal/Cargo.toml
@@ -33,6 +33,7 @@ bytemuck = { workspace = true }
compio = { workspace = true }
iggy_binary_protocol = { workspace = true }
server_common = { 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..1f35487d5 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>();
@@ -315,6 +316,11 @@ impl PrepareJournal {
// 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 +350,67 @@ 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.
+ //
+ // TODO(wal-integrity): two gaps remain in this scan's coverage.
+ // (a) No format/version gate on `checksum_body`. A WAL written before
+ // this field was sealed carries `checksum_body == 0`, so every entry
+ // fails and boot is refused. Fail-safe, no silent loss, but a hard
+ // upgrade break with no migration path. Gate on a WAL/entry format
+ // version before verifying, or treat a zero checksum as unsealed.
+ // (b) 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.
+ 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 {}): prepare body \
+ checksum mismatch with {} bytes of entries following; refusing \
+ to truncate and discard the committed suffix",
+ header.op,
+ 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
@@ -810,6 +857,13 @@ mod tests {
let total_size = HEADER_SIZE + body_size;
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total_size);
+ // Recognizable pattern, then seal its checksum so the scan accepts it (the
+ // primary seals this in `project`).
+ 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 = u128::from(XxHash3_64::oneshot(&buffer.as_slice()[HEADER_SIZE..]));
+
let header = bytemuck::checked::from_bytes_mut::<PrepareHeader>(
&mut buffer.as_mut_slice()[..HEADER_SIZE],
);
@@ -817,13 +871,81 @@ 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]
diff --git a/core/journal/src/superblock.rs b/core/journal/src/superblock.rs
new file mode 100644
index 000000000..411f0d60b
--- /dev/null
+++ b/core/journal/src/superblock.rs
@@ -0,0 +1,579 @@
+// 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.
+//!
+//! 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 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;
+
+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_latest_payload`].
+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>),
+ /// Slots held bytes but none yielded a usable record: torn write on every
+ /// copy, checksum failure, foreign file, or an unrecognized format version
+ /// (carried in `version` when that was the cause). 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>,
+ /// 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. Reads both
+ /// slots to resume the sequence counter and aim the next write at the staler
+ /// slot. Missing files mean a fresh store.
+ ///
+ /// # 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 seq_a = read_sequence(&dir.join(FILE_A)).await?;
+ let seq_b = read_sequence(&dir.join(FILE_B)).await?;
+
+ // Latest sequence across both slots; a missing slot counts as 0.
+ 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),
+ #[cfg(debug_assertions)]
+ writing: Cell::new(false),
+ })
+ }
+}
+
+impl SuperblockStore for PingPongSuperblock {
+ async fn write(&self, payload: &[u8]) -> io::Result<()> {
+ // Single-writer contract: callers MUST serialize `write` (the metadata
+ // durability 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)?;
+ #[cfg(debug_assertions)]
+ assert!(
+ !self.writing.replace(true),
+ "PingPongSuperblock::write called concurrently; writes must be externally serialized"
+ );
+ let result = atomic_replace(&self.dir, slot.file_name(), record).await;
+ #[cfg(debug_assertions)]
+ self.writing.set(false);
+ result?;
+ 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))
+ }
+}
+
+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>> {
+ 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)
+}
+
+/// Parse and verify a record. Returns `(sequence, payload)` or `None` if the
+/// bytes are not a valid, checksum-clean record of a known version.
+fn parse_record(bytes: &[u8]) -> Option<(u64, Vec<u8>)> {
+ if bytes.len() < MIN_RECORD_LEN {
+ return None;
+ }
+ if u32::from_le_bytes(bytes[0..4].try_into().ok()?) != SUPERBLOCK_MAGIC {
+ return None;
+ }
+ if u16::from_le_bytes(bytes[4..6].try_into().ok()?) != SUPERBLOCK_VERSION {
+ return None;
+ }
+ let sequence = u64::from_le_bytes(bytes[8..16].try_into().ok()?);
+ let payload_len = u32::from_le_bytes(bytes[24..28].try_into().ok()?) as usize;
+
+ let checksum_start = HEADER_LEN.checked_add(payload_len)?;
+ if bytes.len() != checksum_start.checked_add(CHECKSUM_LEN)? {
+ return None;
+ }
+ let stored = u64::from_le_bytes(
+ bytes[checksum_start..checksum_start + CHECKSUM_LEN]
+ .try_into()
+ .ok()?,
+ );
+ if checksum(&bytes[..checksum_start]) != stored {
+ return None;
+ }
+ Some((sequence, bytes[HEADER_LEN..checksum_start].to_vec()))
+}
+
+/// Decode the newer valid record's payload from the two slots' raw bytes.
+///
+/// Mirrors [`PingPongSuperblock::read_latest`]'s newest-verifying-wins selection
+/// without a runtime. `slot_a` / `slot_b` are the file contents read from
+/// [`SLOT_FILE_NAMES`], or `None` for a missing slot. `None` when neither holds a
+/// checksum-clean record of a known version.
+#[must_use]
+pub fn decode_latest_payload(slot_a: Option<&[u8]>, slot_b: Option<&[u8]>) -> Option<Vec<u8>> {
+ match (slot_a.and_then(parse_record), slot_b.and_then(parse_record)) {
+ (None, None) => None,
+ (Some((_, payload)), None) | (None, Some((_, payload))) => Some(payload),
+ (Some((seq_a, payload_a)), Some((seq_b, payload_b))) => {
+ Some(if seq_a >= seq_b { payload_a } else { payload_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: bad magic, too short, or a length/checksum
+ /// failure on a record of the current version.
+ Corrupt,
+ /// Magic matched but the format `version` is unknown to this build, so
+ /// nothing past it can be trusted or checksum-verified.
+ UnsupportedVersion(u16),
+ /// A checksum-clean record of the current version.
+ Valid { sequence: u64, payload: Vec<u8> },
+}
+
+/// Classify one slot's raw, non-empty bytes.
+fn classify(bytes: &[u8]) -> SlotClass {
+ // Too short to hold even the framing, or a foreign file: unusable bytes.
+ if bytes.len() < MIN_RECORD_LEN {
+ return SlotClass::Corrupt;
+ }
+ if u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) != SUPERBLOCK_MAGIC {
+ return SlotClass::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. Surface the version so recovery can name
+ // it (a downgrade).
+ return SlotClass::UnsupportedVersion(version);
+ }
+ // `parse_record` re-checks magic/version (cheap) and applies this build's
+ // length + checksum validation.
+ match parse_record(bytes) {
+ Some((sequence, payload)) => SlotClass::Valid { sequence, payload },
+ None => SlotClass::Corrupt,
+ }
+}
+
+/// Combine the two slots into one outcome. Newest valid record wins; a torn or
+/// unrecognized newer slot falls back to a valid older one. With no valid record
+/// anywhere: `Empty` iff both slots were absent, else `Unreadable`, reporting an
+/// unrecognized version over generic corruption since it points at a downgrade.
+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::Valid { payload, .. }) => {
+ SuperblockContents::Present(payload)
+ }
+ (SlotClass::Absent, SlotClass::Absent) => SuperblockContents::Empty,
+ (SlotClass::UnsupportedVersion(version), _)
+ | (_, SlotClass::UnsupportedVersion(version)) => SuperblockContents::Unreadable {
+ version: Some(version),
+ },
+ _ => SuperblockContents::Unreadable { version: None },
+ }
+}
+
+async fn read_sequence(path: &Path) -> io::Result<Option<u64>> {
+ // Only a valid record contributes a sequence; an absent, corrupt, or
+ // unrecognized-version slot counts as 0 when aiming the next write.
+ Ok(match read_slot(path).await? {
+ SlotClass::Valid { sequence, .. } => Some(sequence),
+ SlotClass::Absent | SlotClass::Corrupt | SlotClass::UnsupportedVersion(_) => None,
+ })
+}
+
+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);
+ }
+ let len = usize::try_from(len)
+ .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "superblock file too large"))?;
+ 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);
+
+ 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?;
+
+ 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_newer_slot_corrupt_when_read_latest_should_fall_back_to_older() {
+ 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();
+
+ assert_eq!(
+ present(sb.read_latest().await.unwrap()).as_deref(),
+ Some(&b"first"[..])
+ );
+ }
+
+ #[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)
+ }
+ );
+ }
+
+ #[compio::test]
+ async fn given_two_writes_when_decode_latest_payload_from_disk_should_match_read_latest() {
+ // The off-runtime decoder must select the same record 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 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();
+ assert_eq!(
+ decode_latest_payload(slot_a.as_deref(), slot_b.as_deref()).as_deref(),
+ present(sb.read_latest().await.unwrap()).as_deref(),
+ );
+ assert_eq!(
+ decode_latest_payload(slot_a.as_deref(), slot_b.as_deref()).as_deref(),
+ Some(&b"second"[..]),
+ );
+ }
+
+ // 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 durability 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);
+ }
+}
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 7ba69922b..39a9bd336 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -23,11 +23,11 @@ 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, 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,
+ CLIENTS_TABLE_MAX, Canceled, ClientTable, ClientTableSnapshot, CommitLogEvent, 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,
@@ -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;
@@ -104,19 +106,31 @@ impl IggySnapshot {
/// # 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.
+ ///
+ /// 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,
@@ -159,6 +173,19 @@ impl IggySnapshot {
Self::decode(data.as_slice())
}
+
+ /// Deterministic content checksum, paired into the superblock's
+ /// `checkpoint_checksum` so a torn or mismatched snapshot/superblock pairing is
+ /// detectable on boot. Computed over the encoded msgpack bytes with the same
+ /// `XxHash3_64` the WAL and superblock use, widened to the `u128` the durable
+ /// record reserves. `encode` is deterministic for a fixed schema, so
+ /// persist-time and boot-time (encode-of-decoded) checksums agree in a build.
+ ///
+ /// # Errors
+ /// `SnapshotError::Serialize` if msgpack encoding fails.
+ pub fn checksum(&self) -> Result<u128, SnapshotError> {
+ Ok(u128::from(calculate_checksum(&self.encode()?)))
+ }
}
impl Snapshot for IggySnapshot {
@@ -199,7 +226,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>,
@@ -208,6 +239,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> {
@@ -226,6 +261,7 @@ impl<M> SnapshotCoordinator<M> {
data_dir,
create_snapshot,
checkpoint_margin: Cell::new(Self::CHECKPOINT_MARGIN),
+ last_checkpoint: Cell::new((0, 0)),
}
}
@@ -238,63 +274,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 durability lock freezes this
+ // core.
+ let encoded = snapshot.encode()?;
+ let checksum = u128::from(calculate_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(())
}
}
@@ -445,6 +489,99 @@ 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 recovery reconstruction reaches
+/// identical state from the same log through one apply path. Register creates 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. `in_flight` reports an in-progress request for a client (only
+/// the live path has a pipeline, recovery passes a constant `false`), and
+/// `fire_notifier` runs the post-commit hook (a no-op during recovery, 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>,
+ in_flight: impl Fn(u128) -> bool,
+ 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();
+ let reply = if header.operation == Operation::Register {
+ // Register: no state-machine op; the session is created below.
+ build_reply_message(&header, &bytes::Bytes::new())
+ } else if header.operation == Operation::Logout {
+ // Rebalance the consumer groups the client left, Logout's only
+ // state-machine effect. The recovery path skips it (a pre-existing
+ // `recover` gap), so it lives here, not in the shared client-table
+ // selection.
+ mux_stm.streams().remove_consumer_group_member(
+ header.client,
+ iggy_common::IggyTimestamp::from(header.timestamp),
+ );
+ build_reply_message(&header, &bytes::Bytes::new())
+ } else {
+ // 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);
+ build_reply_message_with(&header, apply.reply_body_len(), |dst| {
+ apply.write_reply_body(dst);
+ })
+ };
+ commit_to_client_table(client_table, &header, reply, in_flight);
+}
+
+/// The client-table half of a committed op: create or drop a session, or cache the
+/// reply for at-most-once dedup.
+///
+/// Split out of [`apply_committed_prepare`] so recovery reconstructs the client
+/// table through the same selection. Recovery drives the state machine through the
+/// object-safe `GatedApply` trait and cannot name `StreamsFrontend`, so this half is
+/// deliberately state-machine-free: the caller has already applied the
+/// state-machine effect and built `reply`.
+pub(crate) fn commit_to_client_table(
+ client_table: &RefCell<ClientTable>,
+ header: &PrepareHeader,
+ reply: Message<ReplyHeader>,
+ in_flight: impl Fn(u128) -> bool,
+) {
+ if header.operation == Operation::Register {
+ client_table
+ .borrow_mut()
+ .commit_register(header.client, header.user_id, reply, in_flight);
+ } else if header.operation == Operation::Logout {
+ client_table.borrow_mut().remove_client(header.client);
+ } else {
+ // Cache only if the session still exists, since a later Logout may have
+ // evicted the client; `commit_reply` no-ops otherwise.
+ let session = client_table.borrow().get_session(header.client);
+ if let Some(session) = session {
+ client_table
+ .borrow_mut()
+ .commit_reply(header.client, session, reply);
+ }
+ }
+}
+
/// Late-bound callback invoked after every committed op on shard 0's metadata
/// commit path (via `gated_apply`, including a gated no-op).
///
@@ -472,6 +609,23 @@ 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 durability 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, and the checkpoint's WAL drain likewise must not run twice at once.
+ /// 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.
+ durability_lock: LocalGate,
/// State machine - lives on all shards
pub mux_stm: M,
pub allocator: ConsensusGroupAllocator,
@@ -516,6 +670,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 {
@@ -526,6 +681,8 @@ where
consensus,
journal,
snapshot,
+ superblock,
+ durability_lock: LocalGate::new(),
mux_stm,
allocator,
coordinator,
@@ -546,6 +703,26 @@ 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 reconstructed from the WAL at boot, replacing the
+ /// empty one `new` created, so a returning client keeps its session and
+ /// at-most-once dedup cache across a restart. Server-ng bootstrap calls this on
+ /// shard 0. The client table is metadata-only, absent from the peer factory
+ /// bundle, so installing it after construction is safe.
+ pub fn set_client_table(&self, client_table: ClientTable) {
+ *self.client_table.borrow_mut() = client_table;
+ }
+
/// Install the resolved byte value used for `MaxTopicSize::ServerDefault`.
/// Server-ng bootstrap calls this with `system.topic.max_size` on every
/// shard (responses read it too); only shard 0's copy feeds admission.
@@ -2006,6 +2183,75 @@ 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.
+ ///
+ /// `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;
+ }
+ // Serialize every durability write 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. See `mark_superblock_durable` for why the state is captured before
+ // the await rather than re-read after it.
+ let _durable = self.durability_lock.acquire().await;
+ if !consensus.needs_superblock_persist() {
+ return true;
+ }
+ // 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);
+ true
+ }
+ Err(error) => {
+ tracing::error!(
+ replica_id = consensus.replica(),
+ %error,
+ "superblock persist failed; withholding view-scoped send"
+ );
+ false
+ }
+ }
+ }
+}
+
impl<B, P, J, S, M> IggyMetadata<VsrConsensus<B, P>, J, S, M>
where
B: MessageBus,
@@ -2031,6 +2277,20 @@ where
let Some(coordinator) = &self.coordinator else {
return;
};
+ // Serialize the whole checkpoint (snapshot persist, superblock write, WAL
+ // drain) against sibling durability ops on this shard. 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, race the superblock write, 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.
+ let _durable = self.durability_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
@@ -2040,31 +2300,73 @@ 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() {
+ let state = consensus.vsr_state(snap_op, checksum);
+ if let Err(error) = superblock.dyn_write(&state.to_bytes()).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"
+ %error,
+ "checkpoint superblock write failed; withholding WAL drain"
);
+ return;
}
+ consensus.mark_superblock_durable(state.view, state.log_view);
}
+
+ 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)]
@@ -2295,55 +2597,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());
- let in_flight = |c: u128| consensus.pipeline().borrow().has_message_from_client(c);
- self.client_table.borrow_mut().commit_register(
- header.client,
- header.user_id,
- reply,
- in_flight,
- );
- } 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);
- });
- // Cache only if session still exists. WAL replay may carry a
- // reply for a later-evicted client; `commit_reply` no-ops.
- let session = self.client_table.borrow().get_session(header.client);
- if let Some(session) = session {
- self.client_table
- .borrow_mut()
- .commit_reply(header.client, session, reply);
- } else {
- tracing::trace!(
- client = header.client,
- op = op,
- "commit_journal: client evicted while being prepared; skipping cache"
- );
- }
- }
+ // mutation and counter bump. The post-commit notifier 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,
+ |c| consensus.pipeline().borrow().has_message_from_client(c),
+ |operation| self.fire_commit_notifier(operation),
+ prepare,
+ );
consensus.advance_commit_min(op);
debug!("commit_journal: committed op={op}");
}
@@ -2387,6 +2650,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;
@@ -2733,6 +3004,56 @@ mod tests {
);
}
+ #[test]
+ fn populated_snapshot_reencode_and_checksum_are_stable() {
+ // Recovery recomputes the checkpoint checksum as hash(encode(decode(disk)))
+ // and compares it to the stored hash(encode(state)). They agree only if a
+ // decode then re-encode is byte-identical, which holds only while every
+ // serialized collection keeps a deterministic order. Fold a multi-slot client
+ // table (the envelope's own Vec) and assert both the raw re-encode and the
+ // checksum survive 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![
+ Some(consensus::ClientEntrySnapshot {
+ client_id: 1,
+ session: 10,
+ user_id: 1,
+ reply: vec![1, 2, 3],
+ }),
+ None,
+ Some(consensus::ClientEntrySnapshot {
+ client_id: 2,
+ session: 20,
+ user_id: 2,
+ 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 break the checkpoint \
+ checksum cross-check and refuse boot on a healthy node"
+ );
+ assert_eq!(
+ snapshot.checksum().unwrap(),
+ decoded.checksum().unwrap(),
+ "persist-time and boot-time checkpoint checksums must agree"
+ );
+ assert_ne!(
+ snapshot.checksum().unwrap(),
+ IggySnapshot::new(7).checksum().unwrap(),
+ "the checksum must track content, else it could not detect a torn snapshot"
+ );
+ }
+
type TestMux = MuxStateMachine<variadic!(Users, Streams)>;
/// Build a peer-shard-style `IggyMetadata` with `consensus`,
@@ -2741,7 +3062,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]
@@ -2917,7 +3238,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> {
@@ -3130,6 +3451,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3281,6 +3603,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
Some(dir.path().to_path_buf()),
);
@@ -3427,6 +3750,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3552,6 +3876,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
@@ -3685,6 +4010,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 88413b14d..3b7494f1e 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -15,14 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-use crate::impls::metadata::IggySnapshot;
+use crate::impls::metadata::{IggySnapshot, commit_to_client_table};
use crate::stm::StateMachine;
use crate::stm::authz::GatedApply;
use crate::stm::snapshot::{MetadataSnapshot, RestoreSnapshot, Snapshot, SnapshotError};
+use consensus::{
+ CLIENTS_TABLE_MAX, ClientTable, ClientTableDecodeError, VsrState, VsrStateError,
+ build_reply_message, build_reply_message_with,
+};
+use iggy_binary_protocol::Operation;
use iggy_binary_protocol::consensus::PrepareHeader;
use iggy_common::IggyError;
use journal::prepare_journal::{JournalError, PrepareJournal};
+use journal::superblock::{PingPongSuperblock, SuperblockContents, SuperblockStore};
use server_common::Message;
+use std::cell::RefCell;
use std::fmt;
use std::path::Path;
@@ -33,6 +40,40 @@ pub enum RecoveryError {
Journal(JournalError),
StateMachine(IggyError),
Io(std::io::Error),
+ /// Opening or reading the durable superblock failed.
+ Superblock(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 {
+ checkpoint_op: u64,
+ snapshot_op: u64,
+ },
+ /// The paired snapshot's checksum does not match the superblock's record (a
+ /// corrupt or torn snapshot).
+ CheckpointChecksumMismatch {
+ 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(VsrStateError),
+ /// Slots held bytes but none yielded a usable record: a torn write on every
+ /// copy, a checksum failure, or an unrecognized format version (carried in
+ /// `version` when that was the cause). A superblock was written and is now
+ /// unreadable, so refuse boot rather than treat the node as a fresh deployment.
+ /// An empty superblock is a different case, a genuinely fresh or
+ /// not-yet-checkpointed node, and is NOT an error.
+ SuperblockUnreadable {
+ version: Option<u16>,
+ },
}
impl fmt::Display for RecoveryError {
@@ -42,6 +83,43 @@ 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(e) => write!(f, "recovery superblock error: {e}"),
+ Self::ClientTableDecode(e) => write!(f, "recovery client-table decode error: {e}"),
+ Self::CheckpointAheadOfSnapshot {
+ checkpoint_op,
+ snapshot_op,
+ } => write!(
+ f,
+ "superblock references checkpoint op {checkpoint_op} but the on-disk snapshot is \
+ only at op {snapshot_op}: a lost or reverted snapshot write, refusing boot"
+ ),
+ Self::CheckpointChecksumMismatch {
+ op,
+ expected,
+ actual,
+ } => write!(
+ f,
+ "snapshot at op {op} does not match the superblock's checkpoint checksum \
+ (superblock {expected:#034x}, snapshot {actual:#034x}): corrupt or torn snapshot, \
+ refusing boot"
+ ),
+ Self::SuperblockUndecodable(e) => write!(
+ f,
+ "superblock record is present and checksum-clean but its payload did not decode \
+ into VSR state ({e}): unrecoverable durable consensus state, refusing boot"
+ ),
+ Self::SuperblockUnreadable {
+ version: Some(version),
+ } => write!(
+ f,
+ "superblock present but its format version {version} is unrecognized by this build \
+ (a downgrade, or a corrupt version field): refusing boot"
+ ),
+ Self::SuperblockUnreadable { version: None } => write!(
+ f,
+ "superblock present but unreadable on every copy (torn write or checksum failure): \
+ a written superblock is now corrupt, refusing boot"
+ ),
}
}
}
@@ -52,11 +130,22 @@ 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(e) => Some(e),
+ Self::ClientTableDecode(e) => Some(e),
+ Self::SuperblockUndecodable(e) => Some(e),
+ Self::CheckpointAheadOfSnapshot { .. }
+ | Self::CheckpointChecksumMismatch { .. }
+ | Self::SuperblockUnreadable { .. } => 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)
@@ -85,7 +174,29 @@ 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 verify the snapshot's
+ /// integrity against the checkpoint pairing before decoding 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 reconstructed from the replayed committed ops: per-client
+ /// sessions and the at-most-once dedup cache, empty when the WAL held no
+ /// `Register`. Installed on the metadata so a returning client is recognized
+ /// after a restart instead of hitting `NoSession`.
+ 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.
///
@@ -125,7 +236,7 @@ pub struct RecoveredMetadata<M> {
/// 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,
@@ -147,6 +258,58 @@ where
} else {
None
};
+
+ // 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(RecoveryError::Superblock)?;
+ // 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(RecoveryError::Superblock)?
+ {
+ 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(RecoveryError::SuperblockUndecodable)?,
+ )
+ }
+ SuperblockContents::Unreadable { version } => {
+ return Err(RecoveryError::SuperblockUnreadable { version });
+ }
+ SuperblockContents::Empty => None,
+ };
+
+ // Verify before decode: 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);
+ let snapshot_checksum = match snapshot.as_ref() {
+ Some(snapshot) => snapshot.checksum()?,
+ None => 0,
+ };
+ verify_checkpoint_pairing(recovered_state.as_ref(), snapshot_op, snapshot_checksum)?;
+
let replay_from = snapshot
.as_ref()
.map_or(0, |snapshot| snapshot.sequence_number() + 1);
@@ -188,6 +351,27 @@ where
.fold(snapshot_floor, u64::max)
};
+ // Reconstruct the client table alongside the state machine through the same
+ // selection the live commit walk uses (`commit_to_client_table`), so a returning
+ // client keeps its session and at-most-once dedup cache instead of hitting
+ // `NoSession`. The pipeline holds no in-flight requests during recovery, hence
+ // the constant `false`.
+ //
+ // Base it on the checkpoint's folded table (the drained WAL prefix), then replay
+ // the committed suffix on top, mirroring how the state machine is restored from
+ // the snapshot then advanced by the WAL.
+ let client_table = RefCell::new(
+ 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)?,
+ 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 {
@@ -206,15 +390,25 @@ where
format!("failed to read journal entry for op={}", header.op),
))
})?;
- // WAL replay must recompute authorization denials identically to the
- // primary/backup commit paths, so it goes through the same gate.
- let reply = mux_stm.gated_update(entry)?;
+ // Register and Logout are session ops with no state-machine effect here
+ // (Logout's consumer-group rebalance is not reconstructed, a pre-existing
+ // `recover` gap). Every other op applies through the same authorization gate
+ // as the primary/backup commit paths.
+ let reply =
+ if header.operation == Operation::Register || header.operation == Operation::Logout {
+ build_reply_message(header, &bytes::Bytes::new())
+ } else {
+ let apply = mux_stm.gated_update(entry)?;
+ build_reply_message_with(header, apply.reply_body_len(), |dst| {
+ apply.write_reply_body(dst);
+ })
+ };
+ commit_to_client_table(&client_table, header, reply, |_| false);
tracing::debug!(
target: "iggy.metadata.diag",
op = header.op,
operation = ?header.operation,
user_id = header.user_id,
- reply = ?reply,
"recovery replayed op"
);
last_applied_op = Some(header.op);
@@ -223,12 +417,61 @@ where
Ok(RecoveredMetadata {
journal,
snapshot,
+ superblock,
+ recovered_state,
+ snapshot_checkpoint: (snapshot_op, snapshot_checksum),
mux_stm,
+ client_table: client_table.into_inner(),
last_applied_op,
last_journaled_op,
})
}
+/// 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.
+const fn verify_checkpoint_pairing(
+ recovered: Option<&VsrState>,
+ snapshot_op: u64,
+ snapshot_checksum: u128,
+) -> Result<(), RecoveryError> {
+ let Some(state) = recovered else {
+ return Ok(());
+ };
+ if state.checkpoint_op > snapshot_op {
+ return Err(RecoveryError::CheckpointAheadOfSnapshot {
+ checkpoint_op: state.checkpoint_op,
+ snapshot_op,
+ });
+ }
+ if state.checkpoint_op == snapshot_op && state.checkpoint_checksum != snapshot_checksum {
+ return Err(RecoveryError::CheckpointChecksumMismatch {
+ op: state.checkpoint_op,
+ expected: state.checkpoint_checksum,
+ actual: snapshot_checksum,
+ });
+ }
+ Ok(())
+}
+
#[cfg(test)]
#[allow(clippy::cast_possible_truncation)]
mod tests {
@@ -244,6 +487,69 @@ mod tests {
const HEADER_SIZE: usize = size_of::<PrepareHeader>();
+ 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() {
+ // No superblock yet (fresh or not-yet-checkpointed): nothing to cross-check.
+ assert!(verify_checkpoint_pairing(None, 42, 0xabc).is_ok());
+
+ // Exact pairing.
+ let matching = vsr_state_with_checkpoint(42, 0xabc);
+ assert!(verify_checkpoint_pairing(Some(&matching), 42, 0xabc).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).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),
+ 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),
+ 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),
+ 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)
}
@@ -253,6 +559,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],
);
@@ -261,9 +572,62 @@ mod tests {
header.op = op;
header.commit = commit;
header.operation = Operation::CreateStream;
+ header.checksum_body = checksum_body;
+ Message::try_from(buffer).unwrap()
+ }
+
+ /// A committed `Register` for `client` at `op` (the session is the op). Body
+ /// is empty, matching how the primary projects a Register.
+ fn make_register(op: u64, client: u128, user_id: u32) -> Message<PrepareHeader> {
+ let mut buffer = Owned::<4096>::zeroed(HEADER_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],
+ );
+ header.size = HEADER_SIZE as u32;
+ header.command = Command2::Prepare;
+ header.op = op;
+ header.operation = Operation::Register;
+ header.client = client;
+ header.user_id = user_id;
+ header.checksum_body = checksum_body;
Message::try_from(buffer).unwrap()
}
+ #[compio::test]
+ async fn recover_reconstructs_client_table_from_register() {
+ let dir = tempdir().unwrap();
+ let metadata_dir = dir.path().join("metadata");
+ std::fs::create_dir_all(&metadata_dir).unwrap();
+ let journal_path = metadata_dir.join("journal.wal");
+ let client_id = 42u128;
+ {
+ let journal = PrepareJournal::open(&journal_path, 0).await.unwrap();
+ journal
+ .append(make_register(1, client_id, 7).deep_copy())
+ .await
+ .unwrap();
+ }
+
+ // solo=true: the single journaled op is committed, so it is replayed.
+ let recovered = recover::<TestStm>(
+ dir.path(),
+ true,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ |_| {},
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(
+ recovered.client_table.get_session(client_id),
+ Some(1),
+ "the Register's session must be reconstructed into the client table from the WAL"
+ );
+ }
+
#[compio::test]
async fn recover_empty_state() {
let dir = tempdir().unwrap();
@@ -433,4 +797,92 @@ mod tests {
let loaded = IggySnapshot::load(&path).unwrap();
assert_eq!(loaded.sequence_number(), 99);
}
+
+ #[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(),
+ false,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ |_| {},
+ )
+ .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(),
+ false,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ |_| {},
+ )
+ .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(),
+ false,
+ journal::prepare_journal::DEFAULT_SLOT_COUNT,
+ |_| {},
+ )
+ .await
+ .unwrap();
+ assert!(recovered.recovered_state.is_none());
+ }
}
diff --git a/core/metadata/src/lib.rs b/core/metadata/src/lib.rs
index 5aee2f77a..4a23c6b3a 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::{CommitNotifier, IggyMetadata, MetadataSubmitError};
+pub use impls::metadata::{
+ 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..1ff2b7d8c 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -103,6 +103,16 @@ 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`). Recovery trusts a checkpoint by recomputing
+/// `hash(encode(decode(disk)))` and comparing it to the persist-time
+/// `hash(encode(state))` (`recovery::verify_checkpoint_pairing`). An unordered
+/// collection reorders when decoded and re-encoded, so the checksums diverge and a
+/// healthy node refuses boot after a restart. 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 +127,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 +156,7 @@ impl MetadataSnapshot {
sequence_number,
users: None,
streams: None,
+ client_table: None,
}
}
@@ -317,6 +334,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 d6a9524bd..e6b3db230 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -1719,6 +1719,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)>,
@@ -1917,6 +1923,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;
@@ -1969,6 +1976,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 e7784a81e..cb44aa977 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 1fcd78495..459dabd00 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -40,6 +40,7 @@ use consensus::{
// `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 +49,7 @@ use iggy_common::defaults::{
};
use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyByteSize, PartitionStats, variadic};
use journal::prepare_journal::PrepareJournal;
+use journal::superblock::DynSuperblockStore;
use journal::{Journal, JournalHandle};
use message_bus::client_listener::{self, RequestHandler};
use message_bus::installer;
@@ -926,6 +928,10 @@ async fn shard_main(
recovered.snapshot,
recovered.last_applied_op,
recovered.last_journaled_op,
+ recovered.client_table,
+ recovered.superblock,
+ recovered.recovered_state,
+ recovered.snapshot_checkpoint,
)),
)
}
@@ -944,33 +950,74 @@ 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) =
- if let Some((journal, snapshot, last_applied_op, last_journaled_op)) = 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),
- );
- (Some(consensus), Some(journal), snapshot)
- } else {
- (None, None, None)
- };
+ let (
+ metadata_consensus,
+ journal_for_metadata,
+ snapshot_for_metadata,
+ superblock_for_metadata,
+ checkpoint_seed,
+ recovered_client_table,
+ ) = if let Some((
+ journal,
+ snapshot,
+ last_applied_op,
+ last_journaled_op,
+ client_table,
+ superblock,
+ recovered_state,
+ snapshot_checkpoint,
+ )) = 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);
+ // `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(
+ &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),
+ recovered_state,
+ );
+ let superblock: Rc<dyn DynSuperblockStore> = Rc::new(superblock);
+ (
+ Some(consensus),
+ Some(journal),
+ snapshot,
+ Some(superblock),
+ snapshot_checkpoint,
+ Some(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)),
);
+ // Install the client table reconstructed from shard 0's WAL, so a returning
+ // client keeps its session and dedup cache across a restart. Peer shards keep
+ // the empty table they never use.
+ if let Some(client_table) = recovered_client_table {
+ metadata.set_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());
@@ -1705,6 +1752,7 @@ fn restore_metadata_consensus(
bus: Rc<IggyMessageBus>,
prepare_queue_depth: usize,
normal_heartbeat_ticks: u64,
+ recovered_state: Option<VsrState>,
) -> VsrConsensus<Rc<IggyMessageBus>> {
let mut consensus = VsrConsensus::new(
cluster_id,
@@ -1718,12 +1766,29 @@ fn restore_metadata_consensus(
LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2),
);
consensus.set_normal_heartbeat_ticks(normal_heartbeat_ticks);
+ // 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);
}
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 89160a63a..4730c1bb7 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -2949,6 +2949,7 @@ mod tests {
Some(consensus),
Some(journal),
None,
+ None,
TestMux::default(),
None,
);
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 71968aa34..3110aa442 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -1097,7 +1097,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 60803f5b5..af98bf244 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -53,7 +53,9 @@ use metadata::IggyMetadata;
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
use partitions::{IggyPartition, IggyPartitions, PollFragments, PollingArgs, PollingConsumer};
-use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use server_common::sharding::{
+ IggyNamespace, METADATA_CONSENSUS_NAMESPACE, PartitionLocation, ShardId,
+};
use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen};
use shards_table::ShardsTable;
use std::cell::RefCell;
@@ -1728,7 +1730,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;
}
@@ -1771,7 +1775,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))
@@ -1827,7 +1833,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
@@ -1956,7 +1964,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 => {}
}
@@ -1977,6 +1991,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 => {}
@@ -2004,7 +2021,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
@@ -2674,7 +2693,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
@@ -2753,10 +2774,13 @@ 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.
let action = VsrAction::SendStartView {
view: consensus.view(),
op: consensus.sequencer().current_sequence(),
commit: consensus.commit_max(),
+ incarnation: 0,
namespace: consensus.namespace(),
};
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
@@ -2823,6 +2847,34 @@ 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, so its `needs_superblock_persist` is
+ // always dirty and it is exempt. `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 { .. }
+ );
+ 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 } => {
@@ -2860,6 +2912,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| {
@@ -2867,6 +2922,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;
});
@@ -2876,6 +2932,7 @@ async fn dispatch_vsr_actions<B, P, J>(
view,
op,
commit,
+ incarnation,
namespace,
} => {
let msg = Message::<StartViewHeader>::new(size_of::<StartViewHeader>())
@@ -2886,6 +2943,7 @@ 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;
});
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 51013cd8c..265b33073 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,
});
@@ -640,10 +681,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.
@@ -681,6 +722,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.
@@ -889,6 +1030,306 @@ 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"
+ );
+ }
+
+ #[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 session_before = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow()
+ .get_session(client_id);
+ assert!(
+ session_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 session_after = sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow()
+ .get_session(client_id);
+ assert_eq!(
+ session_after, session_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.
+ assert_eq!(
+ sim.replicas[1].superblock.read_latest_sync(),
+ None,
+ "a failing superblock persists nothing, so no new view is durable"
+ );
+ }
+
/// 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..6049e4420 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,87 @@ 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.
+ 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);
+ }
+ if let (Some(journal), Some(head)) = (metadata_journal.as_ref(), restored_op) {
+ // A non-empty WAL proves a prior life. A clustered replica 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. Gated on
+ // `head > 0` to match restore_metadata_consensus exactly. Production's
+ // primary-only re-pipeline of an uncommitted suffix is skipped, since a
+ // rejoining replica is always a backup.
+ if !solo && head > 0 {
+ consensus.init_as_backup();
+ consensus.begin_view_probe();
+ } else {
+ consensus.init();
+ }
+ 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) = journal.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);
+ }
+ } else {
+ consensus.init();
+ }
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.
+ // During recovery the pipeline holds no in-flight requests and the commit
+ // notifier is unwired, hence the constant `false` and no-op hooks.
+ 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,
+ |_| false,
+ |_| {},
+ 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 +283,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,