This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch state_transfer_metadata_clients_table
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to
refs/heads/state_transfer_metadata_clients_table by this push:
new 557d40c26 address review comments
557d40c26 is described below
commit 557d40c26402d452ecaab4b928d2e8f3fe58d582
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Sun Aug 2 08:23:13 2026 +0200
address review comments
---
core/binary_protocol/src/consensus/header.rs | 27 +-
core/configs/src/server_ng_config/cluster.rs | 6 +
core/configs/src/server_ng_config/validators.rs | 13 +
core/consensus/src/client_table.rs | 249 +++++---
core/consensus/src/le_cursor.rs | 115 ++++
core/consensus/src/lib.rs | 5 +-
core/consensus/src/plane_helpers.rs | 2 +-
core/consensus/src/state_manifest.rs | 64 +--
core/integration/src/harness/handle/server.rs | 18 +-
.../tests/cluster/metadata_state_transfer.rs | 21 +-
core/journal/src/lib.rs | 21 +-
core/metadata/src/impls/metadata.rs | 328 +++++++++--
core/metadata/src/stm/mod.rs | 2 +-
core/metadata/src/stm/mux.rs | 18 +
core/metadata/src/stm/snapshot.rs | 46 +-
core/metadata/src/stm/stream.rs | 164 +++++-
core/metadata/src/stm/user.rs | 9 +
core/partitions/src/iggy_partition.rs | 2 +-
core/partitions/src/journal.rs | 10 +
core/partitions/src/log.rs | 24 +
core/sdk/src/quic/quic_client.rs | 7 -
core/server-ng/src/bootstrap.rs | 11 +-
core/server-ng/src/responses.rs | 25 +-
core/shard/src/lib.rs | 639 +++++++++++++--------
core/shard/src/router.rs | 22 +-
core/simulator/src/deps.rs | 10 +
26 files changed, 1353 insertions(+), 505 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 93dad5d52..52b9fb6ee 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1364,6 +1364,7 @@ impl ConsensusHeader for RequestStateTransferHeader {
self.size
}
+ #[allow(clippy::cast_possible_truncation)]
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command2::RequestStateTransfer {
return Err(ConsensusError::InvalidCommand {
@@ -1371,6 +1372,16 @@ impl ConsensusHeader for RequestStateTransferHeader {
found: self.command,
});
}
+ // Header-only frame, so the size is fully determined. `EvictionHeader`
+ // pins the same way; the generic `Message::try_from` bound makes this
+ // safe either way, but a validate that checks what it can keeps the
+ // surface uniform across frames.
+ if self.size as usize != HEADER_SIZE {
+ return Err(ConsensusError::InvalidSize {
+ expected: HEADER_SIZE as u32,
+ found: self.size,
+ });
+ }
Ok(())
}
}
@@ -1440,7 +1451,12 @@ impl ConsensusHeader for StateTransferTargetHeader {
));
}
// Unavailable is a bare refusal; a manifest body on it would be
- // ambiguous (which offer would the chunks belong to?).
+ // ambiguous (which offer would the chunks belong to?). An
+ // `available == 1` body is left unbounded here on purpose: it carries
+ // the state manifest, whose entry count and per-artifact/total lengths
+ // are bounded where it is decoded (`STATE_MANIFEST_ENTRIES_MAX`, plus
+ // the receiver's artifact caps), and the generic `Message::try_from`
+ // bound already keeps `size` inside the frame.
if self.available == 0 && self.size as usize != HEADER_SIZE {
return Err(ConsensusError::InvalidField(
"unavailable descriptor must be header-only".to_string(),
@@ -1499,6 +1515,7 @@ impl ConsensusHeader for RequestStateChunkHeader {
self.size
}
+ #[allow(clippy::cast_possible_truncation)]
fn validate(&self) -> Result<(), ConsensusError> {
if self.command != Command2::RequestStateChunk {
return Err(ConsensusError::InvalidCommand {
@@ -1511,6 +1528,14 @@ impl ConsensusHeader for RequestStateChunkHeader {
"chunk len must be non-zero".to_string(),
));
}
+ // Header-only frame; the requested `len` describes the REPLY, which
the
+ // serving side clamps against its own chunk size and the bus ceiling.
+ if self.size as usize != HEADER_SIZE {
+ return Err(ConsensusError::InvalidSize {
+ expected: HEADER_SIZE as u32,
+ found: self.size,
+ });
+ }
Ok(())
}
}
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
index 22f750402..a10a5384f 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -71,6 +71,12 @@ const MAX_VIEW_PROBE_ATTEMPTS: u32 = 100;
/// static-asserts it equal to `shard::REPAIR_CHUNK_MAX`.
pub const DEFAULT_REPAIR_CHUNK_MAX: usize = 128;
+/// `size_of::<StateChunkHeader>()`. Duplicated here for the same reason as
+/// [`DEFAULT_REPAIR_CHUNK_MAX`]; `core/server-ng`'s bootstrap static-asserts
it
+/// against the real header. Used to reject a `[message_bus] max_message_size`
+/// too small to carry a single state-transfer chunk.
+pub const STATE_CHUNK_HEADER_LEN: u64 = 256;
+
/// Upper bound on `repair_chunk_max`. A chunk rides the per-peer bus queue, so
/// the load-bearing rule is `repair_chunk_max <
message_bus.peer_queue_capacity`
/// (enforced at the top level); this standalone ceiling is a typo guard.
diff --git a/core/configs/src/server_ng_config/validators.rs
b/core/configs/src/server_ng_config/validators.rs
index 7b24d322c..c00d55df8 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -26,6 +26,7 @@
//! net.
use super::COMPONENT_NG;
+use super::cluster::STATE_CHUNK_HEADER_LEN;
use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig};
use crate::ConfigurationError;
use err_trail::ErrContext;
@@ -202,6 +203,18 @@ impl Validatable<ConfigurationError> for ServerNgConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // State-transfer chunks ride the same bus. A cap that cannot carry one
+ // header plus a byte of payload makes every rejoin that needs a
+ // transfer impossible, and the failure surfaces only as a replica
+ // connection tearing down when the frame is rejected on the read side.
+ let bus_cap = self.message_bus.max_message_size.as_bytes_u64();
+ if bus_cap <= STATE_CHUNK_HEADER_LEN {
+ eprintln!(
+ "{COMPONENT_NG} message_bus.max_message_size ({bus_cap}) must
exceed the {STATE_CHUNK_HEADER_LEN}-byte state-chunk header: state transfer
serves artifact chunks over this bus, and a frame above the cap is rejected by
the receiving transport, which tears down the whole replica connection"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
// WS frame chain: websocket.max_frame_size <=
websocket.max_message_size
// <= message_bus.max_message_size. The bus's WS / WSS install path
takes
// its frame tuning from [websocket], so a WS ceiling above the bus's
own
diff --git a/core/consensus/src/client_table.rs
b/core/consensus/src/client_table.rs
index 2360dc10e..20767574d 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::le_cursor::{LeCursor, Truncated, split_verified_trailer};
use iggy_binary_protocol::consensus::ConsensusError;
use iggy_binary_protocol::{GenericHeader, ReplyHeader};
use serde::{Deserialize, Serialize};
@@ -24,10 +25,8 @@ use server_common::{
};
use std::collections::{HashMap, VecDeque};
use std::fmt;
-use std::hash::Hasher;
use std::mem::size_of;
use tracing::trace;
-use twox_hash::XxHash3_64;
/// Refcounted wrapper around a committed reply.
///
@@ -952,9 +951,15 @@ impl ClientTable {
}
}
-/// Failure decoding an encoded client table (state transfer install path).
+/// Failure decoding the state-transfer WIRE encoding of a client table
+/// ([`ClientTable::encode`] / [`ClientTable::decode`]).
+///
+/// Distinct from [`ClientTableDecodeError`], which covers the msgpack
+/// CHECKPOINT encoding read off local disk. The two formats validate the same
+/// invariants against differently-trusted inputs: a checkpoint is this node's
+/// own bytes, while these arrive from a peer.
#[derive(Debug)]
-pub enum ClientTableCodecError {
+pub enum ClientTableWireError {
/// Byte stream ended mid-field.
Truncated,
/// Leading magic is not [`CLIENT_TABLE_MAGIC`].
@@ -968,9 +973,17 @@ pub enum ClientTableCodecError {
/// An entry carries an empty reply ring (violates the never-empty
/// invariant registration establishes).
EmptyRing,
+ /// Two entries claim the same `client_id`. Indexing them would leave one
+ /// slot occupied but unindexed, which desynchronizes the capacity check in
+ /// [`ClientTable::commit_register`] from the actual occupancy.
+ DuplicateClientId { slot: usize, client_id: u128 },
+ /// A reply ring longer than [`REPLY_RING_CAPACITY`]. `push_latest` only
+ /// evicts on equality, so an over-capacity ring grows without bound, and
+ /// `encode` writes its length as a `u8`.
+ RingTooLong { slot: usize, len: u8, max: usize },
}
-impl std::fmt::Display for ClientTableCodecError {
+impl std::fmt::Display for ClientTableWireError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Truncated => write!(f, "encoded client table truncated"),
@@ -984,15 +997,38 @@ impl std::fmt::Display for ClientTableCodecError {
}
Self::InvalidReply => write!(f, "encoded client table holds an
invalid cached reply"),
Self::EmptyRing => write!(f, "encoded client table entry has an
empty reply ring"),
+ Self::DuplicateClientId { slot, client_id } => write!(
+ f,
+ "encoded client table repeats client {client_id} at entry
{slot}"
+ ),
+ Self::RingTooLong { slot, len, max } => write!(
+ f,
+ "encoded client table entry {slot} has a {len}-reply ring, max
{max}"
+ ),
}
}
}
-impl std::error::Error for ClientTableCodecError {}
+impl std::error::Error for ClientTableWireError {}
+
+impl From<Truncated> for ClientTableWireError {
+ fn from(_: Truncated) -> Self {
+ Self::Truncated
+ }
+}
/// Format tag for [`ClientTable::encode`]; bump on layout change.
pub const CLIENT_TABLE_MAGIC: [u8; 4] = *b"ICT1";
+/// Per-entry fixed fields in the wire encoding: `client(u128) epoch(u64)
+/// user_id(u32) watermark(u64) watermark_checksum(u128) ring_len(u8)`.
+const ENCODED_ENTRY_FIXED_LEN: usize = size_of::<u128>()
+ + size_of::<u64>()
+ + size_of::<u32>()
+ + size_of::<u64>()
+ + size_of::<u128>()
+ + size_of::<u8>();
+
impl ClientTable {
/// Encode the table for state transfer.
///
@@ -1001,12 +1037,33 @@ impl ClientTable {
/// watermark_checksum(u128) ring_len(u8) [reply_len(u32) reply_bytes]*`,
/// terminated by an `XxHash3_64(8)` over everything before it.
///
- /// Slot order makes the bytes deterministic across caught-up replicas:
- /// slots are apply-derived state.
+ /// Slot order makes the bytes deterministic across caught-up replicas
+ /// that reached this state the same way. Not a cross-replica byte
+ /// identity: entries compact into `0..count` here, so a replica that
+ /// installed a transfer re-slots its clients and later registrations land
+ /// elsewhere than on a replica that never did.
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn encode(&self) -> Vec<u8> {
- let mut out = Vec::with_capacity(64 * self.index.len() + 16);
+ // Size exactly rather than guess: each cached reply is a full wire
+ // message, so at the default client cap a guessed reservation is off
by
+ // orders of magnitude and costs several reallocs of a multi-MB buffer
+ // on the serving primary's pump, once per offer build.
+ let entries = self.slots.iter().flatten();
+ let reserved = CLIENT_TABLE_MAGIC.len()
+ + size_of::<u32>()
+ + entries
+ .map(|entry| {
+ ENCODED_ENTRY_FIXED_LEN
+ + entry
+ .ring
+ .iter()
+ .map(|reply| size_of::<u32>() + reply.bytes.len())
+ .sum::<usize>()
+ })
+ .sum::<usize>()
+ + size_of::<u64>();
+ let mut out = Vec::with_capacity(reserved);
out.extend_from_slice(&CLIENT_TABLE_MAGIC);
out.extend_from_slice(&(self.index.len() as u32).to_le_bytes());
for (slot_idx, slot) in self.slots.iter().enumerate() {
@@ -1024,9 +1081,9 @@ impl ClientTable {
out.extend_from_slice(bytes);
}
}
- let mut hasher = XxHash3_64::new();
- hasher.write(&out);
- out.extend_from_slice(&hasher.finish().to_le_bytes());
+ debug_assert_eq!(out.len() + size_of::<u64>(), reserved, "encode
reservation");
+ let trailer = crate::state_manifest::state_artifact_checksum(&out);
+ out.extend_from_slice(&trailer.to_le_bytes());
out
}
@@ -1037,33 +1094,25 @@ impl ClientTable {
/// back, not trusted from the wire, so it cannot drift from the ring.
///
/// # Errors
- /// [`ClientTableCodecError`] on truncation, magic/checksum mismatch,
- /// capacity overflow, or an undecodable cached reply.
+ /// [`ClientTableWireError`] on truncation, magic/checksum mismatch,
+ /// capacity overflow, a duplicate `client_id`, an out-of-range ring
+ /// length, or an undecodable cached reply.
///
/// # Panics
/// Unreachable: slice-to-array conversions are length-checked first.
- pub fn decode(bytes: &[u8], max_clients: usize) -> Result<Self,
ClientTableCodecError> {
- const TRAILER: usize = size_of::<u64>();
- let content_len = bytes
- .len()
- .checked_sub(TRAILER)
- .ok_or(ClientTableCodecError::Truncated)?;
- let (content, trailer) = bytes.split_at(content_len);
- let expected = u64::from_le_bytes(trailer.try_into().expect("trailer
is 8 bytes"));
- let mut hasher = XxHash3_64::new();
- hasher.write(content);
- let actual = hasher.finish();
- if expected != actual {
- return Err(ClientTableCodecError::ChecksumMismatch { expected,
actual });
- }
+ pub fn decode(bytes: &[u8], max_clients: usize) -> Result<Self,
ClientTableWireError> {
+ let content = split_verified_trailer(bytes).map_err(|mismatch| match
mismatch {
+ Some((expected, actual)) => ClientTableWireError::ChecksumMismatch
{ expected, actual },
+ None => ClientTableWireError::Truncated,
+ })?;
- let mut reader = CodecReader { bytes: content };
+ let mut reader = LeCursor::new(content);
if reader.take(CLIENT_TABLE_MAGIC.len())? != CLIENT_TABLE_MAGIC {
- return Err(ClientTableCodecError::BadMagic);
+ return Err(ClientTableWireError::BadMagic);
}
let count = reader.u32()?;
if count as usize > max_clients {
- return Err(ClientTableCodecError::TooManyEntries {
+ return Err(ClientTableWireError::TooManyEntries {
count,
max: max_clients,
});
@@ -1078,7 +1127,19 @@ impl ClientTable {
let watermark_checksum = reader.u128()?;
let ring_len = reader.u8()?;
if ring_len == 0 {
- return Err(ClientTableCodecError::EmptyRing);
+ return Err(ClientTableWireError::EmptyRing);
+ }
+ // The artifact checksum only proves the bytes survived transit; it
+ // says nothing about the peer that computed them. An over-capacity
+ // ring is admitted forever after (`push_latest` evicts only on
+ // equality) and eventually wraps `encode`'s `u8` length, making
+ // this table permanently un-transferable onward.
+ if usize::from(ring_len) > REPLY_RING_CAPACITY {
+ return Err(ClientTableWireError::RingTooLong {
+ slot: slot_idx,
+ len: ring_len,
+ max: REPLY_RING_CAPACITY,
+ });
}
let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY);
for _ in 0..ring_len {
@@ -1086,9 +1147,9 @@ impl ClientTable {
let reply_bytes = reader.take(reply_len)?;
let owned =
Owned::<MESSAGE_ALIGN>::copy_from_slice(reply_bytes);
let message = Message::<GenericHeader>::try_from(owned)
- .map_err(|_| ClientTableCodecError::InvalidReply)?
+ .map_err(|_| ClientTableWireError::InvalidReply)?
.try_into_typed::<ReplyHeader>()
- .map_err(|_| ClientTableCodecError::InvalidReply)?;
+ .map_err(|_| ClientTableWireError::InvalidReply)?;
ring.push_back(CachedReply::from_message(message));
}
let latest_commit = ring
@@ -1105,44 +1166,32 @@ impl ClientTable {
client_id,
latest_commit,
});
- table.index.insert(client_id, slot_idx);
+ // Reject rather than overwrite, as the checkpoint decoder does. An
+ // overwrite leaves the displaced slot occupied but unindexed, and
+ // `commit_register` sizes its eviction check off `index.len()`: a
+ // full table would then skip eviction, find no free slot, and
+ // panic the shard.
+ if let Some(first_slot) = table.index.insert(client_id, slot_idx) {
+ return Err(ClientTableWireError::DuplicateClientId {
+ slot: first_slot,
+ client_id,
+ });
+ }
}
- if !reader.bytes.is_empty() {
- return Err(ClientTableCodecError::Truncated);
+ if !reader.remaining().is_empty() {
+ return Err(ClientTableWireError::Truncated);
}
Ok(table)
}
-}
-/// Little-endian cursor over the encoded table content.
-struct CodecReader<'a> {
- bytes: &'a [u8],
-}
-
-impl<'a> CodecReader<'a> {
- const fn take(&mut self, len: usize) -> Result<&'a [u8],
ClientTableCodecError> {
- if self.bytes.len() < len {
- return Err(ClientTableCodecError::Truncated);
- }
- let (head, tail) = self.bytes.split_at(len);
- self.bytes = tail;
- Ok(head)
- }
-
- fn u8(&mut self) -> Result<u8, ClientTableCodecError> {
- Ok(self.take(1)?[0])
- }
-
- fn u32(&mut self) -> Result<u32, ClientTableCodecError> {
- Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("4B")))
- }
-
- fn u64(&mut self) -> Result<u64, ClientTableCodecError> {
- Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("8B")))
- }
-
- fn u128(&mut self) -> Result<u128, ClientTableCodecError> {
- Ok(u128::from_le_bytes(self.take(16)?.try_into().expect("16B")))
+ /// Slot capacity, i.e. the largest table this one can absorb from a peer.
+ ///
+ /// Sized at construction from the configured cap, then raised by
+ /// [`Self::from_snapshot`] to cover any slot the checkpoint holds, so this
+ /// can exceed `[metadata] clients_table_max`.
+ #[must_use]
+ pub const fn capacity(&self) -> usize {
+ self.slots.len()
}
}
@@ -1984,19 +2033,19 @@ mod tests {
flipped[8] ^= 0xFF;
assert!(matches!(
ClientTable::decode(&flipped, 4),
- Err(ClientTableCodecError::ChecksumMismatch { .. })
+ Err(ClientTableWireError::ChecksumMismatch { .. })
));
// Truncation.
assert!(matches!(
ClientTable::decode(&encoded[..encoded.len() - 1], 4),
- Err(ClientTableCodecError::ChecksumMismatch { .. } |
ClientTableCodecError::Truncated)
+ Err(ClientTableWireError::ChecksumMismatch { .. } |
ClientTableWireError::Truncated)
));
// Capacity overflow.
assert!(matches!(
ClientTable::decode(&encoded, 0),
- Err(ClientTableCodecError::TooManyEntries { .. })
+ Err(ClientTableWireError::TooManyEntries { .. })
));
let empty = ClientTable::new(4).encode();
@@ -2007,4 +2056,64 @@ mod tests {
0
);
}
+
+ /// Re-stamp a hand-edited body so it passes the trailer check and the
+ /// per-field validations are what the decode actually exercises.
+ fn reseal(mut content: Vec<u8>) -> Vec<u8> {
+ let trailer = crate::state_manifest::state_artifact_checksum(&content);
+ content.extend_from_slice(&trailer.to_le_bytes());
+ content
+ }
+
+ // A duplicate client_id would collapse the index onto one slot, leaving
the
+ // other occupied but unindexed. `commit_register` sizes its eviction check
+ // off `index.len()`, so a full table would then skip eviction, find no
free
+ // slot, and panic the shard.
+ #[test]
+ fn decode_rejects_a_duplicate_client_id() {
+ let mut table = ClientTable::new(2);
+ table.commit_register(7, TEST_USER_ID, make_register_reply(7, 10));
+ table.commit_register(9, TEST_USER_ID, make_register_reply(9, 20));
+ let encoded = table.encode();
+
+ // Rewrite the second entry's client_id to match the first. Entries are
+ // fixed-width up to their ring, and both rings hold one register reply
+ // of equal length, so the second entry starts at a computable offset.
+ let content = &encoded[..encoded.len() - size_of::<u64>()];
+ let header_len = CLIENT_TABLE_MAGIC.len() + size_of::<u32>();
+ let reply_len = (content.len() - header_len - 2 *
ENCODED_ENTRY_FIXED_LEN) / 2;
+ let second = header_len + ENCODED_ENTRY_FIXED_LEN + reply_len;
+ let mut duped = content.to_vec();
+ duped[second..second +
size_of::<u128>()].copy_from_slice(&7u128.to_le_bytes());
+
+ assert!(matches!(
+ ClientTable::decode(&reseal(duped), 2),
+ Err(ClientTableWireError::DuplicateClientId { client_id: 7, .. })
+ ));
+ }
+
+ // `push_latest` evicts only on equality, so an over-capacity ring is
+ // admitted permanently and grows on every later reply until `encode`'s u8
+ // length wraps and the table stops being transferable at all.
+ #[test]
+ fn decode_rejects_a_ring_longer_than_capacity() {
+ let mut table = ClientTable::new(1);
+ table.commit_register(3, TEST_USER_ID, make_register_reply(3, 10));
+ let encoded = table.encode();
+
+ let content = &encoded[..encoded.len() - size_of::<u64>()];
+ let mut oversized = content.to_vec();
+ // ring_len is the last fixed field of the entry.
+ let ring_len_at = CLIENT_TABLE_MAGIC.len() + size_of::<u32>() +
ENCODED_ENTRY_FIXED_LEN - 1;
+ assert_eq!(oversized[ring_len_at], 1, "entry's ring holds one reply");
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ oversized[ring_len_at] = REPLY_RING_CAPACITY as u8 + 1;
+ }
+
+ assert!(matches!(
+ ClientTable::decode(&reseal(oversized), 1),
+ Err(ClientTableWireError::RingTooLong { .. })
+ ));
+ }
}
diff --git a/core/consensus/src/le_cursor.rs b/core/consensus/src/le_cursor.rs
new file mode 100644
index 000000000..5b6b36556
--- /dev/null
+++ b/core/consensus/src/le_cursor.rs
@@ -0,0 +1,115 @@
+// 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.
+
+//! Little-endian read cursor shared by the crate's hand-rolled wire codecs
+//! (the state-transfer manifest and the client-table wire encoding).
+//!
+//! Each codec has its own error enum, so the cursor reports the one failure it
+//! can produce ([`Truncated`]) and every codec converts it through `From`.
+//!
+//! [`Truncated`]: Truncated
+
+/// The cursor ran past the end of its slice.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Truncated;
+
+/// Little-endian cursor over an encoded body.
+pub struct LeCursor<'a> {
+ bytes: &'a [u8],
+}
+
+impl<'a> LeCursor<'a> {
+ #[must_use]
+ pub const fn new(bytes: &'a [u8]) -> Self {
+ Self { bytes }
+ }
+
+ /// Bytes not yet consumed. A codec that requires its input fully consumed
+ /// checks this after its last field.
+ #[must_use]
+ pub const fn remaining(&self) -> &'a [u8] {
+ self.bytes
+ }
+
+ /// # Errors
+ /// [`Truncated`] if fewer than `len` bytes remain.
+ pub const fn take(&mut self, len: usize) -> Result<&'a [u8], Truncated> {
+ if self.bytes.len() < len {
+ return Err(Truncated);
+ }
+ let (head, tail) = self.bytes.split_at(len);
+ self.bytes = tail;
+ Ok(head)
+ }
+
+ /// # Errors
+ /// [`Truncated`] if the slice is exhausted.
+ pub fn u8(&mut self) -> Result<u8, Truncated> {
+ Ok(self.take(1)?[0])
+ }
+
+ /// # Errors
+ /// [`Truncated`] if fewer than 4 bytes remain.
+ ///
+ /// # Panics
+ /// Unreachable: the slice is length-checked before the array conversion.
+ pub fn u32(&mut self) -> Result<u32, Truncated> {
+ Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("4B")))
+ }
+
+ /// # Errors
+ /// [`Truncated`] if fewer than 8 bytes remain.
+ ///
+ /// # Panics
+ /// Unreachable: the slice is length-checked before the array conversion.
+ pub fn u64(&mut self) -> Result<u64, Truncated> {
+ Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("8B")))
+ }
+
+ /// # Errors
+ /// [`Truncated`] if fewer than 16 bytes remain.
+ ///
+ /// # Panics
+ /// Unreachable: the slice is length-checked before the array conversion.
+ pub fn u128(&mut self) -> Result<u128, Truncated> {
+ Ok(u128::from_le_bytes(self.take(16)?.try_into().expect("16B")))
+ }
+}
+
+/// Split a trailing `XxHash3_64` stamp off an encoded body and verify it.
+///
+/// Both codecs terminate with the same 8-byte trailer over everything before
+/// it. Returns the content with the trailer removed.
+///
+/// # Errors
+/// `Err(None)` when the input is too short to hold a trailer,
`Err(Some((expected,
+/// actual)))` when the stamp does not match; the caller maps both into its own
+/// error enum.
+///
+/// # Panics
+/// Unreachable: the split point is derived from the trailer's own size.
+pub fn split_verified_trailer(bytes: &[u8]) -> Result<&[u8], Option<(u64,
u64)>> {
+ let content_len = bytes.len().checked_sub(size_of::<u64>()).ok_or(None)?;
+ let (content, trailer) = bytes.split_at(content_len);
+ let expected = u64::from_le_bytes(trailer.try_into().expect("trailer is 8
bytes"));
+ let actual = crate::state_manifest::state_artifact_checksum(content);
+ if expected == actual {
+ Ok(content)
+ } else {
+ Err(Some((expected, actual)))
+ }
+}
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index f4c38bce8..7b5457a04 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -148,9 +148,10 @@ where
}
pub mod client_table;
+pub mod le_cursor;
pub use client_table::{
- CachedReply, ClientEntrySnapshot, ClientTable, ClientTableCodecError,
ClientTableDecodeError,
- ClientTableSnapshot, CommitReply,
+ CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError,
ClientTableSnapshot,
+ ClientTableWireError, CommitReply,
};
pub mod state_manifest;
pub use state_manifest::{
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index e252afd46..4515f2ba7 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -42,7 +42,7 @@ pub async fn pipeline_prepare_common<C, F>(
assert!(consensus.is_normal(), "on_request: status must be normal");
assert!(
!consensus.is_transferring(),
- "on_request: must not be syncing"
+ "on_request: must not be transferring state"
);
consensus.verify_pipeline();
diff --git a/core/consensus/src/state_manifest.rs
b/core/consensus/src/state_manifest.rs
index ea226b270..2841c6c5e 100644
--- a/core/consensus/src/state_manifest.rs
+++ b/core/consensus/src/state_manifest.rs
@@ -35,6 +35,7 @@
//!
//! [`kind`]: StateArtifact::kind
+use crate::le_cursor::{LeCursor, Truncated, split_verified_trailer};
use std::hash::Hasher;
use twox_hash::XxHash3_64;
@@ -128,6 +129,12 @@ impl std::fmt::Display for StateManifestError {
impl std::error::Error for StateManifestError {}
+impl From<Truncated> for StateManifestError {
+ fn from(_: Truncated) -> Self {
+ Self::Truncated
+ }
+}
+
/// Format tag for [`encode_state_manifest`]; bump on incompatible change
/// (appending entry fields is compatible, see `entry_len`).
pub const STATE_MANIFEST_MAGIC: [u8; 4] = *b"ISM1";
@@ -167,9 +174,8 @@ pub fn encode_state_manifest(artifacts: &[StateArtifact])
-> Vec<u8> {
out.extend_from_slice(&artifact.len.to_le_bytes());
out.extend_from_slice(&artifact.checksum.to_le_bytes());
}
- let mut hasher = XxHash3_64::new();
- hasher.write(&out);
- out.extend_from_slice(&hasher.finish().to_le_bytes());
+ let trailer = state_artifact_checksum(&out);
+ out.extend_from_slice(&trailer.to_le_bytes());
out
}
@@ -186,21 +192,12 @@ pub fn encode_state_manifest(artifacts: &[StateArtifact])
-> Vec<u8> {
/// # Panics
/// Unreachable: slice-to-array conversions are length-checked first.
pub fn decode_state_manifest(bytes: &[u8]) -> Result<Vec<StateArtifact>,
StateManifestError> {
- const TRAILER: usize = size_of::<u64>();
- let content_len = bytes
- .len()
- .checked_sub(TRAILER)
- .ok_or(StateManifestError::Truncated)?;
- let (content, trailer) = bytes.split_at(content_len);
- let expected = u64::from_le_bytes(trailer.try_into().expect("trailer is 8
bytes"));
- let mut hasher = XxHash3_64::new();
- hasher.write(content);
- let actual = hasher.finish();
- if expected != actual {
- return Err(StateManifestError::ChecksumMismatch { expected, actual });
- }
+ let content = split_verified_trailer(bytes).map_err(|mismatch| match
mismatch {
+ Some((expected, actual)) => StateManifestError::ChecksumMismatch {
expected, actual },
+ None => StateManifestError::Truncated,
+ })?;
- let mut reader = ManifestReader { bytes: content };
+ let mut reader = LeCursor::new(content);
if reader.take(STATE_MANIFEST_MAGIC.len())? != STATE_MANIFEST_MAGIC {
return Err(StateManifestError::BadMagic);
}
@@ -215,8 +212,7 @@ pub fn decode_state_manifest(bytes: &[u8]) ->
Result<Vec<StateArtifact>, StateMa
let mut artifacts = Vec::with_capacity(count as usize);
for _ in 0..count {
- let entry = reader.take(entry_len as usize)?;
- let mut entry = ManifestReader { bytes: entry };
+ let mut entry = LeCursor::new(reader.take(entry_len as usize)?);
artifacts.push(StateArtifact {
kind: entry.u8()?,
frontier: entry.u64()?,
@@ -224,40 +220,12 @@ pub fn decode_state_manifest(bytes: &[u8]) ->
Result<Vec<StateArtifact>, StateMa
checksum: entry.u64()?,
});
}
- if !reader.bytes.is_empty() {
+ if !reader.remaining().is_empty() {
return Err(StateManifestError::Truncated);
}
Ok(artifacts)
}
-/// Little-endian cursor over the encoded manifest content.
-struct ManifestReader<'a> {
- bytes: &'a [u8],
-}
-
-impl<'a> ManifestReader<'a> {
- const fn take(&mut self, len: usize) -> Result<&'a [u8],
StateManifestError> {
- if self.bytes.len() < len {
- return Err(StateManifestError::Truncated);
- }
- let (head, tail) = self.bytes.split_at(len);
- self.bytes = tail;
- Ok(head)
- }
-
- fn u8(&mut self) -> Result<u8, StateManifestError> {
- Ok(self.take(1)?[0])
- }
-
- fn u32(&mut self) -> Result<u32, StateManifestError> {
- Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("4B")))
- }
-
- fn u64(&mut self) -> Result<u64, StateManifestError> {
- Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("8B")))
- }
-}
-
#[cfg(test)]
mod tests {
use super::*;
diff --git a/core/integration/src/harness/handle/server.rs
b/core/integration/src/harness/handle/server.rs
index 7d6fefbc1..34c549807 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -211,6 +211,21 @@ impl ServerHandle {
.map_or(0, |log| strip_ansi(&log).matches(marker).count())
}
+ /// This node's stdout log with ANSI escapes stripped, the same text
+ /// [`Self::stdout_occurrences`] matches against. Empty when the log is
+ /// missing or unreadable.
+ ///
+ /// For callers that need to PARSE a marker's fields (`checkpoint_op=193`
+ /// reaches the file as `checkpoint_op\x1b[0m\x1b[2m=\x1b[0m193`) rather
+ /// than just count occurrences of it.
+ #[must_use]
+ pub fn stdout_plain(&self) -> String {
+ self.stdout_path
+ .as_ref()
+ .and_then(|path| fs::read_to_string(path).ok())
+ .map_or_else(String::new, |log| strip_ansi(&log))
+ }
+
/// Returns a `ClientBuilder` using the test transport.
///
/// Returns an error if no test transport is configured.
@@ -1010,8 +1025,6 @@ impl ServerHandle {
/// the committed prefix the node missed no longer exists as WAL entries on
/// the peers that checkpointed it.
pub fn restart_from_clean_slate(&mut self) -> Result<(), TestBinaryError> {
- let cleanup = self.config.cleanup;
- self.config.cleanup = false;
self.stop_dependents()?;
self.stop()?;
@@ -1023,7 +1036,6 @@ impl ServerHandle {
})?;
}
- self.config.cleanup = cleanup;
self.start()
}
}
diff --git a/core/integration/tests/cluster/metadata_state_transfer.rs
b/core/integration/tests/cluster/metadata_state_transfer.rs
index 27eeafaec..b66bfacf7 100644
--- a/core/integration/tests/cluster/metadata_state_transfer.rs
+++ b/core/integration/tests/cluster/metadata_state_transfer.rs
@@ -193,23 +193,10 @@ async fn
given_checkpointed_cluster_when_fresh_node_joins_late_should_state_tran
async fn assert_pairing_matches_install(harness: &TestHarness, node: usize) {
const PAIRING_MARKER: &str = "state transfer recorded its checkpoint
pairing";
- // The server colors its tracing fields, so `checkpoint_op=193` reaches
the log
- // as `checkpoint_op\x1b[0m\x1b[2m=\x1b[0m193`. Strip escapes before
matching,
- // the same reason `ServerHandle::stdout_occurrences` does.
- let (stdout, _stderr) = harness.node(node).collect_logs();
- let mut plain = String::with_capacity(stdout.len());
- let mut chars = stdout.chars();
- while let Some(c) = chars.next() {
- if c == '\u{1b}' {
- for escaped in chars.by_ref() {
- if escaped.is_ascii_alphabetic() {
- break;
- }
- }
- } else {
- plain.push(c);
- }
- }
+ // Escapes stripped by the harness: the server colors its tracing fields,
so
+ // `checkpoint_op=193` reaches the log as
+ // `checkpoint_op\x1b[0m\x1b[2m=\x1b[0m193`.
+ let plain = harness.node(node).stdout_plain();
let recorded_op: u64 = plain
.lines()
.filter(|line| line.contains(PAIRING_MARKER))
diff --git a/core/journal/src/lib.rs b/core/journal/src/lib.rs
index a66a515fc..1159573c3 100644
--- a/core/journal/src/lib.rs
+++ b/core/journal/src/lib.rs
@@ -63,14 +63,19 @@ where
/// Snapshot watermark: entries at or below it are evictable. `0` for
/// journals without snapshot bookkeeping.
- fn snapshot_op(&self) -> u64 {
- 0
- }
-
- /// Advance the snapshot watermark (see [`Self::snapshot_op`]). No-op for
- /// journals without snapshot bookkeeping. State transfer uses this to
- /// mark pre-transfer residents superseded by the installed snapshot.
- fn set_snapshot_op(&self, _op: u64) {}
+ ///
+ /// Required rather than defaulted, and paired with
+ /// [`Self::set_snapshot_op`]: a wrapper that forwards one while inheriting
+ /// the other is silently broken in one direction and panics in the other
+ /// (an implementation whose setter asserts monotonicity would see a getter
+ /// stuck at `0` hand it a watermark below the real one). Journals without
+ /// snapshot bookkeeping answer `0` and no-op the setter EXPLICITLY.
+ fn snapshot_op(&self) -> u64;
+
+ /// Advance the snapshot watermark (see [`Self::snapshot_op`]). State
+ /// transfer uses this to mark pre-transfer residents superseded by the
+ /// installed snapshot.
+ fn set_snapshot_op(&self, op: u64);
}
// TODO: Move to other crate.
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 7f616fd61..fa3d791c0 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -809,6 +809,11 @@ pub struct IggyMetadata<C, J, S, M> {
/// them (re-running `commit_register` would double-bump epochs). `0`
/// outside state transfer (no op is skipped). Monotone per install.
client_table_frontier: Cell<u64>,
+ /// Last built [`StateTransferOffer`], shared by every requester of the
same
+ /// snapshot generation. Rebuilding per request re-reads and re-decodes the
+ /// whole snapshot on shard 0's pump, and hands each requester its own
+ /// multi-MB copy.
+ transfer_offer_cache: RefCell<Option<Rc<StateTransferOffer>>>,
}
impl<C, J, S, M> IggyMetadata<C, J, S, M>
@@ -849,11 +854,34 @@ where
default_max_topic_size: Cell::new(u64::MAX),
default_message_expiry: Cell::new(u64::MAX),
client_table_frontier: Cell::new(0),
+ transfer_offer_cache: RefCell::new(None),
}
}
}
impl<C, J, S, M> IggyMetadata<C, J, S, M> {
+ /// Slot capacity of the LIVE client table, i.e. the largest transferred
+ /// table this replica can absorb.
+ ///
+ /// Read at decode instead of a separately plumbed `clients_table_max`: the
+ /// live table sizes itself to `max(configured, highest recovered slot +
1)`,
+ /// so a serving primary can legitimately hold more entries than this
node's
+ /// raw config value and decoding against that value would reject every
+ /// round.
+ #[must_use]
+ pub fn client_table_capacity(&self) -> usize {
+ self.client_table.borrow().capacity()
+ }
+
+ /// Drop the cached state-transfer offer, releasing its snapshot copy.
+ ///
+ /// Called by the shard's expiry sweep once no requester holds an offer:
+ /// the cache exists to collapse repeat builds within one rejoin, not to
+ /// pin a snapshot for the life of the process.
+ pub fn clear_state_transfer_offer_cache(&self) {
+ self.transfer_offer_cache.borrow_mut().take();
+ }
+
/// Install (or replace) the post-commit notifier. Passing `None`
/// removes any previous one. Server-ng bootstrap calls this on shard 0
/// only; peer shards never commit metadata locally.
@@ -1338,20 +1366,125 @@ where
/// One state-transfer serving payload.
///
-/// The on-disk snapshot bytes plus the live client table, both
+/// The on-disk snapshot payload plus the live client table, both
/// frontier-stamped. Built by [`IggyMetadata::state_transfer_offer`] on the
-/// serving primary; the shard caches it per requester and serves chunks
-/// from it.
+/// serving primary; the shard serves chunks out of it and shares one instance
+/// across every requester of the same snapshot generation.
pub struct StateTransferOffer {
/// Serving primary's applied frontier when the offer was built; the
/// receiver's tail repair targets past this.
pub commit_op: u64,
- /// Manifest entries, index-aligned with `payloads`. Metadata plane:
- /// `[METADATA_SNAPSHOT (frontier = sequence_number),
- /// CLIENT_TABLE (frontier = commit_min at encode)]`.
- pub artifacts: Vec<consensus::StateArtifact>,
- /// Artifact bytes, chunk-served by `(manifest index, offset)`.
- pub payloads: Vec<Vec<u8>>,
+ /// The offered snapshot's `sequence_number`, i.e. the generation this
+ /// offer describes. Reused as the cache key: a later checkpoint rewrites
+ /// `snapshot.bin` and invalidates every payload below.
+ pub snapshot_seq: u64,
+ /// Manifest entries paired with their bytes. One `Vec` of pairs rather
+ /// than two index-aligned `Vec`s: the manifest is encoded in one file and
+ /// the chunks served in another, so a desync would be invisible at both
+ /// ends. Metadata plane: `[METADATA_SNAPSHOT (frontier = sequence_number),
+ /// CLIENT_TABLE (frontier = commit_min at encode)]`.
+ ///
+ /// Payloads are refcounted so n simultaneous rejoiners share one copy
+ /// rather than pinning n multi-MB snapshots on shard 0.
+ pub artifacts: Vec<(consensus::StateArtifact, Rc<Vec<u8>>)>,
+}
+
+impl StateTransferOffer {
+ /// Manifest entries for the descriptor body.
+ #[must_use]
+ pub fn manifest(&self) -> Vec<consensus::StateArtifact> {
+ self.artifacts.iter().map(|(entry, _)| *entry).collect()
+ }
+
+ /// Bytes of the artifact at `index` in manifest order.
+ #[must_use]
+ pub fn payload(&self, index: usize) -> Option<&[u8]> {
+ self.artifacts.get(index).map(|(_, bytes)| bytes.as_slice())
+ }
+
+ /// Number of artifacts on offer.
+ #[must_use]
+ pub const fn len(&self) -> usize {
+ self.artifacts.len()
+ }
+
+ /// Whether the offer carries no artifacts at all.
+ #[must_use]
+ pub const fn is_empty(&self) -> bool {
+ self.artifacts.is_empty()
+ }
+
+ /// Total advertised bytes across every artifact.
+ #[must_use]
+ pub fn total_len(&self) -> u64 {
+ self.artifacts.iter().map(|(entry, _)| entry.len).sum()
+ }
+}
+
+/// Why this replica cannot serve a state transfer right now.
+///
+/// Named rather than folded into `None` so the refusal the requester sees is
+/// logged with its actual cause: "no snapshot persisted" and "snapshot.bin is
+/// corrupt" call for opposite operator responses.
+#[derive(Debug)]
+pub enum StateTransferUnavailable {
+ /// Not a caught-up primary, so a client-table read would not be
+ /// authoritative.
+ NotCaughtUpPrimary,
+ /// This shard has no snapshot coordinator, so it never checkpoints.
+ NoCoordinator,
+ /// No snapshot has ever been persisted. The WAL still holds the full
+ /// history, so the requester's journal repair covers its whole gap.
+ NoSnapshot,
+ /// `snapshot.bin` exists but could not be read, or failed its integrity
+ /// trailer. Refusing is strictly better than shipping it: the receiver
+ /// would re-seal the corruption under a fresh valid trailer.
+ SnapshotUnreadable(SnapshotError),
+}
+
+impl std::fmt::Display for StateTransferUnavailable {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::NotCaughtUpPrimary => write!(f, "not a caught-up primary"),
+ Self::NoCoordinator => write!(f, "no snapshot coordinator on this
shard"),
+ Self::NoSnapshot => write!(f, "no snapshot has been persisted
yet"),
+ Self::SnapshotUnreadable(source) => {
+ write!(f, "persisted snapshot is unreadable: {source}")
+ }
+ }
+ }
+}
+
+impl std::error::Error for StateTransferUnavailable {
+ fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+ match self {
+ Self::SnapshotUnreadable(source) => Some(source),
+ _ => None,
+ }
+ }
+}
+
+/// What a completed [`IggyMetadata::install_state_transfer`] landed.
+///
+/// A degraded install is reported HERE rather than as an `Err`, because it is
+/// a success: the snapshot, table, frontiers and commit point are all in
+/// place by the time the pairing write is attempted. Returning it as an error
+/// invites a caller to treat a completed install as a failure and redo it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct InstallOutcome {
+ /// The receiver's new applied frontier, `max(snapshot_seq,
+ /// local_applied)`. These differ whenever a serving peer offered a
+ /// snapshot BEHIND this replica and the local state machine was kept.
+ pub applied_frontier: u64,
+ /// Whether the transferred checkpoint's `(checkpoint_op, checksum)`
+ /// pairing reached the durable superblock.
+ ///
+ /// `false` leaves the install fully usable: the coordinator already holds
+ /// the new pairing, so the next superblock write (view change or
+ /// checkpoint) records it. Until then a crash recovers the PREVIOUS
+ /// checkpoint and this replica transfers again -- correct, just wasted
+ /// work.
+ pub pairing_durable: bool,
}
impl<B, J, S, M> IggyMetadata<VsrConsensus<B>, J, S, M>
@@ -1366,45 +1499,96 @@ where
Error = iggy_common::IggyError,
>,
{
- /// Build a state-transfer offer for a restarted peer, or `None` when
- /// this replica cannot serve one right now: not a caught-up primary
- /// (table reads would not be authoritative), or no snapshot has ever
- /// been persisted (the WAL then still holds the full history and the
- /// requester's journal repair covers the whole gap).
+ /// Build a state-transfer offer for a restarted peer.
///
- /// The snapshot is served as the on-disk bytes verbatim -- possibly
- /// stale, which costs nothing: the receiver journal-repairs
- /// `(snapshot_seq, commit_max]` afterwards through the existing repair
- /// machinery. The table is encoded live at this instant; both frontier
- /// stamps read `commit_min` inside one synchronous region, so they are
- /// mutually consistent.
- #[must_use]
- pub fn state_transfer_offer(&self) -> Option<StateTransferOffer> {
- let consensus = self.consensus.as_ref()?;
+ /// The snapshot is served as the on-disk PAYLOAD, with its integrity
+ /// trailer verified and stripped. Both halves matter. Verified, because a
+ /// flipped bit inside the payload that still msgpack-decodes would
+ /// otherwise be re-sealed on the receiver under a fresh valid trailer and
+ /// a matching pairing: a fault the source node refuses to boot over would
+ /// become undetectable on the second node. Stripped, because the receiver
+ /// re-persists what it is sent through `write_durably`, which appends a
+ /// trailer of its own -- shipping the sealed file grows `snapshot.bin` by
+ /// one trailer per transfer generation and leaves it byte-shape-different
+ /// from a locally checkpointed one.
+ ///
+ /// The snapshot may be stale, which costs nothing: the receiver
+ /// journal-repairs `(snapshot_seq, commit_max]` afterwards through the
+ /// existing repair machinery. The table is encoded live at this instant;
+ /// both frontier stamps read `commit_min` inside one synchronous region,
+ /// so they are mutually consistent.
+ ///
+ /// The result is cached and shared: a repeat request for the same snapshot
+ /// generation reuses it instead of re-reading and re-decoding the file on
+ /// shard 0's pump.
+ ///
+ /// # Errors
+ /// [`StateTransferUnavailable`] naming why this replica cannot serve.
+ pub fn state_transfer_offer(&self) -> Result<Rc<StateTransferOffer>,
StateTransferUnavailable> {
+ let consensus = self
+ .consensus
+ .as_ref()
+ .ok_or(StateTransferUnavailable::NoCoordinator)?;
if !is_caught_up_primary(consensus) {
- return None;
+ return Err(StateTransferUnavailable::NotCaughtUpPrimary);
}
- let coordinator = self.coordinator.as_ref()?;
- let snapshot = std::fs::read(coordinator.snapshot_path()).ok()?;
- let snapshot_seq =
IggySnapshot::decode(&snapshot).ok()?.sequence_number();
+ let coordinator = self
+ .coordinator
+ .as_ref()
+ .ok_or(StateTransferUnavailable::NoCoordinator)?;
+ let path = coordinator.snapshot_path();
+ if !path.exists() {
+ return Err(StateTransferUnavailable::NoSnapshot);
+ }
+ let sealed = std::fs::read(&path)
+ .map_err(|source|
StateTransferUnavailable::SnapshotUnreadable(source.into()))?;
+ // Verifies the trailer and hands back the payload alone.
+ let (payload, _) =
+ split_trailer(&sealed,
&path).map_err(StateTransferUnavailable::SnapshotUnreadable)?;
+ // Still decoded rather than read off `last_checkpoint()`:
`write_durably`
+ // renames before the parent-dir fsync, so a DirSync failure leaves
the new
+ // file live with that cell stale, and the offer would then
under-advertise
+ // the frontier it is actually shipping.
+ let snapshot_seq = IggySnapshot::decode(payload)
+ .map_err(StateTransferUnavailable::SnapshotUnreadable)?
+ .sequence_number();
+
+ // Reuse the cached offer for this generation. Only the SNAPSHOT half
is
+ // expensive to rebuild, and the cached table is merely older, never
+ // incoherent: its frontier is stamped at its own encode, and the
receiver
+ // replays everything above that frontier during tail repair.
+ if let Some(cached) = self.transfer_offer_cache.borrow().as_ref()
+ && cached.snapshot_seq == snapshot_seq
+ {
+ return Ok(Rc::clone(cached));
+ }
+
let commit_op = consensus.commit_min();
let table = self.client_table.borrow().encode();
- Some(StateTransferOffer {
+ let offer = Rc::new(StateTransferOffer {
commit_op,
+ snapshot_seq,
artifacts: vec![
- consensus::StateArtifact::for_bytes(
- consensus::artifact_kind::METADATA_SNAPSHOT,
- snapshot_seq,
- &snapshot,
+ (
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::METADATA_SNAPSHOT,
+ snapshot_seq,
+ payload,
+ ),
+ Rc::new(payload.to_vec()),
),
- consensus::StateArtifact::for_bytes(
- consensus::artifact_kind::CLIENT_TABLE,
- commit_op,
- &table,
+ (
+ consensus::StateArtifact::for_bytes(
+ consensus::artifact_kind::CLIENT_TABLE,
+ commit_op,
+ &table,
+ ),
+ Rc::new(table),
),
],
- payloads: vec![snapshot, table],
- })
+ });
+ *self.transfer_offer_cache.borrow_mut() = Some(Rc::clone(&offer));
+ Ok(offer)
}
/// Install a fetched state transfer: persist + restore the snapshot,
@@ -1424,12 +1608,13 @@ where
/// reconciler's periodic full diff against the committed STM, which
/// reads the restored state on its next tick.
///
- /// Returns the installed snapshot sequence (the receiver's new applied
- /// frontier).
+ /// Returns an [`InstallOutcome`]: the new applied frontier, plus whether
+ /// the transferred checkpoint's pairing reached the durable superblock.
///
/// # Errors
/// [`SnapshotError`] when the snapshot bytes do not decode, the persist
- /// fails, or the in-place restore is rejected.
+ /// fails, or the in-place restore is rejected. Every `Err` here means
+ /// NOTHING was installed.
///
/// # Panics
/// If called on a shard without consensus (state transfer is a shard-0
@@ -1441,7 +1626,7 @@ where
client_table: ClientTable,
table_frontier: u64,
commit_op: u64,
- ) -> Result<u64, SnapshotError>
+ ) -> Result<InstallOutcome, SnapshotError>
where
M: RestoreSnapshotInPlace<MetadataSnapshot>,
{
@@ -1489,21 +1674,34 @@ where
let local_applied = consensus.commit_min();
let snapshot_ahead = snapshot_seq > local_applied;
- // Hold the superblock gate across the ENTIRE install, not just its
- // superblock write. A checkpoint does the same pair of steps --
rewrite
- // `snapshot.bin`, record `(checkpoint_op, checksum)` -- so
interleaving
- // the two could leave the file from one and the durable pairing from
the
- // other, which is precisely the torn pairing the superblock exists to
- // detect. Serializing against `checkpoint_if_needed` makes that
- // impossible by construction rather than by argument (a transferring
- // replica withholds acks, so it should not be committing, but "should"
- // is not an invariant this path can rely on).
+ // Serialize the whole install against a concurrent checkpoint, in the
+ // checkpoint's own lock order (`checkpoint_lock` then
+ // `superblock_lock`), so the two cannot deadlock against each other.
+ //
+ // Both do the same pair of steps -- rewrite `snapshot.bin`, record
+ // `(checkpoint_op, checksum)` -- and `checkpoint_if_needed` holds
+ // `checkpoint_lock` across BOTH while taking `superblock_lock` only
+ // around the pairing write. `superblock_lock` alone therefore
+ // serializes nothing against the checkpoint's file rewrite: interleave
+ // them and the file comes from one while the durable pairing describes
+ // the other, which is exactly the torn pairing the superblock exists
to
+ // detect (a crash inside that window refuses boot with
+ // `CheckpointChecksumMismatch`). Checkpoints run on spawned tasks, so
a
+ // prepare that passed preflight before the transfer armed can drive
one
+ // during this install's superblock await -- a transferring replica
+ // withholds acks, but "should not be committing" is not an invariant
+ // this path can rest on.
//
- // Deadlock-free: nothing between here and the write awaits, and
- // `write_superblock` takes no lock of its own -- the checkpoint path
- // calls it under this same gate.
- let _superblock_gate = if snapshot_ahead && self.superblock.is_some() {
- Some(self.superblock_lock.acquire().await)
+ // Deadlock-free: nothing between here and the superblock write awaits,
+ // and `write_superblock` takes no lock of its own.
+ let _install_gates = if snapshot_ahead {
+ let checkpoint = self.checkpoint_lock.acquire().await;
+ let superblock = if self.superblock.is_some() {
+ Some(self.superblock_lock.acquire().await)
+ } else {
+ None
+ };
+ Some((checkpoint, superblock))
} else {
None
};
@@ -1589,23 +1787,27 @@ where
// checkpoint with the WAL intact and the transfer simply retries; a
crash
// after it recovers the transferred state. Failing here withholds
nothing
// already written -- the snapshot on disk subsumes the recorded
pairing, which
- // `verify_checkpoint_pairing` accepts -- but it is still reported so
the
- // caller can retry rather than treat the install as fully durable.
+ // `verify_checkpoint_pairing` accepts -- so it is reported as a
DEGRADED
+ // install rather than a failed one.
+ let mut pairing_durable = true;
if snapshot_ahead && let Some(superblock) = self.superblock.as_ref() {
- // Already under `_superblock_gate`, acquired above; re-acquiring
here
+ // Already under `_install_gates`, acquired above; re-acquiring
here
// would deadlock on the same non-reentrant gate.
- if !self.write_superblock(consensus, superblock.as_ref()).await {
+ pairing_durable = self.write_superblock(consensus,
superblock.as_ref()).await;
+ if !pairing_durable {
tracing::error!(
snapshot_seq,
commit_op,
"state transfer installed but the superblock write failed;
the \
transferred checkpoint is not durable yet"
);
- return Err(SnapshotError::SuperblockNotDurable { op:
snapshot_seq });
}
}
- Ok(snapshot_seq.max(local_applied))
+ Ok(InstallOutcome {
+ applied_frontier: snapshot_seq.max(local_applied),
+ pairing_durable,
+ })
}
/// Submit `Register` from in-process, await commit. Wire reply still fires
@@ -2756,7 +2958,7 @@ where
assert!(consensus.is_normal(), "promotion: status must be normal");
assert!(
!consensus.is_transferring(),
- "promotion: must not be syncing"
+ "promotion: must not be transferring state"
);
consensus.verify_pipeline();
match reply_sender {
diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs
index 0763f4401..698dbd3c0 100644
--- a/core/metadata/src/stm/mod.rs
+++ b/core/metadata/src/stm/mod.rs
@@ -441,7 +441,7 @@ macro_rules! collect_handlers {
},
)*
[<$state Command>]::RestoreSnapshot(snapshot) => {
- *self =
Self::inner_from_snapshot(snapshot.clone());
+ self.restore_in_place(snapshot.clone());
$crate::stm::result::ApplyReply::default()
},
});
diff --git a/core/metadata/src/stm/mux.rs b/core/metadata/src/stm/mux.rs
index 027b4bcaf..f9e163fb2 100644
--- a/core/metadata/src/stm/mux.rs
+++ b/core/metadata/src/stm/mux.rs
@@ -194,9 +194,23 @@ where
Tail: RestoreSnapshotInPlace<SnapshotData>,
{
fn restore_snapshot_in_place(&self, snapshot: &SnapshotData) -> Result<(),
SnapshotError> {
+ // Two-phase, and the reason this is not a plain `?`-chain: the halves
+ // mutate independently with no rollback between them, and the caller
+ // has already persisted the transferred snapshot and seeded its
+ // pairing by the time it gets here. A half-restored mux would then be
+ // fed the local WAL's contiguous suffix by the follow-up
+ // `commit_journal`, replaying it into only the half that moved. The
+ // boot path fail-stops on the same input; this makes the running path
+ // agree.
+ self.check_restorable(snapshot)?;
self.0.restore_snapshot_in_place(snapshot)?;
self.1.restore_snapshot_in_place(snapshot)
}
+
+ fn check_restorable(&self, snapshot: &SnapshotData) -> Result<(),
SnapshotError> {
+ self.0.check_restorable(snapshot)?;
+ self.1.check_restorable(snapshot)
+ }
}
impl<T, SnapshotData> RestoreSnapshotInPlace<SnapshotData> for
MuxStateMachine<T>
@@ -206,6 +220,10 @@ where
fn restore_snapshot_in_place(&self, snapshot: &SnapshotData) -> Result<(),
SnapshotError> {
self.inner.restore_snapshot_in_place(snapshot)
}
+
+ fn check_restorable(&self, snapshot: &SnapshotData) -> Result<(),
SnapshotError> {
+ self.inner.check_restorable(snapshot)
+ }
}
impl<SnapshotData, T> FillSnapshot<SnapshotData> for MuxStateMachine<T>
diff --git a/core/metadata/src/stm/snapshot.rs
b/core/metadata/src/stm/snapshot.rs
index d80c15a33..ca28a30a4 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -53,11 +53,6 @@ pub enum SnapshotError {
commit_op: u64,
table_frontier: u64,
},
- /// A state-transfer install wrote its snapshot but could not durably
record the
- /// paired `(checkpoint_op, checksum)` in the superblock. The on-disk
snapshot
- /// subsumes the stale pairing, so recovery still accepts it; the caller
retries
- /// so the pairing (and the integrity check it powers) stops lagging.
- SuperblockNotDurable { op: u64 },
}
/// Stage at which snapshot persistence failed.
@@ -109,13 +104,6 @@ impl fmt::Display for SnapshotError {
commit_op {commit_op}, table frontier {table_frontier}"
)
}
- Self::SuperblockNotDurable { op } => {
- write!(
- f,
- "state transfer installed a snapshot at op {op} but could
not \
- durably pair it in the superblock"
- )
- }
}
}
}
@@ -128,8 +116,7 @@ impl std::error::Error for SnapshotError {
Self::Io(e) | Self::Persist { source: e, .. } => Some(e),
Self::ChecksumMismatch { .. }
| Self::Truncated { .. }
- | Self::IncoherentManifest { .. }
- | Self::SuperblockNotDurable { .. } => None,
+ | Self::IncoherentManifest { .. } => None,
}
}
}
@@ -310,6 +297,16 @@ pub trait RestoreSnapshot<S>: Sized {
pub trait RestoreSnapshotInPlace<S> {
/// Replace this state machine's contents from the snapshot.
fn restore_snapshot_in_place(&self, snapshot: &S) -> Result<(),
SnapshotError>;
+
+ /// Whether this state machine can restore from `snapshot`, WITHOUT
+ /// mutating anything.
+ ///
+ /// A mux restores its halves one at a time, so a snapshot missing the
+ /// second half would otherwise leave the first restored and the second on
+ /// pre-transfer state -- a split the caller cannot undo, because the
+ /// transferred snapshot was already persisted and its pairing seeded.
+ /// Every half agrees here before any half mutates.
+ fn check_restorable(&self, snapshot: &S) -> Result<(), SnapshotError>;
}
/// Base case for the recursive tuple pattern - unit type terminates the
recursion.
@@ -317,6 +314,10 @@ impl<S> RestoreSnapshotInPlace<S> for () {
fn restore_snapshot_in_place(&self, _snapshot: &S) -> Result<(),
SnapshotError> {
Ok(())
}
+
+ fn check_restorable(&self, _snapshot: &S) -> Result<(), SnapshotError> {
+ Ok(())
+ }
}
/// Base case for the recursive tuple pattern - unit type terminates the
recursion.
@@ -399,6 +400,23 @@ macro_rules! impl_fill_restore {
})
}
}
+
+ fn check_restorable(
+ &self,
+ snapshot: &$crate::stm::snapshot::MetadataSnapshot,
+ ) -> Result<(), $crate::stm::snapshot::SnapshotError> {
+ use serde::de::Error as _;
+ use $crate::stm::snapshot::SnapshotError;
+ if snapshot.$field.is_none() {
+ return Err(SnapshotError::Deserialize(
+ rmp_serde::decode::Error::custom(format_args!(
+ "Snapshot Restore Error: {}",
+ stringify!($field)
+ )),
+ ));
+ }
+ Ok(())
+ }
}
};
}
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 767affc57..9e94d2ae4 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -27,7 +27,7 @@ use crate::stm::result::{
};
use crate::stm::snapshot::Snapshotable;
use crate::{collect_handlers, define_state, impl_fill_restore};
-use ahash::AHashMap;
+use ahash::{AHashMap, AHashSet};
use bytes::{BufMut, Bytes, BytesMut};
use iggy_binary_protocol::codec::{WireDecode, WireEncode};
// Only `seed_namespace` (below, sim/test-gated) uses these at module scope;
@@ -461,6 +461,43 @@ impl StatsRegistry {
!(*sid == stream_id && *tid == topic_id && *pid >=
first_removed)
});
}
+
+ /// Drop every entry the snapshot does not describe, keeping the rest.
+ ///
+ /// Used by the in-place restore (state transfer), which replaces the whole
+ /// stream tree but must not replace the registry: partition counters live
+ /// only here (never snapshotted), so survivors have to keep their `Arc`s
+ /// or every already-materialized partition reads (0,0,0,0) forever. Slab
+ /// keys are recycled, so anything the snapshot dropped has to go with it.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ fn retain_from_snapshot(&self, snapshot: &StreamsSnapshot) {
+ let mut live_streams: AHashSet<usize> = AHashSet::new();
+ let mut live_topics: AHashSet<(usize, usize)> = AHashSet::new();
+ let mut live_partitions: AHashSet<(usize, usize, usize)> =
AHashSet::new();
+ for (stream_key, stream) in &snapshot.items {
+ live_streams.insert(*stream_key);
+ for (topic_key, topic) in &stream.topics {
+ live_topics.insert((*stream_key, *topic_key));
+ for partition in &topic.partitions {
+ live_partitions.insert((*stream_key, *topic_key,
partition.id));
+ }
+ }
+ }
+ self.streams
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|id, _| live_streams.contains(id));
+ self.topics
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|key, _| live_topics.contains(key));
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|key, _| live_partitions.contains(key));
+ }
}
define_state! {
@@ -1945,22 +1982,43 @@ impl Snapshotable for Streams {
fn from_snapshot(
snapshot: Self::Snapshot,
) -> Result<Self, crate::stm::snapshot::SnapshotError> {
- Ok(StreamsInner::inner_from_snapshot(snapshot).into())
+ // Boot: no live registry exists yet, so mint one. Safe because
+ // `new_from_empty` clones this single inner onto the other left-right
+ // buffer rather than building a second one.
+ Ok(StreamsInner::inner_from_snapshot(snapshot,
Arc::new(StatsRegistry::default())).into())
}
}
impl StreamsInner {
- /// Build a complete `StreamsInner` from a snapshot section. Shared by
- /// wrapper construction ([`Snapshotable::from_snapshot`]) and the
- /// in-place restore command (state transfer), which absorbs it on both
- /// left-right buffers.
- pub(crate) fn inner_from_snapshot(snapshot: StreamsSnapshot) -> Self {
+ /// Rebuild from a snapshot section IN PLACE, keeping the live stats
+ /// registry.
+ ///
+ /// The restore command is absorbed on BOTH left-right buffers, so minting
+ /// a registry here would hand the two buffers different `Arc`s and split
+ /// every direct partition-plane counter increment by publish parity --
+ /// exactly what [`StatsRegistry`] exists to prevent. Carrying the registry
+ /// across also preserves the `Arc<PartitionStats>` the data plane
+ /// registered at bootstrap and reconcile, which nothing in a snapshot can
+ /// reconstruct (partition counters are not snapshotted).
+ pub(crate) fn restore_in_place(&mut self, snapshot: StreamsSnapshot) {
+ let registry = Arc::clone(&self.stats_registry);
+ // Slab keys are recycled, so an entry left over from a stream the
+ // snapshot does not have would hand its counters to whatever lands in
+ // that slot next.
+ registry.retain_from_snapshot(&snapshot);
+ *self = Self::inner_from_snapshot(snapshot, registry);
+ }
+
+ /// Build a complete `StreamsInner` from a snapshot section against
+ /// `stats_registry`. Shared by wrapper construction
+ /// ([`Snapshotable::from_snapshot`]) and the in-place restore command
+ /// (state transfer), which absorbs it on both left-right buffers.
+ pub(crate) fn inner_from_snapshot(
+ snapshot: StreamsSnapshot,
+ stats_registry: Arc<StatsRegistry>,
+ ) -> Self {
let mut index: AHashMap<Arc<str>, usize> = AHashMap::new();
let mut stream_entries: Vec<(usize, Stream)> = Vec::new();
- // Register restored stats in the shared registry so both left-right
- // buffers (and any post-restore op) reference one `Arc` per
- // stream/topic (see `StatsRegistry`).
- let stats_registry = Arc::new(StatsRegistry::default());
for (slab_key, stream_snap) in snapshot.items {
let stream_stats = stats_registry.stream(slab_key);
@@ -2469,4 +2527,88 @@ mod tests {
buffer.as_mut_slice()[header_size..].copy_from_slice(&body);
Message::try_from(buffer).unwrap()
}
+
+ /// One stream, one topic, one partition, materialized in the registry the
+ /// way the data plane does at bootstrap.
+ fn inner_with_registered_partition() -> StreamsInner {
+ let mut inner = StreamsInner::new();
+ create_stream(&mut inner, "alpha");
+ let create_topic = CreateTopicWithAssignmentsRequest {
+ request: make_topic_request(0, 1, "logs"),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ };
+ let _ = StateHandler::apply(&create_topic, &mut inner,
IggyTimestamp::now());
+ let topic_stats = inner.items[0].topics[0].stats.clone();
+ inner.stats_registry.partition(0, 0, 0, topic_stats);
+ inner
+ }
+
+ // The restore command is absorbed on BOTH left-right buffers. Minting a
+ // registry per call would hand the two buffers different `Arc`s, so a
+ // direct partition-plane increment would land on one buffer and vanish on
+ // the next publish -- the `messages_count_inconsistent` failure the
+ // registry exists to prevent.
+ #[test]
+ fn in_place_restore_keeps_one_registry_across_both_buffers() {
+ let mut first = inner_with_registered_partition();
+ let snapshot = Streams::from(first.clone()).to_snapshot();
+ let mut second = first.clone();
+
+ first.restore_in_place(snapshot.clone());
+ second.restore_in_place(snapshot);
+
+ assert!(
+ Arc::ptr_eq(&first.stats_registry, &second.stats_registry),
+ "both buffers must keep the one shared registry"
+ );
+ let from_first = first
+ .stats_registry
+ .partition_get(0, 0, 0)
+ .expect("survivor keeps its partition stats");
+ let from_second = second
+ .stats_registry
+ .partition_get(0, 0, 0)
+ .expect("survivor keeps its partition stats");
+ assert!(Arc::ptr_eq(&from_first, &from_second));
+ }
+
+ // Partition counters live only in the registry (never snapshotted), so an
+ // install that dropped them would leave every already-materialized
+ // partition reading zeroes with no way to recover them.
+ #[test]
+ fn in_place_restore_keeps_survivor_partition_stats() {
+ let mut inner = inner_with_registered_partition();
+ let stats = inner.stats_registry.partition_get(0, 0,
0).expect("stats");
+ stats.increment_messages_count(42);
+ let snapshot = Streams::from(inner.clone()).to_snapshot();
+
+ inner.restore_in_place(snapshot);
+
+ let after = inner
+ .stats_registry
+ .partition_get(0, 0, 0)
+ .expect("partition survived the restore, so its stats must too");
+ assert!(Arc::ptr_eq(&stats, &after));
+ assert_eq!(after.messages_count_inconsistent(), 42);
+ }
+
+ // Slab keys are recycled: an entry left behind by a stream the snapshot
+ // does not carry would hand its counters to whatever lands in that slot.
+ #[test]
+ fn in_place_restore_prunes_entries_the_snapshot_dropped() {
+ let mut inner = inner_with_registered_partition();
+ // A second stream that the snapshot below will not contain.
+ let empty = Streams::from(StreamsInner::new()).to_snapshot();
+ assert!(inner.stats_registry.partition_get(0, 0, 0).is_some());
+
+ inner.restore_in_place(empty);
+
+ assert!(
+ inner.stats_registry.partition_get(0, 0, 0).is_none(),
+ "a partition the snapshot dropped must not keep its registry entry"
+ );
+ }
}
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index f513e059a..66be115d9 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -819,6 +819,15 @@ impl Snapshotable for Users {
}
impl UsersInner {
+ /// Rebuild from a snapshot section IN PLACE (state transfer), absorbed on
+ /// both left-right buffers.
+ ///
+ /// Nothing here is shared across buffers the way `StreamsInner`'s stats
+ /// registry is, so a wholesale replace is correct.
+ pub(crate) fn restore_in_place(&mut self, snapshot: UsersSnapshot) {
+ *self = Self::inner_from_snapshot(snapshot);
+ }
+
/// Build a complete `UsersInner` from a snapshot section. Shared by
/// wrapper construction ([`Snapshotable::from_snapshot`]) and the
/// in-place restore command (state transfer), which absorbs it on both
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 1da90f58f..eb48863c5 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -1371,7 +1371,7 @@ where
);
assert!(
!consensus.is_transferring(),
- "drain_request_queue_into_prepares: must not be syncing"
+ "drain_request_queue_into_prepares: must not be
transferring state"
);
let prepare = req.message.project(consensus);
consensus.verify_pipeline();
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index fb24564f4..da648fcb6 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -719,6 +719,16 @@ impl Journal<PartitionJournalMemStorage> for
PartitionJournal<PartitionJournalMe
#[rustfmt::skip]
type HeaderRef<'a> = &'a Self::Header;
+ /// No snapshot bookkeeping: the partition plane has no checkpoint of its
+ /// own yet, so nothing supersedes journaled entries. Answered explicitly
+ /// (the trait has no default) so partition-plane state transfer has to
+ /// decide this deliberately rather than inherit it.
+ fn snapshot_op(&self) -> u64 {
+ 0
+ }
+
+ fn set_snapshot_op(&self, _op: u64) {}
+
fn header(&self, idx: usize) -> Option<Self::HeaderRef<'_>> {
let headers = unsafe { &mut *self.headers.get() };
headers.get(idx)
diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs
index 4fd5673db..fe53992d2 100644
--- a/core/partitions/src/log.rs
+++ b/core/partitions/src/log.rs
@@ -80,6 +80,30 @@ where
fn entry(&self, header: &Self::Header) -> impl Future<Output =
Option<Self::Entry>> {
self.inner.entry(header)
}
+
+ // Forward EVERY method, including the ones the trait could default. A
+ // wrapper that silently substitutes a default for its inner journal is a
+ // trap: `snapshot_op` would answer 0 and `set_snapshot_op` would vanish,
+ // so a state-transfer install through this wrapper would neither evict
+ // superseded entries nor advance the watermark it thinks it advanced.
+ fn snapshot_op(&self) -> u64 {
+ self.inner.snapshot_op()
+ }
+
+ fn set_snapshot_op(&self, op: u64) {
+ self.inner.set_snapshot_op(op);
+ }
+
+ fn remaining_capacity(&self) -> Option<usize> {
+ self.inner.remaining_capacity()
+ }
+
+ fn drain(
+ &self,
+ ops: std::ops::RangeInclusive<u64>,
+ ) -> impl Future<Output = std::io::Result<Vec<Self::Entry>>> {
+ self.inner.drain(ops)
+ }
}
impl<J: Default> Default for JournalState<J> {
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index c053d2864..9a731f608 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -177,13 +177,6 @@ impl BinaryTransport for QuicClient {
#[cfg(feature = "vsr")]
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
- // The replayed login/register must mint a fresh Register: the
failed
- // attempt already consumed the one-shot register request id and
may
- // have half-bound the session.
- *self
- .consensus_session
- .lock()
- .expect("consensus session mutex poisoned") =
ConsensusSession::new();
}
let server_address =
self.current_server_address.lock().await.to_string();
info!(
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index d3ee674f2..799580274 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -1779,7 +1779,12 @@ async fn build_shard_for_thread(
// per-shard tunable set once here rather than per consensus group.
shard.set_repair_retry_ticks(repair_retry_ticks(config));
shard.set_repair_chunk_max(config.cluster.repair_chunk_max as u64);
- shard.set_clients_table_max(config.metadata.clients_table_max);
+ // Bounds a served state-transfer chunk. A frame above the bus ceiling is
+ // rejected by the RECEIVING transport, which tears the replica connection
+ // down rather than dropping one message.
+ shard.set_bus_max_message_size(
+
usize::try_from(config.message_bus.max_message_size.as_bytes_u64()).unwrap_or(usize::MAX),
+ );
*shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
Ok((shard, sessions))
}
@@ -1813,6 +1818,10 @@ const _: () = assert!(
);
const _: () =
assert!(configs::ng_cluster::DEFAULT_REPAIR_CHUNK_MAX as u64 ==
shard::REPAIR_CHUNK_MAX);
+const _: () = assert!(
+ configs::ng_cluster::STATE_CHUNK_HEADER_LEN
+ == size_of::<iggy_binary_protocol::consensus::StateChunkHeader>() as
u64
+);
/// Convert a consensus-timer interval to whole ticks, floored at one tick so a
/// sub-tick value still fires and saturated on overflow.
fn duration_to_ticks(interval: Duration) -> u64 {
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index 047816bfa..bfa459c68 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -1340,25 +1340,22 @@ pub(crate) fn build_raw_pat_reply(
return Ok(committed);
}
let header_len = std::mem::size_of::<ReplyHeader>();
- // A `Reply` whose result section is nonzero is not a successful commit
but a
- // committed business rejection or a `TransientNotCommitted` retry frame,
both
- // with no payload and no token to ship. Pass it through so the client
decodes
- // the typed result (and, for a transient, replays) instead of having a raw
- // token grafted onto a rejection body.
- if result_code(&committed.as_slice()[header_len..]) != Some(0) {
- return Ok(committed);
- }
let committed_header =
bytemuck::checked::try_from_bytes::<ReplyHeader>(&committed.as_slice()[..header_len])
.map_err(|_| IggyError::InvalidFormat)?;
let commit = committed_header.commit;
let size = committed_header.size as usize;
- // A committed create can still be a business rejection (duplicate name,
- // invalid expiry) whose reply body carries a nonzero result code. Splice
- // the secret only into a genuine success; pass everything else through
- // untouched so the client decodes the committed error instead of a
- // success-shaped token reply (the minted raw secret is simply dropped).
- // Mirrors the HTTP handler's `committed_payload` gate.
+ // A `Reply` whose result section is nonzero is not a successful commit:
+ // a committed business rejection (duplicate name, invalid expiry) or a
+ // `TransientNotCommitted` retry frame, both with no payload and no token
+ // to ship. Splice the secret only into a genuine success; pass everything
+ // else through untouched so the client decodes the typed result (and, for
+ // a transient, replays) instead of having a raw token grafted onto a
+ // rejection body. Mirrors the HTTP handler's `committed_payload` gate.
+ //
+ // Bounded by the header's own `size` rather than running to the end of the
+ // buffer, so a short frame reads as "no result section" instead of into
+ // allocation padding.
let reply_body = committed
.as_slice()
.get(header_len..size)
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 70980aa01..79bc1ba0d 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -745,11 +745,57 @@ struct MetadataRepairSession {
idle_ticks: u32,
}
+/// The metadata state machine, as every handler that walks or restores it
+/// needs it.
+///
+/// A blanket-implemented alias for a three-part bound that was pasted verbatim
+/// at eleven sites across this file and `router.rs`. No API change: anything
+/// satisfying the parts satisfies this.
+pub trait MetadataStm:
+ StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >
+{
+}
+
+impl<M> MetadataStm for M where
+ M: StreamsFrontend
+ + StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ >
+{
+}
+
+/// [`MetadataStm`] plus in-place snapshot restore: the additional capability a
+/// state-transfer install needs over a plain commit walk.
+pub trait RestorableMetadataStm:
+ MetadataStm
+ +
metadata::stm::snapshot::RestoreSnapshotInPlace<metadata::stm::snapshot::MetadataSnapshot>
+{
+}
+
+impl<M> RestorableMetadataStm for M where
+ M: MetadataStm
+ +
metadata::stm::snapshot::RestoreSnapshotInPlace<metadata::stm::snapshot::MetadataSnapshot>
+{
+}
+
/// Chunk size for state-transfer artifact pulls. Lockstep (one in flight),
-/// so the bounded per-peer bus queue can never drop a burst tail; the bus
-/// message cap is far above this.
+/// so the bounded per-peer bus queue can never drop a burst tail. Clamped
+/// against the live bus ceiling by
+/// [`IggyShard::state_chunk_len_max`] rather than assumed to fit.
const STATE_CHUNK_LEN: u32 = 256 * 1024;
+/// Bus frame ceiling assumed before bootstrap overrides it. Matches the
+/// shipped `[message_bus] max_message_size` so the simulator and unit tests
+/// clamp the same way a default deployment does.
+const DEFAULT_BUS_MAX_MESSAGE_SIZE: usize = 64 * 1024 * 1024;
+
/// Stall rounds a receiver spends on ONE peer before abandoning the transfer
/// and falling back to journal repair. The retry has no peer re-selection, so
/// this is what keeps a peer that died mid-transfer from wedging the rejoining
@@ -763,6 +809,14 @@ const STATE_TRANSFER_MAX_STALL_RETRIES: u32 = 5;
/// protocol has no completion frame) or gave up.
const STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE: u32 = 10;
+/// Lifetime of a FULLY SERVED offer, as a multiple of the repair-retry
+/// interval. It only has to outlive the receiver re-requesting a lost final
+/// chunk, but the receiver's stall re-request fires at exactly one such
+/// interval, so a one-interval grace is a coin flip against its own retry plus
+/// a network hop -- and losing the race costs a full re-pull (`UnknownOffer`
+/// drops the session with every byte already downloaded).
+const STATE_TRANSFER_SERVED_EXPIRY_MULTIPLE: u32 = 3;
+
/// One artifact of an accepted transfer target: its manifest entry plus the
/// bytes received so far (chunks are sequential, so `buf.len()` doubles as
/// the next request offset).
@@ -794,11 +848,6 @@ struct MetadataTransferSession {
/// Whether a descriptor has been accepted (an accepted EMPTY manifest is
/// distinguishable from "still waiting").
target_accepted: bool,
- /// Stall rounds this session has burned. Bounded by
- /// [`STATE_TRANSFER_MAX_STALL_RETRIES`]: the retry always targets the same
- /// `peer`, so a peer that dies mid-transfer would otherwise wedge the
- /// rejoining node forever.
- attempts: u32,
/// Ticks with no frame progress; at the configured repair-retry
/// threshold the missing piece is re-requested.
idle_ticks: u32,
@@ -808,9 +857,12 @@ struct MetadataTransferSession {
/// primary). Keyed by requester replica id so a rebooted requester's fresh
/// nonce replaces the stale offer; chunks must all come from ONE offer or
/// the artifact checksums cannot hold.
+///
+/// The offer itself is refcounted, so simultaneous rejoiners on the same
+/// snapshot generation share one copy of the snapshot bytes.
struct ServedStateTransfer {
nonce: u128,
- offer: metadata::StateTransferOffer,
+ offer: Rc<metadata::StateTransferOffer>,
/// Ticks since this offer last served a chunk. An offer owns a full copy
of
/// the snapshot and the encoded client table, so a completed or abandoned
/// transfer must not pin them for the process lifetime. There is no
@@ -945,17 +997,27 @@ where
/// `[cluster] repair_chunk_max` at bootstrap.
repair_chunk_max: Cell<u64>,
- /// Capacity for a state-transferred client table
- /// ([`consensus::ClientTable::decode`]). Defaults to
- /// [`consensus::CLIENTS_TABLE_MAX`]; server-ng overrides it from
- /// `[metadata] clients_table_max` at bootstrap so the installed table
- /// matches the configured capacity instead of the compile-time default.
- clients_table_max: Cell<usize>,
-
/// Live stalled-repair retry threshold in consensus ticks. Defaults to
/// [`partitions::REPAIR_RETRY_TICKS`]; server-ng overrides it from
/// `[cluster] repair_retry_interval` at bootstrap.
repair_retry_ticks: Cell<u32>,
+
+ /// Live `[message_bus] max_message_size`. Bounds a served state chunk: a
+ /// frame above this is rejected by the RECEIVING transport, which tears
+ /// down the whole replica connection. Defaults to a value that leaves
+ /// [`STATE_CHUNK_LEN`] usable; server-ng overrides it at bootstrap.
+ bus_max_message_size: Cell<usize>,
+
+ /// Consecutive metadata state-transfer rounds that made no progress.
+ ///
+ /// Deliberately NOT on [`MetadataTransferSession`]: three of the four
+ /// arming sites mint a fresh session, so a per-session counter bounded
+ /// nothing -- a permanently undecodable peer cycled abandon -> repair ->
+ /// `RangeEvicted` -> re-arm at zero forever. Held here it survives the
+ /// cycle, and it is reset by actual progress (see
+ /// [`IggyShard::note_metadata_transfer_progress`]) so scattered transient
+ /// stalls cannot accumulate into abandoning a nearly-complete transfer.
+ metadata_transfer_attempts: Cell<u32>,
}
impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T>
@@ -1052,7 +1114,8 @@ where
metadata_transfer_offers: RefCell::new(HashMap::new()),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
- clients_table_max: Cell::new(consensus::CLIENTS_TABLE_MAX),
+ bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
+ metadata_transfer_attempts: Cell::new(0),
})
}
@@ -1070,12 +1133,11 @@ where
self.repair_chunk_max.set(chunk);
}
- /// Override the state-transfer client-table capacity from configuration
- /// (`[metadata] clients_table_max`). Called once per shard at bootstrap;
- /// the simulator and tests keep the compile-time
- /// [`consensus::CLIENTS_TABLE_MAX`] default.
- pub fn set_clients_table_max(&self, max_clients: usize) {
- self.clients_table_max.set(max_clients);
+ /// Override the message-bus frame ceiling from configuration
+ /// (`[message_bus] max_message_size`). Called once per shard at bootstrap;
+ /// the simulator and tests keep the compile-time default.
+ pub fn set_bus_max_message_size(&self, max_message_size: usize) {
+ self.bus_max_message_size.set(max_message_size);
}
/// Hand a metadata consensus submit (login/logout) to shard 0.
@@ -1286,7 +1348,8 @@ where
metadata_transfer_offers: RefCell::new(HashMap::new()),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
- clients_table_max: Cell::new(consensus::CLIENTS_TABLE_MAX),
+ bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
+ metadata_transfer_attempts: Cell::new(0),
}
}
@@ -2104,12 +2167,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- >,
+ M: MetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -2121,9 +2179,12 @@ where
if planes.0.persist_superblock_if_needed(consensus).await {
dispatch_vsr_actions(consensus, planes.0.journal.as_ref(),
&actions).await;
}
+ // Same transfer gate as `on_start_view` and `on_commit`: the
+ // pre-install STM must not walk while a transfer is in flight.
if actions
.iter()
.any(|action| matches!(action, VsrAction::CommitJournal))
+ && !consensus.is_transferring()
{
planes.0.commit_journal().await;
}
@@ -2162,12 +2223,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- >,
+ M: MetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -2176,6 +2232,12 @@ where
&& consensus.namespace() == header.namespace
{
let actions = consensus.handle_start_view(PlaneKind::Metadata,
&header);
+ // Every rejection path (wrong primary, old view, stale
incarnation,
+ // below the commit floor, self-sent) returns no actions, and an
+ // adopted StartView always emits at least `CommitJournal`. That
+ // makes emptiness the adoption signal -- and the arms below must
+ // not fire on a StartView this replica did not adopt.
+ let adopted = !actions.is_empty();
if planes.0.persist_superblock_if_needed(consensus).await {
dispatch_vsr_actions(consensus, planes.0.journal.as_ref(),
&actions).await;
}
@@ -2186,6 +2248,12 @@ where
// snapshot already contains, and the transfer replaces the table
// anyway.
//
+ // Gated on `adopted`: a stale StartView leaves `header.replica`
+ // pointing at a replica that need not be primary, and re-arming on
+ // one would re-mint the nonce (dropping the descriptor already in
+ // flight through the nonce filter) and, before the budget moved
off
+ // the session, reset the retry bound as well.
+ //
// Outside the superblock gate above: that gate fail-closes the VSR
// actions this replica would VOUCH with (notably `PrepareOk`)
until
// the adopted view is durable. Requesting a transfer vouches for
@@ -2193,24 +2261,26 @@ where
// withholds `PrepareOk` on its own (`is_transferring`). Gating it
// would also wedge the one path that repairs a replica whose gap
// sits below every peer's floor.
- if consensus.state_transfer_stage() ==
consensus::StateTransferStage::AwaitingTarget {
- let nonce = iggy_common::random_id::get_uuid();
- *self.metadata_transfer.borrow_mut() =
Some(MetadataTransferSession {
- nonce,
- peer: header.replica,
- commit_op: 0,
- artifacts: Vec::new(),
- target_accepted: false,
- attempts: 0,
- idle_ticks: 0,
- });
+ if adopted
+ && consensus.state_transfer_stage() ==
consensus::StateTransferStage::AwaitingTarget
+ {
tracing::info!(
shard = self.id,
peer = header.replica,
"adopted a live view while awaiting transfer; requesting
metadata state transfer"
);
- self.send_request_state_transfer(consensus, header.replica,
nonce)
- .await;
+ self.arm_metadata_transfer(consensus, header.replica).await;
+ return;
+ }
+ // Mid-transfer the pre-install STM must not walk: the snapshot
+ // being installed already contains those ops, and a walk that
+ // advances `commit_min` past the incoming `snapshot_seq` flips the
+ // install to table-only (no STM restore, no persist, no pairing)
+ // while still reporting success. Landing inside the install's
+ // superblock await instead trips `set_commit_floor`'s anti-rewind
+ // assert. The `AwaitingTarget` return above covers only that one
+ // stage; `Fetching` and `Installing` fall through to here.
+ if consensus.is_transferring() {
return;
}
// `dispatch_vsr_actions` deliberately no-ops `CommitJournal` (it
@@ -2297,12 +2367,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- >,
+ M: MetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -2684,12 +2749,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- >,
+ M: MetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -2762,25 +2822,15 @@ where
if consensus.state_transfer_stage() ==
consensus::StateTransferStage::Idle {
*self.metadata_repair.borrow_mut() = None;
consensus.begin_state_transfer_await();
- let nonce = iggy_common::random_id::get_uuid();
- *self.metadata_transfer.borrow_mut() =
Some(MetadataTransferSession {
- nonce,
- peer: header.replica,
- commit_op: 0,
- artifacts: Vec::new(),
- target_accepted: false,
- attempts: 0,
- idle_ticks: 0,
- });
tracing::info!(
shard = self.id,
peer = header.replica,
retained_from = header.op,
local_commit = consensus.commit_min(),
+ attempts = self.metadata_transfer_attempts.get(),
"metadata repair floor evicted; converting to
state transfer"
);
- self.send_request_state_transfer(consensus,
header.replica, nonce)
- .await;
+ self.arm_metadata_transfer(consensus,
header.replica).await;
} else {
tracing::debug!(
shard = self.id,
@@ -3052,7 +3102,7 @@ where
) where
B: MessageBus,
{
- let manifest = offer.map(|offer|
consensus::encode_state_manifest(&offer.artifacts));
+ let manifest = offer.map(|offer|
consensus::encode_state_manifest(&offer.manifest()));
let total_size =
size_of::<StateTransferTargetHeader>() +
manifest.as_ref().map_or(0, Vec::len);
let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
@@ -3125,12 +3175,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- >,
+ M: MetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -3140,17 +3185,34 @@ where
if consensus.namespace() != header.namespace {
return;
}
- let offer = planes.0.state_transfer_offer();
let cluster = consensus.cluster();
let self_id = consensus.replica();
- if let Some(offer) = offer {
- tracing::info!(
+
+ // First-wins per (requester, nonce). A stall-retry
`RequestStateTransfer`
+ // reuses the session nonce, and rebuilding under it would replace a
+ // manifest the receiver may already have accepted: the client table is
+ // encoded live, so a rebuild that is SHORTER (a client logged out
between
+ // the two builds) lands the receiver's cursor exactly at the new
length
+ // and it re-requests an empty tail forever. Re-answering with the SAME
+ // offer is also what makes the retry idempotent.
+ let cached = self
+ .metadata_transfer_offers
+ .borrow_mut()
+ .get_mut(&header.replica)
+ .filter(|served| served.nonce == header.nonce)
+ .map(|served| {
+ // A descriptor retry proves the requester is alive and still
+ // wants THIS offer, so it counts as liveness: without the
reset
+ // the offer could age out mid-retry and the rebuild that
+ // replaced it is exactly what first-wins exists to prevent.
+ served.idle_ticks = 0;
+ Rc::clone(&served.offer)
+ });
+ if let Some(offer) = cached {
+ tracing::debug!(
shard = self.id,
requester = header.replica,
- commit_op = offer.commit_op,
- artifacts = offer.artifacts.len(),
- total_len = offer.artifacts.iter().map(|a| a.len).sum::<u64>(),
- "serving metadata state transfer"
+ "re-answering a state transfer request from the offer already
served"
);
self.send_state_transfer_target(
cluster,
@@ -3161,31 +3223,60 @@ where
Some(&offer),
)
.await;
- self.metadata_transfer_offers.borrow_mut().insert(
- header.replica,
- ServedStateTransfer {
- nonce: header.nonce,
- offer,
- idle_ticks: 0,
- fully_served: false,
- },
- );
- } else {
- tracing::info!(
- shard = self.id,
- requester = header.replica,
- "cannot serve metadata state transfer (not a caught-up \
- primary, or no snapshot persisted); requester falls back"
- );
- self.send_state_transfer_target(
- cluster,
- self_id,
- header.replica,
- header.nonce,
- header.namespace,
- None,
- )
- .await;
+ return;
+ }
+
+ match planes.0.state_transfer_offer() {
+ Ok(offer) => {
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ commit_op = offer.commit_op,
+ snapshot_seq = offer.snapshot_seq,
+ artifacts = offer.len(),
+ total_len = offer.total_len(),
+ "serving metadata state transfer"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ Some(&offer),
+ )
+ .await;
+ self.metadata_transfer_offers.borrow_mut().insert(
+ header.replica,
+ ServedStateTransfer {
+ nonce: header.nonce,
+ offer,
+ idle_ticks: 0,
+ fully_served: false,
+ },
+ );
+ }
+ Err(reason) => {
+ // Log the ACTUAL reason: "no snapshot yet" is routine and the
+ // requester recovers through journal repair, while an
unreadable
+ // or corrupt `snapshot.bin` is an operator-visible fault on
THIS
+ // node that the old catch-all message actively misattributed.
+ tracing::info!(
+ shard = self.id,
+ requester = header.replica,
+ %reason,
+ "cannot serve metadata state transfer; requester falls
back"
+ );
+ self.send_state_transfer_target(
+ cluster,
+ self_id,
+ header.replica,
+ header.nonce,
+ header.namespace,
+ None,
+ )
+ .await;
+ }
}
}
@@ -3201,14 +3292,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
{
/// Alloc cap per artifact: a corrupt length field must not OOM the
/// shard. Far above any real metadata snapshot or client table.
@@ -3348,6 +3432,10 @@ where
let Some(ref consensus) = planes.0.consensus else {
return;
};
+ // Same clamp the serving side applies, so a bus ceiling below
+ // `STATE_CHUNK_LEN` shrinks the ask instead of leaving the server to
+ // silently serve less than was requested.
+ let chunk_len_max = self.state_chunk_len_max() as u64;
let request = {
let session = self.metadata_transfer.borrow();
session.as_ref().and_then(|session| {
@@ -3362,7 +3450,7 @@ where
let offset = artifact.buf.len() as u64;
let remaining = artifact.entry.len - offset;
#[allow(clippy::cast_possible_truncation)]
- let len = remaining.min(u64::from(STATE_CHUNK_LEN)) as u32;
+ let len = remaining.min(chunk_len_max) as u32;
#[allow(clippy::cast_possible_truncation)]
Some((session.nonce, session.peer, index as u32, offset, len))
})
@@ -3382,6 +3470,68 @@ where
}
}
+ /// Arm a fresh metadata transfer session against `peer` and request its
+ /// descriptor.
+ ///
+ /// Every arming site goes through here. Three near-identical session
+ /// literals had already drifted on the retry budget, which is why that
+ /// budget now lives on the shard ([`Self::metadata_transfer_attempts`])
+ /// instead of being re-minted with each session.
+ #[allow(clippy::future_not_send)]
+ async fn arm_metadata_transfer<P>(&self, consensus: &VsrConsensus<B, P>,
peer: u8)
+ where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ *self.metadata_transfer.borrow_mut() = Some(MetadataTransferSession {
+ nonce,
+ peer,
+ commit_op: 0,
+ artifacts: Vec::new(),
+ target_accepted: false,
+ idle_ticks: 0,
+ });
+ self.send_request_state_transfer(consensus, peer, nonce)
+ .await;
+ }
+
+ /// Largest state-chunk PAYLOAD this side will put on the wire.
+ ///
+ /// Clamped so header + payload stays inside the bus ceiling. Above it the
+ /// RECEIVING transport rejects the frame and tears down the entire replica
+ /// connection, which surfaces to an operator as an unexplained link flap.
+ /// Both ends derive their chunk size from this same function, so a bus cap
+ /// below [`STATE_CHUNK_LEN`] shrinks the chunk rather than making large
+ /// artifacts untransferable.
+ fn state_chunk_len_max(&self) -> usize {
+ let budget = self
+ .bus_max_message_size
+ .get()
+ .saturating_sub(size_of::<StateChunkHeader>());
+ // A bus cap at or below one header cannot carry a chunk at all. Serve
+ // one byte at a time rather than zero: a zero-length chunk is the
+ // livelock `on_request_state_chunk` refuses, and the boot validator
+ // rejects this configuration anyway.
+ budget.clamp(1, STATE_CHUNK_LEN as usize)
+ }
+
+ /// Burn one retry round; `true` once the budget is exhausted.
+ fn burn_metadata_transfer_attempt(&self) -> bool {
+ let attempts = self.metadata_transfer_attempts.get() + 1;
+ self.metadata_transfer_attempts.set(attempts);
+ attempts > STATE_TRANSFER_MAX_STALL_RETRIES
+ }
+
+ /// Real progress: reset the retry budget.
+ ///
+ /// The budget bounds CONSECUTIVE failures, not lifetime ones. Without this
+ /// five stalls scattered across a large transfer would abandon one that
was
+ /// nearly done, throwing away every byte already pulled.
+ fn note_metadata_transfer_progress(&self) {
+ self.metadata_transfer_attempts.set(0);
+ }
+
/// Serve one chunk out of the cached offer. An unknown nonce (offer
/// evicted, e.g. the serving process restarted) answers with an
/// `available = 0` descriptor so the requester restarts its session.
@@ -3401,6 +3551,12 @@ where
let cluster = consensus.cluster();
let self_id = consensus.replica();
+ // Never serve a frame the receiving transport will reject: anything
past
+ // `max_message_size` tears down the whole replica connection, which
reads
+ // as an unexplained link flap. Bounded by the requester's own ask,
this
+ // side's chunk size, and what the bus will carry.
+ let chunk_len_max = self.state_chunk_len_max();
+
// Frame built inside the borrow; every send runs after it drops (a
// RefCell borrow must not cross an await on the shard).
// Out-of-bounds requests are dropped silently inside the block.
@@ -3410,39 +3566,54 @@ where
.get_mut(&header.replica)
.filter(|served| served.nonce == header.nonce);
served.map_or(Some(ChunkReply::UnknownOffer), |served| {
- // Serving a chunk is the only liveness signal the offer gets;
- // the expiry sweep drops it once these stop arriving.
- served.idle_ticks = 0;
// Manifest-index addressing: an index past the offer is a
// requester bug (or a stale frame) and is dropped below.
- let last_artifact =
served.offer.payloads.len().saturating_sub(1);
- let artifact_bytes = served.offer.payloads.get(header.artifact
as usize)?;
+ let last_artifact = served.offer.len().saturating_sub(1);
+ let artifact_bytes = served.offer.payload(header.artifact as
usize)?;
let start = header.offset as usize;
- let end = start.saturating_add(header.len as usize);
- // Tail of the final artifact: the receiver now holds
everything
+ // A request AT the end of an artifact has nothing left to
serve.
+ // Answering it with `Some(&[])` -- which `get(len..len)`
happily
+ // returns -- would extend nothing on the receiver, reset both
+ // sides' idle counters, and be re-requested at the same offset
+ // forever: an unbounded empty-frame ping-pong with the
rejoining
+ // replica withholding `PrepareOk` for the life of the process.
+ // Reachable when a rebuilt offer is SHORTER than the manifest
the
+ // receiver accepted (a client logged out between the two
builds).
+ if start >= artifact_bytes.len() {
+ return None;
+ }
+ let end = start
+ .saturating_add((header.len as usize).min(chunk_len_max))
+ .min(artifact_bytes.len());
+ let payload = artifact_bytes.get(start..end)?;
+ // Only now that bytes are actually going out: an
out-of-bounds or
+ // stale frame must not flip a live offer onto the short
expiry.
+ // Tail of the final artifact means the receiver holds
everything
// the manifest promised, so the offer only has to outlive a
// possible re-request of this very chunk.
if header.artifact as usize == last_artifact && end >=
artifact_bytes.len() {
served.fully_served = true;
}
- artifact_bytes
- .get(start..end.min(artifact_bytes.len()))
- .map(|payload| {
- let total_size = size_of::<StateChunkHeader>() +
payload.len();
- let mut chunk =
Message::<StateChunkHeader>::new(total_size);
- chunk.as_mut_slice()[size_of::<StateChunkHeader>()..]
- .copy_from_slice(payload);
- ChunkReply::Chunk(chunk.transmute_header(|_, h: &mut
StateChunkHeader| {
- h.command = Command2::StateChunk;
- h.cluster = cluster;
- h.replica = self_id;
- h.nonce = header.nonce;
- h.namespace = header.namespace;
- h.artifact = header.artifact;
- h.offset = header.offset;
- h.size = total_size as u32;
- }))
- })
+ // Serving a chunk is the only liveness signal the offer gets;
+ // the expiry sweep drops it once these stop arriving. Set here
+ // rather than on entry so a request that serves NOTHING cannot
+ // keep an abandoned offer alive.
+ served.idle_ticks = 0;
+ let total_size = size_of::<StateChunkHeader>() + payload.len();
+ let mut chunk = Message::<StateChunkHeader>::new(total_size);
+
chunk.as_mut_slice()[size_of::<StateChunkHeader>()..].copy_from_slice(payload);
+ Some(ChunkReply::Chunk(chunk.transmute_header(
+ |_, h: &mut StateChunkHeader| {
+ h.command = Command2::StateChunk;
+ h.cluster = cluster;
+ h.replica = self_id;
+ h.nonce = header.nonce;
+ h.namespace = header.namespace;
+ h.artifact = header.artifact;
+ h.offset = header.offset;
+ h.size = total_size as u32;
+ },
+ )))
})
};
match reply {
@@ -3492,14 +3663,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
{
let header = *msg.header();
let planes = self.plane.inner();
@@ -3536,9 +3700,19 @@ where
);
return;
}
+ // A zero-byte payload is not progress: it extends nothing and the
+ // same offset is re-requested immediately. Resetting the liveness
+ // counters on one is what turned a short rebuilt offer into an
+ // unbounded empty-frame ping-pong. The serving side refuses to
+ // produce these now; the guard stays because a peer running an
+ // older build still can.
+ if payload.is_empty() {
+ return;
+ }
artifact.buf.extend_from_slice(payload);
session.idle_ticks = 0;
}
+ self.note_metadata_transfer_progress();
self.on_transfer_progress().await;
}
@@ -3557,19 +3731,35 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StreamsFrontend
- + StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
{
let planes = self.plane.inner();
let Some(ref consensus) = planes.0.consensus else {
return;
};
+ // The stage is the authority on whether this transfer is still wanted,
+ // and it can be cleared from OUTSIDE this file: the probe-exhausted
+ // election fallback lives in the consensus crate, which cannot reach
+ // `metadata_transfer`, so it drops the stage to `Idle` (legal from
+ // `Fetching`, hence silent) and leaves the session armed with its
nonce
+ // intact. Chunks then keep arriving, the pull completes, and the
+ // `Installing` transition below asserts on an `Idle -> Installing`
edge
+ // that takes down shard 0. Drop the abandoned session here instead --
+ // this is the single funnel both descriptor acceptance and chunk
+ // arrival pass through.
+ let stage = consensus.state_transfer_stage();
+ if stage != consensus::StateTransferStage::Fetching {
+ if self.metadata_transfer.borrow().is_some() {
+ tracing::info!(
+ shard = self.id,
+ ?stage,
+ "metadata state transfer was abandoned out from under its
session; \
+ dropping it"
+ );
+ *self.metadata_transfer.borrow_mut() = None;
+ }
+ return;
+ }
let complete = {
let session = self.metadata_transfer.borrow();
match session.as_ref() {
@@ -3592,7 +3782,6 @@ where
.expect("session checked above");
let peer = session.peer;
let commit_op = session.commit_op;
- let attempts = session.attempts;
// Per-artifact integrity, then pick the pieces this plane installs.
// Unknown kinds are refused rather than skipped: an artifact the
@@ -3633,10 +3822,23 @@ where
let decoded = if damaged {
None
} else if let (Some(snapshot), Some((table_bytes, table_frontier))) =
(snapshot, table) {
- match consensus::ClientTable::decode(&table_bytes,
self.clients_table_max.get()) {
+ // Decode against the LIVE table's capacity, not a separately
+ // plumbed config cell. The serving primary's table can
legitimately
+ // hold more entries than this node's configured cap --
heterogeneous
+ // config, or homogeneous config after a cap reduction, since
+ // `from_snapshot` sizes capacity to `max(min_slots, highest_slot +
+ // 1)`. Against the raw config value that is a deterministic
+ // `TooManyEntries` every round: a permanent, log-only join
failure.
+ let capacity = planes.0.client_table_capacity();
+ match consensus::ClientTable::decode(&table_bytes, capacity) {
Ok(table) => Some((snapshot, table, table_frontier)),
Err(error) => {
- tracing::error!(shard = self.id, %error, "transferred
client table undecodable");
+ tracing::error!(
+ shard = self.id,
+ capacity,
+ %error,
+ "transferred client table undecodable"
+ );
None
}
}
@@ -3652,17 +3854,16 @@ where
// Damage is usually transit corruption, which a re-fetch fixes.
But
// it can also be permanent -- a peer whose artifacts this build
// cannot decode, or an unknown artifact kind -- and that re-offers
- // identically every round. Carry the attempt count across the
- // restart and share the stall budget, so a re-fetch that keeps
- // failing gives up instead of pulling the whole snapshot forever.
- // Distinct from the stall path: frames ARE flowing here, so
+ // identically every round. The budget lives on the shard, so it
+ // survives the abandon -> repair -> re-arm cycle and a re-fetch
that
+ // keeps failing gives up instead of pulling the whole snapshot
+ // forever. Distinct from the stall path: frames ARE flowing here,
so
// `idle_ticks` never accumulates and that sweep can never fire.
- let attempts = attempts + 1;
- if attempts > STATE_TRANSFER_MAX_STALL_RETRIES {
+ if self.burn_metadata_transfer_attempt() {
tracing::warn!(
shard = self.id,
peer,
- attempts,
+ attempts = self.metadata_transfer_attempts.get(),
"state transfer artifacts kept failing to decode;
abandoning \
and falling back to journal repair"
);
@@ -3678,18 +3879,7 @@ where
if consensus.state_transfer_stage() ==
consensus::StateTransferStage::Fetching {
consensus.set_state_transfer_stage(consensus::StateTransferStage::AwaitingTarget);
}
- let nonce = iggy_common::random_id::get_uuid();
- *self.metadata_transfer.borrow_mut() =
Some(MetadataTransferSession {
- nonce,
- peer,
- commit_op: 0,
- artifacts: Vec::new(),
- target_accepted: false,
- attempts,
- idle_ticks: 0,
- });
- self.send_request_state_transfer(consensus, peer, nonce)
- .await;
+ self.arm_metadata_transfer(consensus, peer).await;
return;
};
@@ -3699,42 +3889,41 @@ where
.install_state_transfer(&snapshot, table, table_frontier,
commit_op)
.await
{
- Ok(applied_frontier) => {
+ Ok(outcome) => {
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
- // `applied_frontier`, not the transferred snapshot's op: the
install
- // returns `max(snapshot_seq, local_applied)`, which differs
whenever a
- // serving peer offers a snapshot BEHIND this replica
(checkpoints are
- // node-local) and the local state machine is kept instead.
- tracing::info!(
- shard = self.id,
- applied_frontier,
- commit_op,
- table_frontier,
- "metadata state transfer installed; handing tail to
journal repair"
- );
+ // A completed install: the budget starts fresh for any later
+ // rejoin rather than carrying this one's stalls forward.
+ self.note_metadata_transfer_progress();
+ if outcome.pairing_durable {
+ // `applied_frontier`, not the transferred snapshot's op:
the install
+ // returns `max(snapshot_seq, local_applied)`, which
differs whenever a
+ // serving peer offers a snapshot BEHIND this replica
(checkpoints are
+ // node-local) and the local state machine is kept instead.
+ tracing::info!(
+ shard = self.id,
+ applied_frontier = outcome.applied_frontier,
+ commit_op,
+ table_frontier,
+ "metadata state transfer installed; handing tail to
journal repair"
+ );
+ } else {
+ // Deliberately NOT prefixed with the success line's text:
+ // the specs match log substrings, so a shared prefix would
+ // let every one of them pass on the degraded path.
+ tracing::warn!(
+ shard = self.id,
+ applied_frontier = outcome.applied_frontier,
+ commit_op,
+ table_frontier,
+ "metadata state transfer landed WITHOUT a durable
checkpoint \
+ pairing; the next superblock write records it"
+ );
+ }
// Walk whatever is already walkable, then let repair fetch
// the (snapshot_seq, commit_max] tail.
planes.0.commit_journal().await;
self.maybe_request_metadata_repair(consensus, peer).await;
}
- // Installed, but its `(checkpoint_op, checksum)` pairing is not
durable
- // yet. Not a failed install: the snapshot, table and frontiers
are all in
- // place, and the coordinator already holds the new pairing, so
the next
- // superblock write (view change or checkpoint) records it. Until
then a
- // crash recovers the PREVIOUS checkpoint and this replica
transfers again,
- // which is correct, just wasted work. Continue as a success.
- Err(metadata::stm::snapshot::SnapshotError::SuperblockNotDurable {
op }) => {
-
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
- tracing::warn!(
- shard = self.id,
- snapshot_seq = op,
- commit_op,
- "metadata state transfer installed but its checkpoint
pairing is \
- not durable yet; the next superblock write records it"
- );
- planes.0.commit_journal().await;
- self.maybe_request_metadata_repair(consensus, peer).await;
- }
Err(error) => {
tracing::error!(
shard = self.id,
@@ -3886,14 +4075,15 @@ where
// built and breaking transfers outright.
let retry_ticks = self.repair_retry_ticks.get().max(1);
let idle_expiry_ticks =
retry_ticks.saturating_mul(STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE);
+ let served_expiry_ticks =
retry_ticks.saturating_mul(STATE_TRANSFER_SERVED_EXPIRY_MULTIPLE);
let mut offers = self.metadata_transfer_offers.borrow_mut();
offers.retain(|requester, served| {
served.idle_ticks += 1;
// A fully-served offer only has to outlive a re-request of its
last
- // chunk, so it goes after one retry interval; anything else is an
+ // chunk, so it goes on the short clock; anything else is an
// abandoned transfer and waits out the full idle window.
let expiry_ticks = if served.fully_served {
- retry_ticks
+ served_expiry_ticks
} else {
idle_expiry_ticks
};
@@ -3908,6 +4098,11 @@ where
}
live
});
+ // Nobody is pulling: release the cached snapshot copy too, rather than
+ // pinning it for the life of the process.
+ if offers.is_empty() {
+ self.plane.metadata().clear_state_transfer_offer_cache();
+ }
}
#[allow(clippy::future_not_send)]
@@ -3968,23 +4163,19 @@ where
return None;
}
session.idle_ticks = 0;
- session.attempts += 1;
- Some((
- session.peer,
- session.nonce,
- session.target_accepted,
- session.attempts,
- ))
+ Some((session.peer, session.nonce, session.target_accepted))
})
};
- if let Some((peer, nonce, target_accepted, attempts)) =
transfer_stalled {
+ if let Some((peer, nonce, target_accepted)) = transfer_stalled {
+ let exhausted = self.burn_metadata_transfer_attempt();
+ let attempts = self.metadata_transfer_attempts.get();
// Retrying the same peer forever is a wedge when that peer is the
// thing that died: nothing in this loop re-selects a target. Give
up
// after a bounded number of rounds and fall back to journal
repair,
// which re-picks a peer and, if the gap is still below its
retained
// floor, answers `RangeEvicted` and arms a fresh transfer against
// whoever is primary now.
- if attempts > STATE_TRANSFER_MAX_STALL_RETRIES {
+ if exhausted {
tracing::warn!(
shard = self.id,
peer,
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 75d829e71..37f265045 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -19,15 +19,13 @@ use crate::metrics::{frame_drop_reason, frame_drop_variant};
use crate::shards_table::{
ShardsTable, calculate_shard_assignment, calculate_shard_from_consensus_ns,
};
-use crate::{IggyShard, LifecycleFrame, Receiver, ShardFrame};
+use crate::{IggyShard, LifecycleFrame, Receiver, RestorableMetadataStm,
ShardFrame};
use consensus::{MetadataHandle, PartitionsHandle};
use crossfire::TrySendError;
use futures::FutureExt;
use iggy_binary_protocol::{ConsensusHeader, GenericHeader, Operation,
PrepareHeader};
use journal::{Journal, JournalHandle};
use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn};
-use metadata::impls::metadata::StreamsFrontend;
-use metadata::stm::StateMachine;
use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
use server_common::{Message, MessageBag};
@@ -301,14 +299,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + StreamsFrontend
- + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
{
// Reused across every pump iteration; pre-size to skip the
// first-drain reallocation.
@@ -440,14 +431,7 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
- M: StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + StreamsFrontend
- + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
{
match frame {
ShardFrame::Consensus { message, .. } => {
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index b557d8bdc..de7c2240e 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -176,6 +176,16 @@ impl<S: Storage<Buffer = Vec<u8>>> Journal<S> for
SimJournal<S> {
where
Self: 'a;
+ /// The simulated journal retains everything for the run, so nothing is
+ /// ever superseded by a snapshot. Answered explicitly (the trait has no
+ /// default) so a simulated state transfer has to opt into a watermark
+ /// rather than silently inherit one that never moves.
+ fn snapshot_op(&self) -> u64 {
+ 0
+ }
+
+ fn set_snapshot_op(&self, _op: u64) {}
+
// TODO(hubcio): validate that the caller's checksum matches the stored
// header - currently this looks up by op only, ignoring the checksum.
// A real journal implementation must reject mismatches.
