This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch partition_superblock
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/partition_superblock by this
push:
new ed61c3e51 address review comments
ed61c3e51 is described below
commit ed61c3e511622b43b71e91091f680500961eb32f
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Aug 6 11:19:01 2026 +0200
address review comments
The superblock grew an offset frontier field without a decode that tolerates
the old length, so every record written before this change refused boot. The
decode now zero-fills a 58-byte record instead, and the layout guard gives
the
new field a distinct value so a transposed write cannot pass.
Two partition transfer knobs shipped in decimal units and had no floor: an
artifact cap below one legal segment plus one maximum-size batch rejects a
legal transfer deterministically, which reads as a rejoin that never
completes. Both are now binary units, validated at the config level against
the configured segment size and the bus message cap, and pinned to the
runtime
constants at the server-ng build edge.
Boot no longer takes the whole node down for one partition: an untrustworthy
superblock tombstones that group the way a refused segment chain already
does,
and the index-less recovery walk verifies each batch against the filename
anchor before believing its header, so a torn header cannot become the
partition's offset counter. Its two directory walks are fused and its
per-batch
file open is hoisted out of the loop.
On the test side, the transfer specs now assert on the installed bytes and
on
marker counts taken before the fault rather than on whole-log substring
matches, the partition view-durability oracle reads its expectation from the
survivor that was never restarted, and the simulator re-materialises
partitions
on restart so the recovered-view branch it claims to cover is actually
reached.
---
core/binary_protocol/src/consensus/header.rs | 10 +-
core/configs/src/server_ng_config/partition.rs | 51 ++++-
core/configs/src/server_ng_config/validators.rs | 28 +++
core/consensus/src/lib.rs | 4 +-
core/consensus/src/vsr_state.rs | 86 +++++++--
core/integration/src/bench_utils.rs | 13 +-
.../src/harness/orchestrator/harness.rs | 17 ++
.../tests/cluster/partition_state_transfer.rs | 138 ++++++++++++--
.../tests/server/cluster_view_durability_vsr.rs | 17 +-
.../tests/server/partition_view_durability_vsr.rs | 68 +++----
.../server/scenarios/purge_delete_scenario.rs | 32 ++--
core/journal/src/local_gate.rs | 22 ++-
core/journal/src/superblock.rs | 10 +-
core/partitions/src/iggy_partition.rs | 58 +++++-
core/partitions/src/offset_storage.rs | 21 ++-
core/partitions/src/state_transfer.rs | 207 +++++++++++++--------
core/server-ng/config.toml | 22 ++-
core/server-ng/src/bootstrap.rs | 44 ++++-
core/server-ng/src/partition_helpers.rs | 51 +++--
core/server-ng/src/segment_recovery.rs | 128 ++++++++-----
core/server-ng/src/server_error.rs | 23 ++-
core/shard/src/lib.rs | 204 ++++++++++++--------
core/shard/src/metrics.rs | 8 +
core/shard/src/router.rs | 19 +-
core/simulator/src/lib.rs | 97 ++++++----
25 files changed, 959 insertions(+), 419 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index e71f1004c..f26e5c71e 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1426,11 +1426,11 @@ pub struct StateTransferTargetHeader {
/// load.
///
/// This and `commit_max` below claim the HEAD of what used to be the
- /// reserved tail, so every pre-existing field keeps its published offset:
- /// this header ships in the `iggy_binary_protocol` crate, the size assert
- /// cannot catch an equal-size reshuffle, and nothing on the link carries a
- /// version signal -- a mid-struct insertion is silent non-interop between
- /// mixed builds.
+ /// reserved tail, so every pre-existing field keeps its published offset.
+ /// Layout compatibility only: the size assert cannot catch an equal-size
+ /// reshuffle, so a mid-struct insertion would silently move every field
+ /// after it. It says nothing about the semantics of these two -- an older
+ /// peer presents zeros here and serves no partition transfers at all.
pub unavailable_transient: u8,
/// Explicit padding so `commit_max` sits 8-aligned without the implicit
/// padding `NoUninit` forbids.
diff --git a/core/configs/src/server_ng_config/partition.rs
b/core/configs/src/server_ng_config/partition.rs
index 66d428fd7..87f8755c9 100644
--- a/core/configs/src/server_ng_config/partition.rs
+++ b/core/configs/src/server_ng_config/partition.rs
@@ -55,6 +55,18 @@ pub const DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH: usize = 32;
/// sizing endorsement.
pub const MAX_PARTITION_PREPARE_QUEUE_DEPTH: usize = 256;
+/// Mirrors `shard::IggyShard::PARTITION_ARTIFACT_LEN_DEFAULT` (segment ceiling
+/// plus the one whole batch a segment may close past it).
+pub const DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX: u64 = 1024 * 1024 * 1024 + 64 *
1024 * 1024;
+
+/// Mirrors `shard::ServedSegmentCache::RESIDENT_BYTES_DEFAULT`.
+pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 = 2 * 1024 * 1024 *
1024;
+
+/// Upper bound on the two state-transfer byte knobs. A typo guard, not a
sizing
+/// endorsement: both are PER SHARD, so a slipped digit multiplies by the core
+/// count.
+pub const MAX_TRANSFER_BYTES: u64 = 64 * 1024 * 1024 * 1024;
+
/// Mirrors `partitions::EVICTED_RING_CAPACITY`.
pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 4096;
@@ -142,12 +154,24 @@ impl Validatable<ConfigurationError> for PartitionConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
- if self.transfer_served_cache_bytes_max.as_bytes_u64() == 0 {
- eprintln!("{COMPONENT_NG}
partition.transfer_served_cache_bytes_max must be > 0");
+ // The FLOOR on `transfer_artifact_bytes_max` cannot live here (it
needs
+ // `system.segment.size` and the bus cap); it is enforced in the
+ // `ServerNgConfig` validator, which is what turns that
misconfiguration
+ // into a boot error instead of a silent per-partition rejoin livelock.
+ let served_cache = self.transfer_served_cache_bytes_max.as_bytes_u64();
+ if served_cache == 0 || served_cache > MAX_TRANSFER_BYTES {
+ eprintln!(
+ "{COMPONENT_NG} partition.transfer_served_cache_bytes_max
({served_cache} bytes) \
+ must be > 0 and <= {MAX_TRANSFER_BYTES} bytes"
+ );
return Err(ConfigurationError::InvalidConfigurationValue);
}
- if self.transfer_artifact_bytes_max.as_bytes_u64() == 0 {
- eprintln!("{COMPONENT_NG} partition.transfer_artifact_bytes_max
must be > 0");
+ let artifact_bytes = self.transfer_artifact_bytes_max.as_bytes_u64();
+ if artifact_bytes == 0 || artifact_bytes > MAX_TRANSFER_BYTES {
+ eprintln!(
+ "{COMPONENT_NG} partition.transfer_artifact_bytes_max
({artifact_bytes} bytes) \
+ must be > 0 and <= {MAX_TRANSFER_BYTES} bytes"
+ );
return Err(ConfigurationError::InvalidConfigurationValue);
}
let ring_bytes = self.evicted_ring_bytes_max.as_bytes_u64();
@@ -176,6 +200,25 @@ mod tests {
assert!(PartitionConfig::default().validate().is_ok());
}
+ /// The shipped TOML strings are the only thing an operator sees, and
nothing
+ /// else ties them to the constants the code sizes itself against -- a
+ /// decimal/binary slip ("1088 MB" for 1088 MiB) parses fine and ships a
cap
+ /// BELOW the largest legal segment, which livelocks a rejoin per
partition.
+ #[test]
+ fn shipped_transfer_defaults_match_the_runtime_constants() {
+ let config = PartitionConfig::default();
+ assert_eq!(
+ config.transfer_artifact_bytes_max.as_bytes_u64(),
+ DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX,
+ "config.toml transfer_artifact_bytes_max drifted from the runtime
default"
+ );
+ assert_eq!(
+ config.transfer_served_cache_bytes_max.as_bytes_u64(),
+ DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX,
+ "config.toml transfer_served_cache_bytes_max drifted from the
runtime default"
+ );
+ }
+
#[test]
fn rejects_zero_prepare_queue_depth() {
let config = PartitionConfig {
diff --git a/core/configs/src/server_ng_config/validators.rs
b/core/configs/src/server_ng_config/validators.rs
index 8c6941f5e..f7408b7d3 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -183,6 +183,34 @@ impl Validatable<ConfigurationError> for ServerNgConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // A received segment artifact can be one whole batch larger than the
+ // segment cap (rotation checks the cap AFTER appending), and the real
+ // batch bound is the BUS frame cap -- server-ng never enforces
+ // `MAX_PAYLOAD_SIZE`. An artifact ceiling under that floor refuses a
+ // legal segment, and the manifest check is all-or-nothing, so the
+ // partition livelocks re-requesting the same segment from every peer
at
+ // the backoff ceiling. Caught here so it is a boot error rather than
one
+ // partition that silently never rejoins.
+ let artifact_floor = self
+ .system
+ .segment
+ .size
+ .as_bytes_u64()
+ .saturating_add(self.message_bus.max_message_size.as_bytes_u64());
+ if self.partition.transfer_artifact_bytes_max.as_bytes_u64() <
artifact_floor {
+ eprintln!(
+ "{COMPONENT_NG} partition.transfer_artifact_bytes_max ({} B)
must be at least \
+ system.segment.size ({} B) + message_bus.max_message_size ({}
B) = \
+ {artifact_floor} B: a segment may close one whole batch past
its cap, and an \
+ artifact ceiling below that refuses a legal segment and
livelocks the \
+ partition's rejoin",
+ self.partition.transfer_artifact_bytes_max.as_bytes_u64(),
+ self.system.segment.size.as_bytes_u64(),
+ self.message_bus.max_message_size.as_bytes_u64(),
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
self.message_bus
.validate()
.error(|e: &ConfigurationError| {
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index 10b0425c8..4d7a53d53 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -155,8 +155,8 @@ pub use client_table::{
};
pub mod state_manifest;
pub use state_manifest::{
- StateArtifact, StateManifestError, artifact_kind, decode_state_manifest,
encode_state_manifest,
- state_artifact_checksum,
+ StateArtifact, StateArtifactHasher, StateManifestError, artifact_kind,
decode_state_manifest,
+ encode_state_manifest, state_artifact_checksum,
};
pub mod state_transfer;
pub use state_transfer::{
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
index bdef87c32..1044dc55e 100644
--- a/core/consensus/src/vsr_state.rs
+++ b/core/consensus/src/vsr_state.rs
@@ -28,12 +28,24 @@
use std::fmt;
-/// Number of bytes [`VsrState::to_bytes`] produces and [`VsrState::try_from`]
-/// expects: `cluster`(16) + `replica_id`(1) + `replica_count`(1) + `view`(4)
-/// + `log_view`(4) + `commit_max`(8) + `checkpoint_op`(8)
-/// + `checkpoint_checksum`(16) + `offset_frontier`(8).
+/// Number of bytes [`VsrState::to_bytes`] produces: `cluster`(16) +
+/// `replica_id`(1) + `replica_count`(1) + `view`(4) + `log_view`(4) +
+/// `commit_max`(8) + `checkpoint_op`(8) + `checkpoint_checksum`(16) +
+/// `offset_frontier`(8).
pub const ENCODED_LEN: usize = 66;
+/// The layout before `offset_frontier` was appended.
+///
+/// [`VsrState::try_from`] still accepts records of this length and zero-fills
+/// the new field. Without it every superblock already on disk -- the metadata
+/// plane writes one on every view change and checkpoint, single-node included
--
+/// would decode as [`VsrStateError::WrongLength`] and refuse boot as a
+/// durability violation. A version bump instead of this would not help on its
+/// own: `classify` compares the version for exact equality, so a v2 build
turns
+/// every v1 record into `Unreadable`, which is the same refusal wearing a
+/// different name.
+pub const ENCODED_LEN_WITHOUT_FRONTIER: usize = 58;
+
/// The durable consensus state of one replica for one consensus group.
///
/// A view a replica acted in must survive a crash, or it can re-participate in
@@ -109,13 +121,25 @@ impl TryFrom<&[u8]> for VsrState {
type Error = VsrStateError;
fn try_from(bytes: &[u8]) -> Result<Self, Self::Error> {
- // One length check up front puts every field slice below in bounds by
- // construction, so the `try_into`s cannot fail.
- let bytes: &[u8; ENCODED_LEN] =
- bytes.try_into().map_err(|_| VsrStateError::WrongLength {
- expected: ENCODED_LEN,
- actual: bytes.len(),
- })?;
+ // Length-tolerant: a pre-`offset_frontier` record is padded out and
the
+ // new field reads as 0, which is exactly "no recorded frontier" (the
+ // read sites filter it). One length check up front then puts every
+ // field slice below in bounds by construction, so the `try_into`s
+ // cannot fail.
+ let mut padded = [0u8; ENCODED_LEN];
+ match bytes.len() {
+ ENCODED_LEN => padded.copy_from_slice(bytes),
+ ENCODED_LEN_WITHOUT_FRONTIER => {
+ padded[..ENCODED_LEN_WITHOUT_FRONTIER].copy_from_slice(bytes);
+ }
+ actual => {
+ return Err(VsrStateError::WrongLength {
+ expected: ENCODED_LEN,
+ actual,
+ });
+ }
+ }
+ let bytes = &padded;
let state = Self {
cluster: u128::from_le_bytes(field(bytes, 0)),
replica_id: bytes[16],
@@ -213,6 +237,41 @@ mod tests {
assert!(VsrState::try_from(&bytes[..ENCODED_LEN - 1]).is_err());
}
+ /// A superblock written before `offset_frontier` existed must still
decode:
+ /// the metadata plane writes one on every view change, so an exact-length
+ /// decode turns an in-place upgrade into a boot refusal on every
deployment
+ /// that ever ran.
+ #[test]
+ fn given_pre_frontier_record_when_decoded_should_accept_and_zero_fill() {
+ let full = VsrState {
+ cluster: 3,
+ replica_id: 1,
+ replica_count: 3,
+ view: 9,
+ log_view: 8,
+ commit_max: 41,
+ checkpoint_op: 7,
+ checkpoint_checksum: 5,
+ offset_frontier: 77,
+ }
+ .to_bytes();
+
+ let legacy = &full[..ENCODED_LEN_WITHOUT_FRONTIER];
+ let decoded = VsrState::try_from(legacy).expect("a pre-frontier record
must decode");
+ assert_eq!(decoded.offset_frontier, 0, "the new field zero-fills");
+ assert_eq!(decoded.view, 9);
+ assert_eq!(decoded.log_view, 8);
+ assert_eq!(decoded.commit_max, 41);
+ assert_eq!(decoded.checkpoint_op, 7);
+ assert_eq!(decoded.checkpoint_checksum, 5);
+
+ // Anything that is neither layout is still refused.
+ assert!(matches!(
+ VsrState::try_from(&full[..40]),
+ Err(VsrStateError::WrongLength { .. })
+ ));
+ }
+
#[test]
fn given_log_view_past_view_when_decoded_should_reject() {
// Corruption inside the checksummed region can produce a length-valid
record
@@ -228,9 +287,12 @@ mod tests {
commit_max: 0,
checkpoint_op: 0,
checkpoint_checksum: 0,
- offset_frontier: 0,
+ // Distinct and nonzero: with 0 here a transposed write over the
+ // trailing field would still satisfy every assertion below.
+ offset_frontier: 9,
}
.to_bytes();
+ assert_eq!(bytes[58], 9, "offset_frontier must occupy bytes 58..66");
bytes[22] = 5; // log_view = 5, view stays 4
assert_eq!(
diff --git a/core/integration/src/bench_utils.rs
b/core/integration/src/bench_utils.rs
index 74c2d56bd..de2de07d1 100644
--- a/core/integration/src/bench_utils.rs
+++ b/core/integration/src/bench_utils.rs
@@ -31,11 +31,14 @@ const BENCH_FILES_PREFIX: &str = "bench_";
const MESSAGE_BATCHES: u64 = 100;
const MESSAGES_PER_BATCH: u64 = 100;
const DEFAULT_NUMBER_OF_STREAMS: u64 = 8;
-// Generous for a few MB of traffic even in debug builds. Exists because a
-// protocol mismatch (an SDK framing the server does not speak, e.g. a
-// default-features iggy-bench against a vsr cluster) hangs both sides
-// silently instead of erroring.
-const BENCH_WAIT_TIMEOUT: Duration = Duration::from_secs(600);
+// Generous for a few MB of traffic even in debug builds, and deliberately
+// UNDER nextest's harness timeout (`.config/nextest.toml` sigkills at
+// 60s x 5): a longer wait here would never fire, taking the capture dump and
+// the `--features vsr` hint below with it. Exists because a protocol mismatch
+// (an SDK framing the server does not speak, e.g. a default-features
+// iggy-bench against a vsr cluster) hangs both sides silently instead of
+// erroring.
+const BENCH_WAIT_TIMEOUT: Duration = Duration::from_secs(240);
pub fn run_bench_and_wait_for_finish(
server_addr: &str,
diff --git a/core/integration/src/harness/orchestrator/harness.rs
b/core/integration/src/harness/orchestrator/harness.rs
index 435ebfc81..4a661578a 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -518,6 +518,23 @@ impl TestHarness {
.await
}
+ /// Root-authenticated TCP client bound to ONE node of a cluster, unlike
+ /// [`Self::root_client_for`], which always targets node 0.
+ ///
+ /// # Errors
+ ///
+ /// [`TestBinaryError::MissingServer`] when `index` is out of range, or the
+ /// underlying connect/login failure.
+ pub async fn root_client_for_node(&self, index: usize) ->
Result<IggyClient, TestBinaryError> {
+ self.servers
+ .get(index)
+ .ok_or(TestBinaryError::MissingServer)?
+ .tcp_client()?
+ .with_root_login()
+ .connect()
+ .await
+ }
+
/// Create a new client logged in as root for the specified transport.
pub fn client_builder_for(
&self,
diff --git a/core/integration/tests/cluster/partition_state_transfer.rs
b/core/integration/tests/cluster/partition_state_transfer.rs
index 55ac09306..0d3647917 100644
--- a/core/integration/tests/cluster/partition_state_transfer.rs
+++ b/core/integration/tests/cluster/partition_state_transfer.rs
@@ -85,7 +85,10 @@ const MARKER_POLL: Duration = Duration::from_millis(200);
async fn
given_evicted_ring_when_fresh_node_joins_late_should_state_transfer_partition(
harness: &mut TestHarness,
) {
- let client = connect(harness, 0).await;
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
seed_partition(&client).await;
client
.store_consumer_offset(
@@ -186,7 +189,10 @@ async fn
given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
// Node 2 holds a durable prefix, then misses enough traffic that the
// survivors' ring moves past its durable end: its repaired window cannot
// connect, which is exactly the refusal-site trigger.
- let client = connect(harness, 0).await;
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
seed_topic(&client).await;
produce(&client, 40).await;
sleep(Duration::from_secs(1)).await;
@@ -204,6 +210,21 @@ async fn
given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
// buffered non-blocking appender, so a survivor's line can trail node 2's
// install by more than one poll.
await_marker_any(harness, &[0, 1], FULLY_SERVED_MARKER).await;
+
+ // The markers only say the machinery ran. Read node 2's OWN segment bytes
+ // and check every produced payload is there, in ascending file order: a
+ // truncated, short or reordered install fails here and passes above.
+ //
+ // Read off disk rather than polled: the SDK is leader-aware and redirects
+ // a poll to the primary, so no client-side read can be pinned to the
+ // rejoined node.
+ // Two produce runs, each numbering its payloads from 0, so the expected
+ // chain is 0..40 followed by 0..MESSAGES_COUNT.
+ let expected: Vec<String> = (0..40)
+ .chain(0..MESSAGES_COUNT)
+ .map(|sequence| format!("message-{sequence}"))
+ .collect();
+ await_installed_payloads(harness, 2, &expected).await;
}
#[iggy_harness(
@@ -225,7 +246,10 @@ async fn
given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partition(
harness: &mut TestHarness,
) {
- let client = connect(harness, 0).await;
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
seed_topic(&client).await;
// Bulky payloads so the pull spans many 256 KiB chunks: the kill below
// must land while the transfer is provably in flight, and a small
@@ -258,19 +282,30 @@ async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partit
);
sleep(KILL_GATE_POLL).await;
}
+ // Baselines BEFORE the kill. Every marker check below counts occurrences
+ // against these instead of scanning the whole accumulated log: node 2 can
+ // have installed and node 1 can have fully served an earlier attempt while
+ // node 0 was still up, and a `contains` would call those the recovery.
+ let installs_before_kill =
harness.node(2).stdout_occurrences(INSTALL_MARKER);
+ let served_before_kill =
harness.node(1).stdout_occurrences(FULLY_SERVED_MARKER);
+ let abandons_before_kill =
harness.node(2).stdout_occurrences(ABANDON_MARKER);
harness
.stop_node(0)
.expect("stop the serving peer (node 0)");
// The abandon is now deterministic: the pull was in flight against a
// peer that is gone, so the stall budget must exhaust.
- await_marker(harness, 2, ABANDON_MARKER).await;
+ await_new_marker(harness, 2, ABANDON_MARKER, abandons_before_kill).await;
// Recovery: the scheduled re-arm targets the surviving primary. No
// follow-up commit is asserted -- the cluster is quorum-marginal with
// one node down, and an unanswered read mid-election is not a verdict.
- await_marker(harness, 2, INSTALL_MARKER).await;
- await_marker_any(harness, &[1], FULLY_SERVED_MARKER).await;
+ //
+ // Node 1 is named explicitly, not "any survivor": with node 0 dead it is
+ // the only replica left that can serve, so a marker from it is proof the
+ // re-arm found a new peer rather than proof of the pre-kill attempt.
+ await_new_marker(harness, 2, INSTALL_MARKER, installs_before_kill).await;
+ await_new_marker(harness, 1, FULLY_SERVED_MARKER,
served_before_kill).await;
// The manifest the install consumed carried one artifact per sealed
// segment, and each was spilled and renamed separately: the seeded 64 MiB
@@ -291,18 +326,6 @@ async fn
given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partit
}
}
-/// Connect a root-authenticated TCP client to a specific node.
-async fn connect(harness: &TestHarness, node: usize) -> IggyClient {
- harness
- .node(node)
- .tcp_client()
- .expect("tcp client builder")
- .with_root_login()
- .connect()
- .await
- .unwrap_or_else(|e| panic!("connect to node {node}: {e}"))
-}
-
async fn connect_any(harness: &TestHarness, nodes: &[usize]) ->
Option<IggyClient> {
for &node in nodes {
if let Ok(builder) = harness.node(node).tcp_client()
@@ -395,6 +418,85 @@ async fn poll_count(client: &IggyClient, count: u32) ->
Result<u32, IggyError> {
Ok(polled.messages.len() as u32)
}
+/// Polls node `node`'s installed segment files until every payload in
+/// `expected` is present, in the order given.
+///
+/// The payloads are the oracle the markers are not: a short install is missing
+/// the tail, a torn one is missing a middle, and a reordered one fails the
+/// ascending-position check.
+async fn await_installed_payloads(harness: &TestHarness, node: usize,
expected: &[String]) {
+ let data_path = harness.node(node).data_path();
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ loop {
+ if let Err(missing) = installed_payloads_complete(&data_path,
expected) {
+ assert!(
+ Instant::now() < deadline,
+ "node {node}'s installed segments never held the whole
produced batch: {missing}"
+ );
+ sleep(MARKER_POLL).await;
+ continue;
+ }
+ return;
+ }
+}
+
+/// `Ok(())` when every payload in `expected` appears in node-local segment
+/// bytes at a non-decreasing position, otherwise the first discrepancy.
+fn installed_payloads_complete(data_path: &Path, expected: &[String]) ->
Result<(), String> {
+ let mut chain = Vec::new();
+ let mut paths = Vec::new();
+ let _ = walk(data_path, &mut |path| {
+ if is_segment_log(path) {
+ paths.push(path.to_path_buf());
+ }
+ false
+ });
+ // Segment files are named for their zero-padded base offset, so lexical
+ // order is offset order.
+ paths.sort();
+ for path in paths {
+ let Ok(bytes) = std::fs::read(&path) else {
+ return Err(format!("{} could not be read", path.display()));
+ };
+ chain.extend_from_slice(&bytes);
+ }
+ let mut searched_from = 0;
+ for payload in expected {
+ let found = chain[searched_from..]
+ .windows(payload.len())
+ .position(|window| window == payload.as_bytes())
+ // A bare find would match `message-1` inside `message-10`.
+ .map(|offset| searched_from + offset)
+ .filter(|start| {
+ chain
+ .get(start + payload.len())
+ .is_none_or(|byte| !byte.is_ascii_digit())
+ });
+ let Some(start) = found else {
+ return Err(format!(
+ "{payload:?} is absent from the {} installed bytes after
position {searched_from}",
+ chain.len()
+ ));
+ };
+ searched_from = start;
+ }
+ Ok(())
+}
+
+/// [`await_marker`], but satisfied only by an occurrence beyond `baseline` -
+/// the whole-log scan cannot distinguish a line from before the fault.
+async fn await_new_marker(harness: &TestHarness, node: usize, marker: &str,
baseline: usize) {
+ let deadline = Instant::now() + TRANSFER_BUDGET;
+ while harness.node(node).stdout_occurrences(marker) <= baseline {
+ assert!(
+ Instant::now() < deadline,
+ "node {node} never logged {marker:?} again after the fault \
+ (still at the pre-fault count of {baseline}) within
{TRANSFER_BUDGET:?}"
+ );
+ sleep(MARKER_POLL).await;
+ }
+}
+
/// [`await_marker`] over a set of nodes: satisfied by the first one to log it.
async fn await_marker_any(harness: &TestHarness, nodes: &[usize], marker:
&str) {
let deadline = Instant::now() + TRANSFER_BUDGET;
diff --git a/core/integration/tests/server/cluster_view_durability_vsr.rs
b/core/integration/tests/server/cluster_view_durability_vsr.rs
index d69d673e7..a2cc801f4 100644
--- a/core/integration/tests/server/cluster_view_durability_vsr.rs
+++ b/core/integration/tests/server/cluster_view_durability_vsr.rs
@@ -73,7 +73,10 @@ async fn
given_advanced_metadata_view_when_survivor_restarts_should_recover_view
// the primary of view 0 is replica 0). Commit a stream through it, so the
// metadata group has committed state to recover later and exactly one
leader
// is visible.
- let client = connect(harness, 0).await;
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
client
.create_stream(STREAM_NAME)
.await
@@ -155,18 +158,6 @@ async fn
given_advanced_metadata_view_when_survivor_restarts_should_recover_view
);
}
-/// Connect a root-authenticated TCP client to a specific node.
-async fn connect(harness: &TestHarness, node: usize) -> IggyClient {
- harness
- .node(node)
- .tcp_client()
- .expect("tcp client builder")
- .with_root_login()
- .connect()
- .await
- .unwrap_or_else(|e| panic!("connect to node {node}: {e}"))
-}
-
/// Connect to the first node in `nodes` that accepts a connection, `None`
when none
/// do (mid-election, or a node still restarting).
async fn connect_any(harness: &TestHarness, nodes: &[usize]) ->
Option<IggyClient> {
diff --git a/core/integration/tests/server/partition_view_durability_vsr.rs
b/core/integration/tests/server/partition_view_durability_vsr.rs
index acbbb8171..007147399 100644
--- a/core/integration/tests/server/partition_view_durability_vsr.rs
+++ b/core/integration/tests/server/partition_view_durability_vsr.rs
@@ -64,7 +64,10 @@ async fn
given_advanced_partition_view_when_survivor_restarts_should_recover_vie
// Baseline: node 0 is the view-0 primary of every group (its replica id
is 0).
// Commit a topic with ONE partition and a batch of messages through it, so
// exactly one partition consensus group exists and holds committed state.
- let client = connect(harness, 0).await;
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("connect a root client to the node");
client
.create_stream(STREAM_NAME)
.await
@@ -120,11 +123,6 @@ async fn
given_advanced_partition_view_when_survivor_restarts_should_recover_vie
// `view >= 1`) is the settled form: the gate persists `view` the moment a
// replica advances it to vote, before it adopts the new primary's log.
let view_before = wait_for_advanced_partition_view(harness, 2).await;
- assert!(
- view_before.view >= 1 && view_before.log_view >= 1,
- "a survivor that took part in the partition view change must persist \
- view/log_view >= 1, got {view_before:?}"
- );
// Bring the crashed primary back so it rejoins from its own disk:
partition
// recovery opens the superblock, restores its recorded view, and the boot
@@ -162,28 +160,22 @@ async fn
given_advanced_partition_view_when_survivor_restarts_should_recover_vie
// The recovered replica's own BEHAVIOR, not the file's contents: the
// superblock is read only at boot and written only when the persist gate
- // fires, so re-reading it here would return the pre-restart record even if
- // recovery were broken and node 2 came back at view 0. Node 2's boot line
- // reports the view it actually restored, so it must name the recorded one.
+ // fires, so re-reading node 2's own record here would return the
+ // pre-restart bytes even if recovery were broken and node 2 came back at
+ // view 0. Node 2's boot line reports the view it actually restored.
//
- // Skipped when the harness inherits the node's stdout instead of capturing
- // it (`IGGY_TEST_VERBOSE`): the log file is then empty, and asserting on
it
- // would fail spuriously in exactly the mode someone debugging this would
- // use. The serve check above still ran.
- let log_captured = !harness.node(2).stdout_plain().is_empty();
- if !log_captured {
- eprintln!(
- "IGGY_TEST_VERBOSE inherits node stdout, so the restored-view
oracle is \
- unavailable; skipping it"
- );
- return;
- }
+ // The expectation is read from NODE 1, which was never restarted:
comparing
+ // node 2's restored view against node 2's own file would be `x >= x` and
+ // would pass with the restore path deleted.
+ let expected =
read_partition_superblock_state(&harness.node(1).data_path())
+ .expect("the survivor that never restarted holds the cluster's
recorded view");
let restored = restored_partition_view(harness, 2)
.expect("node 2 must log the partition view it restored from its
superblock");
assert!(
- restored.0 >= view_before.view && restored.1 >= view_before.log_view,
- "node 2 restored (view {}, log_view {}) but its superblock recorded
{view_before:?}; \
- a replica that resumes below its recorded view can re-enter a view it
already acted in",
+ restored.0 >= expected.view && restored.1 >= expected.log_view,
+ "node 2 restored (view {}, log_view {}) but the untouched survivor's
record is \
+ {expected:?}; a replica that resumes below the view it already acted
in can \
+ re-enter it",
restored.0,
restored.1
);
@@ -192,10 +184,20 @@ async fn
given_advanced_partition_view_when_survivor_restarts_should_recover_vie
/// `(view, log_view)` the node reports restoring at boot, parsed out of the
/// structured fields of its restore line. `None` while the line is absent.
///
-/// The stdout log is truncated per process start, so a value read after a
-/// restart was logged by the new process.
+/// Reads the node's OWN log file as well as the harness stdout capture: under
+/// `IGGY_TEST_VERBOSE` the child inherits stdout and the capture is empty, and
+/// an oracle that silently skips in the mode someone debugging this would run
+/// is not an oracle.
fn restored_partition_view(harness: &TestHarness, node: usize) -> Option<(u32,
u32)> {
- let log = harness.node(node).stdout_plain();
+ let mut log = harness.node(node).stdout_plain();
+ let own_logs = harness.node(node).data_path().join("logs");
+ if let Ok(entries) = std::fs::read_dir(own_logs) {
+ for entry in entries.flatten() {
+ if let Ok(contents) = std::fs::read_to_string(entry.path()) {
+ log.push_str(&contents);
+ }
+ }
+ }
log.lines()
.filter(|line| line.contains(RESTORED_VIEW_MARKER))
.filter_map(|line| {
@@ -216,18 +218,6 @@ fn field(line: &str, key: &str) -> Option<u32> {
.and_then(|digits| digits.parse().ok())
}
-/// Connect a root-authenticated TCP client to a specific node.
-async fn connect(harness: &TestHarness, node: usize) -> IggyClient {
- harness
- .node(node)
- .tcp_client()
- .expect("tcp client builder")
- .with_root_login()
- .connect()
- .await
- .unwrap_or_else(|e| panic!("connect to node {node}: {e}"))
-}
-
/// Poll the whole pre-crash batch from offset 0; `Ok(count)` of messages seen.
async fn poll_all(client: &IggyClient) -> Result<u32, IggyError> {
let polled = client
diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
index 606002701..8c65c3b10 100644
--- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
+++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
@@ -975,6 +975,11 @@ pub async fn run_purge_topic(harness: &mut TestHarness,
restart_server: bool) {
.purge_topic(&stream_ident, &topic_ident)
.await
.unwrap();
+ // Sampled BEFORE the restart: if the purge already drained the offset
+ // directories, a restart may not resurrect them, and the assert below
stays
+ // instant even in the restart cells. Only the kill-lands-mid-purge case
+ // earns a tolerance.
+ let drained_before_restart = is_dir_empty(&consumers_dir) &&
is_dir_empty(&groups_dir);
maybe_restart(harness, restart_server).await;
// server-ng purges asynchronously (metadata commit -> reconciler -> pump);
@@ -985,22 +990,17 @@ pub async fn run_purge_topic(harness: &mut TestHarness,
restart_server: bool) {
await_segment_layout(&partition_path, &[0]).await;
// --- Verify consumer offsets cleared (memory + disk) ---
- // Polled ONLY in the restart cells: there the kill can land mid-purge, and
- // boot then plants the [0] layout itself (fencing a torn chain, or
- // recovering an already-drained directory) with the offset files still
- // present -- the layout gate above is satisfied BEFORE the reconciler's
- // re-purge (the applied generation is not persisted, so a restart
- // re-purges) clears them. Without a restart the pump clears offsets and
- // files in the SAME frame that plants the layout, so the instant assert is
- // correct there and strictly stronger; a poll would hide a regression that
- // clears them one frame late.
- // vsr-only, and only for the restart cells: the legacy flavor purges
- // synchronously, and without a restart the pump clears offsets and files
in
- // the SAME frame that plants the layout, so the instant assert is correct
- // and strictly stronger there. Kept short -- a client-visible stale offset
- // after purge-then-restart is a real (bounded) window, not something to
- // paper over with a long tolerance.
- let poll_window = if cfg!(feature = "vsr") && restart_server {
+ // ZERO tolerance everywhere except one cell: vsr + restart where the kill
+ // landed mid-purge. There boot plants the [0] layout itself (fencing a
torn
+ // chain, or recovering an already-drained directory) with the offset files
+ // still present, so the layout gate above is satisfied BEFORE the
+ // reconciler's re-purge clears them (the applied generation is not
+ // persisted, so a restart re-purges). Everywhere else the pump clears
+ // offsets and files in the SAME frame that plants the layout, and a poll
+ // would hide a regression that clears them one frame late. Kept short --
+ // a client-visible stale offset after purge-then-restart is a real
+ // (bounded) window, not something to paper over with a long tolerance.
+ let poll_window = if cfg!(feature = "vsr") && restart_server &&
!drained_before_restart {
std::time::Duration::from_secs(2)
} else {
std::time::Duration::ZERO
diff --git a/core/journal/src/local_gate.rs b/core/journal/src/local_gate.rs
index 6d2b7b86e..ca9998f69 100644
--- a/core/journal/src/local_gate.rs
+++ b/core/journal/src/local_gate.rs
@@ -32,6 +32,12 @@ use std::cell::{Cell, RefCell};
/// See the module docs. Callers hold the returned guard across the awaited
/// critical section; dropping it releases the gate and wakes every waiter.
+///
+/// Its exclusion is load-bearing in RELEASE, not only under
+/// `debug_assertions`: this gate is the only enforcement of the superblock
+/// single-writer contract there. The `WritingGuard` tripwire is debug-only,
+/// `PingPongSuperblock::write` takes `&self`, and two overlapping writers
+/// collide on one fixed `.tmp` path and tear a slot while both return `Ok`.
pub struct LocalGate {
busy: Cell<bool>,
waiters: RefCell<Vec<std::task::Waker>>,
@@ -74,9 +80,14 @@ impl<'a> std::future::Future for LocalGateAcquire<'a> {
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Self::Output> {
if self.gate.busy.get() {
- // Re-polls while still busy push a duplicate waker; the extra
- // wake is spurious and harmless at pipeline-queue scale.
- self.gate.waiters.borrow_mut().push(cx.waker().clone());
+ // Deduped, not appended: a waiter driven by a multi-source driver
+ // (`select!`, `join!`) is re-polled on every unrelated wake, and
+ // release wakes the whole list, so a bare push makes draining n
+ // contenders O(n^2) waker clones.
+ let mut waiters = self.gate.waiters.borrow_mut();
+ if !waiters.iter().any(|waiter| waiter.will_wake(cx.waker())) {
+ waiters.push(cx.waker().clone());
+ }
std::task::Poll::Pending
} else {
self.gate.busy.set(true);
@@ -103,11 +114,6 @@ impl Drop for LocalGateGuard<'_> {
}
}
-// NOT behind `cfg(debug_assertions)`: release is exactly the build where this
-// gate is the only enforcement of the superblock single-writer contract (the
-// `WritingGuard` tripwire is debug-only, `write` takes `&self`, and two
-// overlapping writers collide on one fixed `.tmp` path and tear a slot while
-// both return `Ok`).
#[cfg(test)]
mod tests {
use super::*;
diff --git a/core/journal/src/superblock.rs b/core/journal/src/superblock.rs
index 78d04f92c..fd8b76124 100644
--- a/core/journal/src/superblock.rs
+++ b/core/journal/src/superblock.rs
@@ -70,10 +70,12 @@ const MIN_RECORD_LEN: usize = HEADER_LEN + CHECKSUM_LEN;
/// Ceiling on a record's payload, bounding every allocation this module makes
from
/// a length it read off disk (`PrepareJournal::MAX_ENTRY_SIZE` bounds the WAL
for the
-/// same reason). The only payload today is a 58-byte [`VsrState`]; the
headroom is
-/// for a payload that grows fields, not for bulk data. `read_slot` treats a
longer
-/// file as corrupt WITHOUT reading it, and `build_record` refuses to write
one, so a
-/// length this store could have produced is always in bounds.
+/// same reason). The only payload today is a [`consensus::VsrState`], 66
bytes now
+/// that it carries the offset frontier (58 before it, a length its decode
still
+/// accepts); the headroom is for a payload that grows fields, not for bulk
data.
+/// `read_slot` treats a longer file as corrupt WITHOUT reading it, and
+/// `build_record` refuses to write one, so a length this store could have
+/// produced is always in bounds.
const MAX_PAYLOAD_LEN: usize = 4096;
/// Largest record `read_slot` will read into memory.
const MAX_RECORD_LEN: usize = HEADER_LEN + MAX_PAYLOAD_LEN + CHECKSUM_LEN;
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 85ceae6de..0981bf093 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -501,6 +501,16 @@ where
/// THIS partition's group is fenced; the rest of the node keeps serving.
#[allow(clippy::future_not_send)]
async fn write_superblock(&self, superblock: &SB, offset_frontier: u64) ->
bool {
+ // ADVANCE direction: never below what this replica has already minted.
+ // The reset direction (purge) goes through `write_superblock_inner`.
+ let advanced = offset_frontier.max(self.offset_frontier());
+ self.write_superblock_inner(superblock, advanced).await
+ }
+
+ /// The write itself; the advance and reset directions differ only in the
+ /// frontier they hand in.
+ #[allow(clippy::future_not_send)]
+ async fn write_superblock_inner(&self, superblock: &SB, offset_frontier:
u64) -> bool {
// The pairing fields stay `(0, 0)` and `commit_max` is a dead write
// on this plane: nothing reads either back (`restore_partition_view`
// restores view/log_view only), because recovery re-derives the
@@ -514,9 +524,7 @@ where
// view change, or the explicit persist an install issues) leaves a
// lower bound boot can re-seed from.
let mut state = self.consensus.vsr_state(0, 0);
- // Never regresses: a caller recording an incoming frontier passes a
- // larger value, and the ordinary gate passes the live counter.
- state.offset_frontier = offset_frontier.max(self.offset_frontier());
+ state.offset_frontier = offset_frontier;
match superblock.write(&state.to_bytes()).await {
Ok(()) => {
self.consensus
@@ -584,6 +592,26 @@ where
.await
}
+ /// Record a frontier that may be LOWER than the one already on disk.
+ ///
+ /// The frontier is conditionally monotone: it advances everywhere except a
+ /// purge, which legitimately resets the offset space to 0. The advancing
+ /// form cannot express that -- it maxes against the live counter -- and
the
+ /// distinction has to be explicit: a purge that leaves the old frontier
+ /// recorded makes the next boot re-seed the counter to the state the purge
+ /// just erased, and the following append stamps `base_offset` N where
every
+ /// peer stamps 0.
+ #[allow(clippy::future_not_send)]
+ pub async fn reset_offset_frontier(&self) -> bool {
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ let frontier = self.offset_frontier();
+ self.write_superblock_inner(superblock.as_ref(), frontier)
+ .await
+ }
+
/// [`Self::persist_offset_frontier`] for a frontier this replica has not
/// reached yet.
///
@@ -3504,8 +3532,19 @@ where
crate::state_transfer::sweep_staging_except(&partition_dir,
&[]).await;
}
- // Recreate a fresh empty segment at offset 0 with real writers.
let start_offset = 0u64;
+ // Counters reset BEFORE the fallible plant, not after: `?` on
+ // `install_empty_segment` would otherwise leave the live counter at
the
+ // pre-purge value, which is what the router's purge-failure fence then
+ // records and what a restart would re-seed. Safe to reorder --
+ // `install_empty_segment` takes `start_offset` as a parameter and
never
+ // reads the counter, and the partition write lock is held across this
+ // whole body.
+ self.offset.store(start_offset, Ordering::Release);
+ self.dirty_offset.store(start_offset, Ordering::Relaxed);
+ self.should_increment_offset = false;
+
+ // Recreate a fresh empty segment at offset 0 with real writers.
self.install_empty_segment(config, start_offset).await?;
// Make the unlinks AND the replanted dirent durable together: without
// this a crash can resurrect pre-purge segments until the boot
re-purge
@@ -3513,11 +3552,6 @@ where
if let Some(partition_dir) = self.partition_dir.clone() {
let _ = crate::state_transfer::fsync_dir(&partition_dir).await;
}
-
- // Reset the offset counters so new messages start at offset 0.
- self.offset.store(start_offset, Ordering::Release);
- self.dirty_offset.store(start_offset, Ordering::Relaxed);
- self.should_increment_offset = false;
// The boot-time durable line marks recovered bytes that must not be
// re-persisted, but the purge just deleted those bytes and offsets
// restart at 0. Keeping it would make every post-purge batch at or
@@ -3592,6 +3626,12 @@ where
// Same commit frontier, different (now empty) bytes: a cached offer
// built pre-purge would advertise files the purge just unlinked.
self.transfer_offer_cache.borrow_mut().take();
+ // RESET, not advance: the durable frontier still names the pre-purge
+ // offset space, and leaving it there makes the next boot re-seed the
+ // counter to the state this purge just erased -- after which the first
+ // append stamps `base_offset` N while every peer stamps 0. The live
+ // counter is 0 by now, so the reset records 0.
+ self.reset_offset_frontier().await;
Ok(())
}
diff --git a/core/partitions/src/offset_storage.rs
b/core/partitions/src/offset_storage.rs
index ebd4cf9de..393843544 100644
--- a/core/partitions/src/offset_storage.rs
+++ b/core/partitions/src/offset_storage.rs
@@ -25,9 +25,11 @@ use std::path::Path;
const OFFSET_SIZE: usize = core::mem::size_of::<u64>();
pub async fn persist_offset(path: &str, offset: u64, enforce_fsync: bool) ->
Result<(), IggyError> {
- if let Some(parent) = Path::new(path).parent()
- && !parent.exists()
- {
+ // No `exists()` probe first: that is a BLOCKING `std::path` stat on the
pump
+ // in front of every write, which serialises a batched fan-out on stats
+ // before it can submit any I/O. `create_dir_all` is already a no-op on an
+ // existing directory.
+ if let Some(parent) = Path::new(path).parent() {
create_dir_all(parent).await.map_err(|_| {
IggyError::CannotCreateConsumerOffsetsDirectory(parent.display().to_string())
})?;
@@ -114,13 +116,14 @@ async fn read_persisted_offset(path: &str) ->
Result<Option<u64>, IggyError> {
/// # Errors
/// Returns [`IggyError::CannotDeleteConsumerOffsetFile`] if the unlink fails.
pub async fn delete_persisted_offset(path: &str) -> Result<(), IggyError> {
- if !Path::new(path).exists() {
- return Ok(());
+ // NotFound is tolerated on the result instead of probed for: the probe was
+ // a blocking stat on the pump before every unlink, and "already gone" is
+ // exactly the outcome this wants anyway.
+ match remove_file(path).await {
+ Ok(()) => Ok(()),
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
+ Err(_) =>
Err(IggyError::CannotDeleteConsumerOffsetFile(path.to_owned())),
}
-
- remove_file(path)
- .await
- .map_err(|_|
IggyError::CannotDeleteConsumerOffsetFile(path.to_owned()))
}
#[cfg(test)]
diff --git a/core/partitions/src/state_transfer.rs
b/core/partitions/src/state_transfer.rs
index 37ef73409..8f943f0e2 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -35,7 +35,7 @@ use crate::{IggyIndexWriter, IggyPartition};
use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
use consensus::le_cursor::{LeCursor, Truncated, split_verified_trailer};
use consensus::state_manifest::artifact_kind;
-use consensus::{ArtifactProgress, Sequencer as _, state_artifact_checksum};
+use consensus::{ArtifactProgress, Sequencer as _, StateArtifactHasher,
state_artifact_checksum};
use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize};
use journal::superblock::SuperblockStore;
use message_bus::MessageBus;
@@ -220,14 +220,14 @@ pub(crate) struct SegmentChecksumMemo {
/// The stamp is NOT cached alongside: `StateArtifactHasher::finish` takes
/// `&self`, so it is a read of this hasher, and a second copy is just a
/// field that can drift.
- hasher: consensus::state_manifest::StateArtifactHasher,
+ hasher: StateArtifactHasher,
}
impl SegmentChecksumMemo {
fn new() -> Self {
Self {
hashed_len: 0,
- hasher: consensus::state_manifest::StateArtifactHasher::new(),
+ hasher: StateArtifactHasher::new(),
}
}
}
@@ -730,10 +730,16 @@ pub(crate) async fn walk_segment_payload(
let mut since_yield = 0usize;
while position < bytes.len() {
// The walk re-hashes every message (`decode_batch_slice` verifies
- // `batch_checksum`), so a multi-GiB artifact would hold the pump --
and
- // with it consensus ticks and heartbeats for every group on this core
--
- // for the whole pass. Yield on the same cadence the serving side's
- // chunked hash uses.
+ // `batch_checksum`), so a multi-GiB artifact is a long CPU pass on the
+ // pump task. What these yields buy is NOT tick liveness: the consensus
+ // tick is a sibling `select_biased!` arm of this same task and arms
are
+ // not polled while another arm's body awaits, so every group's tick
and
+ // heartbeat on this shard stay frozen for the duration either way (see
+ // the tick-starvation TODO in `shard::router`). They buy the reactor:
+ // detached tasks and io_uring completions make progress instead of
+ // waiting out the whole pass. Moving the verify + walk off the pump is
+ // what would fix the tick, and the nonce re-check after the spill is
+ // already shaped for that.
if since_yield >= OFFER_HASH_CHUNK_LEN {
since_yield = 0;
yield_to_reactor().await;
@@ -1004,6 +1010,12 @@ pub enum PartitionInstallError {
commit_op: u64,
commit_min: u64,
},
+ /// The incoming frontier could not be made durable before the swap, so the
+ /// install refuses rather than enter a window whose only durable witness
+ /// would be the segments the failure path quarantines away.
+ FrontierNotDurable {
+ frontier: u64,
+ },
/// The offer's offset frontier is below this replica's own offset
/// counter, so installing it would rewind the offset space: the next
/// replicated prepare would be re-stamped from the rewound counter and
@@ -1059,6 +1071,10 @@ impl fmt::Display for PartitionInstallError {
f,
"transfer frontier {commit_op} is below the local commit
frontier {commit_min}"
),
+ Self::FrontierNotDurable { frontier } => write!(
+ f,
+ "could not record the incoming offset frontier {frontier}
before the swap"
+ ),
Self::OfferRewindsDurableData {
offer_next_offset,
local_next_offset,
@@ -1101,7 +1117,8 @@ impl From<ConsumerOffsetsWireError> for
PartitionInstallError {
/// Suffix marking a half-transferred file inside the partition directory.
///
/// Provably invisible to boot recovery, which filters on `extension == "log"`,
-/// and swept wholesale at boot (`segment_recovery::sweep_scratch_files`).
+/// and swept wholesale at boot by
+/// `segment_recovery::sweep_scratch_files_and_collect_offsets`.
pub const STAGING_SUFFIX: &str = ".staging";
/// Staging-file names inside the partition directory.
@@ -1116,6 +1133,24 @@ fn staging_paths(partition_dir: &str, start_offset: u64)
-> (PathBuf, PathBuf) {
)
}
+/// Every entry of one partition directory, as paths.
+///
+/// BLOCKING `read_dir` on the pump: compio-fs 0.12 exposes no async directory
+/// walk, and `spawn_blocking` is not an escape either -- the shard executors
run
+/// `thread_pool_limit(0)`. Bounded by the entry count of ONE partition
directory,
+/// but it is a real stall (and under the write lock at the converge site), so
it
+/// stays recorded rather than hidden.
+///
+/// Enumeration only: the three callers keep their own predicates and their own
+/// error policies (propagate / silent skip / log-and-fail), which is what
+/// `sweep_staging_except`'s do-not-widen warning depends on.
+fn segment_dir_entries(partition_dir: &str) -> std::io::Result<Vec<PathBuf>> {
+ Ok(std::fs::read_dir(partition_dir)?
+ .flatten()
+ .map(|entry| entry.path())
+ .collect())
+}
+
/// Move every segment file in `partition_dir` aside into `<dir>.fenced.<n>/`,
/// returning the directory used.
///
@@ -1138,29 +1173,30 @@ fn staging_paths(partition_dir: &str, start_offset:
u64) -> (PathBuf, PathBuf) {
/// whatever the failed quarantine left, so callers tombstone the partition and
/// leave the bytes for an operator.
pub async fn quarantine_segment_files(partition_dir: &str) ->
std::io::Result<String> {
+ // `create_dir`, not stat-then-create: one syscall per attempt instead of
+ // two, and race-free. Deliberately NOT `create_dir_all`, which succeeds on
+ // an existing directory and would silently merge this fence into an
earlier
+ // copy.
let mut target = None;
for attempt in 0..1000 {
let candidate = format!("{partition_dir}.fenced.{attempt}");
- if compio::fs::metadata(&candidate).await.is_ok() {
- continue;
+ match compio::fs::create_dir(&candidate).await {
+ Ok(()) => {
+ target = Some(candidate);
+ break;
+ }
+ // Lost the race for this suffix; the next iteration probes the
+ // next one.
+ Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists =>
{}
+ Err(error) => return Err(error),
}
- target = Some(candidate);
- break;
}
let Some(target) = target else {
return Err(std::io::Error::other(
"a thousand fenced copies of this partition already exist",
));
};
- compio::fs::create_dir_all(&target).await?;
- // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async directory
- // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
- // `thread_pool_limit(0)`. Bounded by the entry count of ONE partition
directory,
- // but it is a real stall (and under the write lock at the converge site),
so it
- // stays recorded rather than hidden.
- let entries = std::fs::read_dir(partition_dir)?;
- for entry in entries.flatten() {
- let path = entry.path();
+ for path in segment_dir_entries(partition_dir)? {
let quarantined = path.to_str().is_some_and(|path| {
[".log", ".index", STAGING_SUFFIX]
.iter()
@@ -1200,16 +1236,10 @@ pub async fn quarantine_segment_files(partition_dir:
&str) -> std::io::Result<St
/// on the partition -- worst at the reuse scan, which runs at
descriptor-accept
/// on a serving partition.
pub(crate) async fn sweep_staging_except(partition_dir: &str, keep: &[&Path]) {
- // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async directory
- // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
- // `thread_pool_limit(0)`. Bounded by the entry count of ONE partition
directory,
- // but it is a real stall (and under the write lock at the converge site),
so it
- // stays recorded rather than hidden.
- let Ok(entries) = std::fs::read_dir(partition_dir) else {
+ let Ok(entries) = segment_dir_entries(partition_dir) else {
return;
};
- for entry in entries.flatten() {
- let path = entry.path();
+ for path in entries {
let is_staging = path
.to_str()
.is_some_and(|path| path.ends_with(STAGING_SUFFIX));
@@ -1312,6 +1342,11 @@ where
if let Some(cached) = self.transfer_offer_cache.borrow().as_ref()
&& cached.commit_op == commit_op
{
+ // Returns BEFORE the chain re-validation below, deliberately:
+ // re-validating on every hit is the walk the cache exists to skip.
+ // Retention GC on an idle partition therefore costs one wasted
+ // round -- the chunk serve fails `Stale` and the eviction path
+ // re-enumerates -- which is the cheaper side of the trade.
return Ok(Rc::clone(cached));
}
@@ -1450,14 +1485,11 @@ where
.borrow_mut()
.remove(&start_offset);
let mut memo = match memo {
- Some(memo) if memo.hashed_len == size => {
- let checksum = memo.hasher.finish();
- self.segment_checksum_cache
- .borrow_mut()
- .insert(start_offset, memo);
- return Ok(checksum);
- }
- Some(memo) if memo.hashed_len < size => memo,
+ // `<=`, so the already-hashed case falls through to the shared
tail:
+ // `hash_segment_range` returns before opening the file when
+ // `from == to`, and the finish + reinsert below is the same work
the
+ // separate arm did.
+ Some(memo) if memo.hashed_len <= size => memo,
// Segment bytes are append-only within one segment instance (the
// failed-index-save path rewinds the writer cursor and returns
// BEFORE the size increment), and every path that plants a fresh
@@ -1544,11 +1576,7 @@ where
// The append counter, not the segment end: retention can GC every
// sealed segment while the counter stands at N, and the receiver
// must resume minting at N either way.
- let next_offset = if self.should_increment_offset {
- self.offset.load(Ordering::Acquire) + 1
- } else {
- 0
- };
+ let next_offset = self.offset_frontier();
ConsumerOffsetsWire {
purge_generation: self.applied_purge_generation,
next_offset,
@@ -1806,11 +1834,7 @@ where
// legitimate rewind, and the artifact carries the generation that
// proves one happened.
let purge_advances = offsets_wire.purge_generation >
self.applied_purge_generation;
- let local_next_offset = if self.should_increment_offset {
- self.offset.load(Ordering::Acquire) + 1
- } else {
- 0
- };
+ let local_next_offset = self.offset_frontier();
if !purge_advances && local_next_offset > 0 &&
offsets_wire.next_offset < local_next_offset
{
return Err(PartitionInstallError::OfferRewindsDurableData {
@@ -1839,8 +1863,25 @@ where
// staged rename lands, and boot sweeps `.log.staging`, so a crash in
// that window would otherwise leave the frontier named by nothing at
// all and the replica would re-mint from 0 against a group at N.
- self.persist_offset_frontier_at(offsets_wire.next_offset)
- .await;
+ //
+ // REFUSED, not logged and continued: this write is the sole durable
+ // carrier of the frontier on the path that matters, and if the
converge
+ // that follows a failed install also fails, the fence quarantines away
+ // the very segments that would otherwise witness the counter. A
+ // storeless partition returns true early, so refusing here cannot
+ // wedge the in-memory case. Nothing has been mutated yet.
+ //
+ // Under `purge_advances` the offer's frontier is legitimately BELOW
the
+ // live counter; the advancing write maxes it back up, which is correct
+ // here -- the reset belongs to `purge`, which records 0 as it runs.
+ if !self
+ .persist_offset_frontier_at(offsets_wire.next_offset)
+ .await
+ {
+ return Err(PartitionInstallError::FrontierNotDurable {
+ frontier: offsets_wire.next_offset,
+ });
+ }
// The write lock spans the convergence too: a mutate failure leaves
// the segment vectors drained, and a concurrent replicated append
// indexing `segments().len() - 1` on the emptied vec is exactly the
@@ -1897,7 +1938,7 @@ where
// Sweep staging strays a dead earlier attempt left behind, keeping
// only what THIS install is about to rename. Bounded disk hygiene;
// the reuse-scan sweeps too, and boot sweeps ALL of `.staging`
- // (`segment_recovery::sweep_scratch_files`), so a transfer abandoned
+ // (`sweep_scratch_files_and_collect_offsets`), so a transfer abandoned
// for good leaks at most until the next restart.
let keep: Vec<&Path> = staged
.iter()
@@ -1967,17 +2008,15 @@ where
// rename ordering POSIX does not grant.
//
// KNOWN WINDOW, above and here: the old chain's unlinks are already
- // durable and no staged log has landed yet, and nothing durable names
- // the offset frontier in between -- a crash there boots to zero
- // segments and counter 0. Bounded, not silent: the gap check drops
- // live prepares at sequencer 0 and the repair floor refuses a `None`
- // stand-in against a nonzero first batch, so the replica takes a clean
- // full re-transfer instead of serving a hole. One narrow door stays
- // open until the frontier gets a durable home (the partition
- // superblock already reserves a field for it):
- // `repaired_window_is_offsets_only` can accept a complete
- // offsets-only window with the counter still at 0, after which the
- // next live append stamps `base_offset` 0 against the group's N.
+ // durable and no staged log has landed yet. The frontier IS named
+ // durably across it -- the install records it in the superblock before
+ // the first unlink and refuses outright if that write fails -- so a
+ // crash here boots to zero segments with the counter re-seeded from
the
+ // record, and the replica takes a clean full re-transfer. The residual
+ // is narrow: a record that predates this install (a fresh joiner's
+ // view-adoption write leaves frontier 0, which reads as no record at
+ // all), where `repaired_window_is_offsets_only` can then accept a
+ // complete offsets-only window with the counter still at 0.
for meta in &staged {
let (_, index_final) = final_paths(partition_dir,
meta.start_offset);
compio::fs::rename(&meta.index_staging, &index_final)
@@ -2391,15 +2430,9 @@ where
// in-memory vectors were already drained, so only the directory
// itself knows what needs unlinking.
if let Some(partition_dir) = self.partition_dir.clone() {
- // BLOCKING read_dir on the pump: compio-fs 0.12 exposes no async
directory
- // walk, and `spawn_blocking` is not an escape either -- the shard
executors run
- // `thread_pool_limit(0)`. Bounded by the entry count of ONE
partition directory,
- // but it is a real stall (and under the write lock at the
converge site), so it
- // stays recorded rather than hidden.
- let swept: Vec<PathBuf> = match std::fs::read_dir(&partition_dir) {
+ let swept: Vec<PathBuf> = match
segment_dir_entries(&partition_dir) {
Ok(entries) => entries
- .flatten()
- .map(|entry| entry.path())
+ .into_iter()
.filter(|path| {
path.to_str().is_some_and(|path| {
[".log", ".index", STAGING_SUFFIX]
@@ -2461,6 +2494,10 @@ where
(staged_was_empty && minted_next_offset >
0).then_some(minted_next_offset);
self.stats.zero_out_all();
self.stats.increment_segments_count(1);
+ // `zero_out_all` clears the reported offset too, and the counter above
+ // sits at `minted_next_offset - 1`: the success path keeps the two in
+ // step, so this one does as well.
+ self.stats.set_current_offset(end);
self.repair = None;
self.transfer_offer_cache.borrow_mut().take();
Ok(())
@@ -2513,6 +2550,10 @@ const INDEX_STRIDE_BYTES: usize = 64 * 1024;
/// Hand the core back to the reactor mid-CPU-pass.
///
+/// Reactor only: the consensus tick shares this task as a sibling
+/// `select_biased!` arm, and arms are not polled while one arm's body awaits,
so
+/// yielding here does not unfreeze ticks or heartbeats.
+///
/// A zero-duration timer, NOT a bare self-waking yield: this runtime does not
/// reliably re-poll a task that woke itself from inside its own poll, and a
/// pump that suspends that way stops driving consensus entirely (the frame
@@ -2540,7 +2581,7 @@ async fn hash_segment_range(
path: &str,
from: u64,
to: u64,
- hasher: &mut consensus::state_manifest::StateArtifactHasher,
+ hasher: &mut StateArtifactHasher,
mut sink: Option<&mut Vec<u8>>,
) -> std::io::Result<()> {
if from >= to {
@@ -2582,9 +2623,10 @@ async fn hash_segment_range(
/// reactor yield per chunk.
///
/// The serving side runs this on the pump to answer a single chunk request, so
-/// it must not hold the core for a whole-file read plus a non-yielding hash
over
-/// up to 2 GiB -- long enough to miss heartbeat and view-change deadlines on
-/// every group this shard owns.
+/// it reads and hashes in chunks rather than in one pass. The yields keep the
+/// REACTOR moving (detached tasks, `io_uring` completions); they do not keep
this
+/// shard's consensus ticks alive, which are a sibling select arm of the same
+/// task and stay frozen for the duration.
///
/// The file may legitimately be LONGER than the entry (an active segment that
/// kept appending after the offer was built); the artifact is the prefix.
@@ -2597,7 +2639,7 @@ pub async fn load_verified_segment_artifact(
log_path: &str,
entry: &consensus::StateArtifact,
) -> Result<Vec<u8>, SegmentLoadError> {
- let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
+ let mut hasher = StateArtifactHasher::new();
#[allow(clippy::cast_possible_truncation)]
let mut bytes = Vec::with_capacity(entry.len as usize);
hash_segment_range(log_path, 0, entry.len, &mut hasher, Some(&mut bytes))
@@ -2629,10 +2671,15 @@ pub enum SegmentLoadError {
impl SegmentLoadError {
fn classify(source: std::io::Error) -> Self {
- // Only kinds the OS actually named earn the hard verdict:
- // `hash_segment_range` wraps read failures in `Error::other`, which
- // erases the kind, and a short read past EOF is the ordinary racing-GC
- // shape. Everything unclassified is therefore stale (retryable).
+ // `raw_os_error`, not just `kind()`: std maps EIO to
+ // `ErrorKind::Uncategorized`, so a dying disk is invisible to a
+ // kind-only match -- the exact case this split exists to catch.
+ // Everything unrecognised stays STALE: a short read past EOF is what a
+ // racing GC unlink-and-recreate legitimately produces.
+ const EIO: i32 = 5;
+ if source.raw_os_error() == Some(EIO) {
+ return Self::LocalFault(source);
+ }
match source.kind() {
std::io::ErrorKind::PermissionDenied => Self::LocalFault(source),
_ => Self::Stale(source),
@@ -2676,7 +2723,7 @@ async fn verify_state_artifact_yielding(entry:
&consensus::StateArtifact, bytes:
if bytes.len() as u64 != entry.len {
return false;
}
- let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
+ let mut hasher = StateArtifactHasher::new();
for chunk in bytes.chunks(OFFER_HASH_CHUNK_LEN) {
hasher.update(chunk);
yield_to_reactor().await;
@@ -2705,7 +2752,7 @@ pub fn offered_purge_generation(offsets_bytes: &[u8]) ->
u64 {
/// artifact is excluded because the scan never looks at it (and it re-encodes
/// per build, so including it would defeat the memo on every rotation).
fn segment_manifest_digest(manifest: &[consensus::StateArtifact]) -> u64 {
- let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
+ let mut hasher = StateArtifactHasher::new();
for entry in manifest
.iter()
.filter(|entry| entry.kind == artifact_kind::SEGMENT_LOG)
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index fd90b696d..d29991238 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -956,17 +956,23 @@ evicted_ring_capacity = 4096
evicted_ring_bytes_max = "16 MiB"
# Byte budget for segment payloads a SERVING shard keeps resident to answer
-# state-transfer chunk requests, per shard (process-wide is this times the
shard
-# count). Sized for concurrent pulls: at exactly one maximum segment a single
-# rejoining node thrashes the cache by itself, and every miss re-reads and
-# re-hashes a whole segment to serve one 256 KiB chunk. Must be > 0.
-transfer_served_cache_bytes_max = "4 GB"
+# state-transfer chunk requests. PER SHARD, and shard count defaults to core
+# count, so the process-wide high-water is this times the core count on top of
+# page cache -- keep that product in mind before raising it. The default holds
+# two maximum-size segments: below one, a single rejoining node thrashes the
+# cache by itself and every miss re-reads and re-hashes a whole segment to
serve
+# one 256 KiB chunk. Running under the budget costs re-reads, not failures.
+# Must be > 0 and <= "64 GiB".
+transfer_served_cache_bytes_max = "2 GiB"
# Alloc ceiling for ONE received state-transfer artifact, per shard. The
# receiver holds it resident through verify, walk and staging write, and up to
-# four transfers run at once. Keep it above the largest legal segment
-# (segment.size plus one max batch) or legal segments are rejected. Must be >
0.
-transfer_artifact_bytes_max = "1088 MB"
+# four transfers run at once. MUST cover system.segment.size plus
+# message_bus.max_message_size (a segment may close one whole batch past its
+# cap): under that, a legal segment is refused, the whole manifest with it, and
+# the partition livelocks re-requesting it from every peer. Boot validates the
+# floor. Must be > 0 and <= "64 GiB".
+transfer_artifact_bytes_max = "1088 MiB"
# Message bus configuration.
# Tunables for the inter-shard / inter-replica internal bus that ships
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index f85d2cfaa..8e4cc506c 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -133,10 +133,11 @@ pub(crate) type ServerNgMetadata = IggyMetadata<
/// The shard type the dispatch layer is generic over.
///
-/// `B`/`MJ`/`S` are free; the metadata state machine (`M`) and shards table
-/// (`T`) are pinned, being identical in production and the simulator.
-/// Production instantiates it as [`ServerNgShard`]; the simulator supplies its
-/// own `B`/`MJ`/`S`.
+/// `B`/`MJ`/`S`/`SB` are free; the metadata state machine (`M`) and shards
+/// table (`T`) are pinned, being identical in production and the simulator.
+/// Production instantiates it as [`ServerNgShard`], defaulting `SB` to the
+/// on-disk [`PingPongSuperblock`]; the simulator supplies its own
+/// `B`/`MJ`/`S`/`SB`.
pub type ShellShard<B, MJ, S, SB = PingPongSuperblock> =
IggyShard<B, MJ, S, ServerNgMuxStateMachine, PapayaShardsTable, SB>;
@@ -1808,6 +1809,33 @@ async fn build_shard_for_thread(
)
.await?
}
+ // An untrustworthy superblock fences ONE group, not the node. The
+ // segment files stay exactly where they are -- unlike a refused
+ // chain, the data on disk is not the thing in doubt -- so there is
+ // nothing to quarantine and nothing to rebuild: rebuilding fresh
+ // would hand this replica a view-0 identity while a record it
+ // cannot read says otherwise. Tombstoned, the namespace stays
+ // unmaterialised and unrouted, the reconciler backs off, and an
+ // operator has every byte plus a message naming the directory.
+ Err(
+ error @ (ServerNgError::PartitionSuperblockIo { .. }
+ | ServerNgError::PartitionSuperblockVersionUnknown { .. }
+ | ServerNgError::PartitionSuperblockUnverifiable { .. }
+ | ServerNgError::PartitionSuperblockUndecodable { .. }
+ | ServerNgError::PartitionSuperblockIdentityMismatch { .. }),
+ ) => {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ %error,
+ "cannot trust this partition's durable consensus state;
tombstoning the \
+ partition and continuing to boot the rest of the shard"
+ );
+ partition_stats.zero_out_all();
+ partitions.tombstone(namespace);
+ continue;
+ }
Err(error) => return Err(error),
};
partitions.insert(namespace, partition);
@@ -1918,6 +1946,14 @@ const _: () = assert!(
const _: () = assert!(
configs::ng_partition::DEFAULT_EVICTED_RING_BYTES_MAX ==
partitions::EVICTED_RING_BYTES_MAX
);
+const _: () = assert!(
+ configs::ng_partition::DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX
+ == shard::PARTITION_ARTIFACT_LEN_DEFAULT
+);
+const _: () = assert!(
+ configs::ng_partition::DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX
+ == shard::SERVED_SEGMENT_CACHE_BYTES_DEFAULT
+);
const _: () =
assert!(configs::ng_cluster::DEFAULT_REPAIR_CHUNK_MAX as u64 ==
shard::REPAIR_CHUNK_MAX);
const _: () = assert!(
diff --git a/core/server-ng/src/partition_helpers.rs
b/core/server-ng/src/partition_helpers.rs
index 278a5184d..ac0200460 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -340,12 +340,20 @@ pub async fn ensure_initial_segment(
return Ok(());
}
- let messages_path = config
- .system
- .get_messages_file_path(stream_id, topic_id, partition_id, 0);
+ // At the RESTORED FRONTIER, not always 0: after a crash inside the
install's
+ // swap window the chain is empty while the recorded frontier is N, and a
+ // segment named 0 would then take the first append's `base_offset = N` --
+ // `rposition(|s| s.start_offset <= offset)` routes every poll for `0..N-1`
+ // into it, the next boot makes that shape durable, and this replica starts
+ // offering peers a segment that claims `[0..N]`.
+ let start_offset = partition.offset_frontier();
+ let messages_path =
+ config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id,
start_offset);
let index_path = config
.system
- .get_index_path(stream_id, topic_id, partition_id, 0);
+ .get_index_path(stream_id, topic_id, partition_id, start_offset);
let enforce_fsync = config.system.partition.enforce_fsync;
// `file_exists = false` TRUNCATES both files, which is load-bearing here:
a
// fenced-and-rebuilt partition (or one whose quarantine failed) can reach
@@ -385,7 +393,7 @@ pub async fn ensure_initial_segment(
.map(|writer| writer.size_counter())
.unwrap_or_default();
partition.log.add_persisted_segment(
- Segment::new(0, config.system.segment.size),
+ Segment::new(start_offset, config.system.segment.size),
storage,
Some(Rc::new(
MessagesWriter::new(
@@ -439,9 +447,9 @@ pub async fn ensure_initial_segment(
/// Mirrors the metadata plane's recovery contract: an EMPTY superblock is a
/// genuinely fresh group (or one that never changed view) and yields `None`;
/// a present record must decode and match this replica's identity; a present
-/// but unverifiable record refuses boot, because treating it as fresh would
-/// let this replica re-enter a view it already acted in. Quarantining only
-/// the affected partition is future work.
+/// but unverifiable record is an error, because treating it as fresh would
+/// let this replica re-enter a view it already acted in. The boot path
+/// tombstones just that partition rather than refusing the whole node.
///
/// The returned store is the ONE open instance for this group: the partition
/// keeps writing through it, and re-opening later would fork the ping-pong
@@ -450,8 +458,8 @@ pub async fn ensure_initial_segment(
/// # Errors
///
/// [`ServerNgError::PartitionSuperblockIo`] when the directory or a slot
-/// cannot be read; the `Unreadable` / `Undecodable` / `IdentityMismatch`
-/// variants when a record exists but cannot be trusted.
+/// cannot be read; the `VersionUnknown` / `Unverifiable` / `Undecodable` /
+/// `IdentityMismatch` variants when a record exists but cannot be trusted.
pub(crate) async fn open_partition_superblock(
partition_dir: &str,
identity: ReplicaIdentity,
@@ -476,12 +484,19 @@ pub(crate) async fn open_partition_superblock(
}
})?)
}
- SuperblockContents::Unreadable { version } => {
- return Err(ServerNgError::PartitionSuperblockUnreadable {
+ SuperblockContents::Unreadable {
+ version: Some(version),
+ } => {
+ return Err(ServerNgError::PartitionSuperblockVersionUnknown {
dir: PathBuf::from(partition_dir),
version,
});
}
+ SuperblockContents::Unreadable { version: None } => {
+ return Err(ServerNgError::PartitionSuperblockUnverifiable {
+ dir: PathBuf::from(partition_dir),
+ });
+ }
SuperblockContents::Empty => None,
};
if let Some(state) = recovered_state.as_ref() {
@@ -718,7 +733,17 @@ pub async fn build_partition_fresh(
// A "fresh" build is also how a FENCED partition comes back (the shard
// tombstones it and the reconciler rebuilds through here), and the fence
// deliberately leaves the superblock in place, so the recorded frontier is
- // what stops the rebuild from re-minting offsets the group already used.
+ // the rebuild's only anchor.
+ //
+ // It is a LOWER BOUND, not a guarantee: the record is written on view
+ // changes and transfer installs, so it lags the counter arbitrarily -- a
+ // fresh joiner that adopted a view while empty and then filled via repair
+ // has a record still reading 0, and this rebuild would re-seed at 0. For
+ // ordinary crash recovery that staleness is harmless (segments survive and
+ // win the max); it is the fence paths that promote the stale bound to sole
+ // source of truth. Closing it needs the runtime fence to persist the
+ // frontier before quarantining, and the boot-path chain refusal to carry
+ // the refused chain's max `end_offset` on its error.
restore_offset_frontier(&mut partition, recovered_state.as_ref());
let current_offset = partition.offset.load(Ordering::Acquire);
diff --git a/core/server-ng/src/segment_recovery.rs
b/core/server-ng/src/segment_recovery.rs
index edd3d76b1..ef9ba5483 100644
--- a/core/server-ng/src/segment_recovery.rs
+++ b/core/server-ng/src/segment_recovery.rs
@@ -33,7 +33,7 @@ use iggy_common::{IggyByteSize, IggyError, PartitionStats};
use partitions::state_transfer::STAGING_SUFFIX;
use partitions::{IggyIndexReader, Segment};
use server_common::SegmentStorage;
-use server_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Header};
+use server_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Header,
decode_batch_slice};
use std::fs;
use std::os::unix::fs::FileExt;
use std::path::PathBuf;
@@ -70,8 +70,12 @@ pub async fn load_persisted_segments(
let partition_path = config
.system
.get_partition_path(stream_id, topic_id, partition_id);
- sweep_scratch_files(&partition_path);
- let mut start_offsets = collect_segment_start_offsets(&partition_path)?;
+ // ONE directory walk feeds both: the sweep only ever unlinks `.staging`
and
+ // orphan `.index` files, never a `.log`, so the log stems it already
+ // collects ARE the post-sweep start-offset set. Note the error policy is
+ // the collect side's (NotFound => empty, anything else => refuse boot);
the
+ // sweep's silent return would swallow an EACCES that must not be ignored.
+ let mut start_offsets =
sweep_scratch_files_and_collect_offsets(&partition_path)?;
start_offsets.sort_unstable();
let enforce_fsync = config.system.partition.enforce_fsync;
@@ -257,13 +261,28 @@ fn ensure_contiguous_chain(
/// recreates a segment at a given base offset opens its index through
/// `SegmentStorage::new(.., file_exists = false)` first, which TRUNCATES: the
/// stale entries are never read, only overwritten.
-fn sweep_scratch_files(partition_path: &str) {
- let Ok(entries) = fs::read_dir(partition_path) else {
- return;
+/// Sweeps boot-time scratch (`.staging` spill, orphan `.index`) and returns
the
+/// start offset parsed out of every remaining zero-padded `.log` file name. A
+/// missing directory means a never-persisted partition.
+fn sweep_scratch_files_and_collect_offsets(
+ partition_path: &str,
+) -> Result<Vec<u64>, ServerNgError> {
+ let entries = match fs::read_dir(partition_path) {
+ Ok(entries) => entries,
+ Err(source) if source.kind() == std::io::ErrorKind::NotFound => return
Ok(Vec::new()),
+ Err(source) => {
+ error!(
+ partition_path,
+ error = %source,
+ "failed to list partition directory during recovery"
+ );
+ return Err(IggyError::CannotReadPartitions.into());
+ }
};
let mut swept = Vec::new();
let mut orphan_candidates = Vec::new();
let mut log_stems = std::collections::HashSet::new();
+ let mut start_offsets = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
let Some(as_str) = path.to_str() else {
@@ -277,6 +296,9 @@ fn sweep_scratch_files(partition_path: &str) {
Some(LOG_EXTENSION) => {
if let Some(stem) = path.file_stem().and_then(|stem|
stem.to_str()) {
log_stems.insert(stem.to_owned());
+ if let Ok(start_offset) = stem.parse::<u64>() {
+ start_offsets.push(start_offset);
+ }
}
}
Some(INDEX_EXTENSION) => orphan_candidates.push(path),
@@ -299,39 +321,6 @@ fn sweep_scratch_files(partition_path: &str) {
);
}
}
-}
-
-/// Parses the zero-padded start offset out of every `.log` file name in the
-/// partition directory. A missing directory means a never-persisted partition.
-fn collect_segment_start_offsets(partition_path: &str) -> Result<Vec<u64>,
ServerNgError> {
- let entries = match fs::read_dir(partition_path) {
- Ok(entries) => entries,
- Err(source) if source.kind() == std::io::ErrorKind::NotFound => return
Ok(Vec::new()),
- Err(source) => {
- error!(
- partition_path,
- error = %source,
- "failed to list partition directory during recovery"
- );
- return Err(IggyError::CannotReadPartitions.into());
- }
- };
-
- let mut start_offsets = Vec::new();
- for entry in entries.flatten() {
- let path = entry.path();
- if path.extension().and_then(|ext| ext.to_str()) !=
Some(LOG_EXTENSION) {
- continue;
- }
- if let Some(start_offset) = path
- .file_stem()
- .and_then(|stem| stem.to_str())
- .and_then(|stem| stem.parse::<u64>().ok())
- {
- start_offsets.push(start_offset);
- }
- }
-
Ok(start_offsets)
}
@@ -401,12 +390,20 @@ async fn recover_segment_bounds(
// true end offset; a header that no longer decodes marks a torn
// tail, which truncates the readable range to the last whole
// batch so the next append overwrites the torn bytes.
+ // Opened ONCE for the walk: the helper used to open the file per
+ // batch, which is an open + pread + close for every batch in the
+ // segment, synchronously, at boot. A failure to open a file that
+ // just stat'd walks nothing, which lands on the divergence refusal
+ // below rather than recovering an indexed segment as empty.
+ let messages = fs::File::open(messages_path).ok();
let mut position = last.position;
let mut end_offset = last.offset;
let mut end_timestamp = last.timestamp;
let mut walked_any = false;
- while position < messages_size {
- let Some(header) = read_batch_header(messages_path, position,
messages_size) else {
+ while let Some(messages) = messages.as_ref()
+ && position < messages_size
+ {
+ let Some(header) = read_batch_header(messages, position,
messages_size) else {
break;
};
let extent = position.saturating_add(header.total_size() as
u64);
@@ -449,24 +446,45 @@ async fn recover_segment_bounds(
// decode or does not fit, which keeps the torn-tail truncation the
// indexed path performs.
_ if messages_size > 0 => {
+ // Opened once, as above. Nothing walked means no whole batch,
+ // which is the `Ok(None)` the tail of this arm already returns.
+ let messages = fs::File::open(messages_path).ok();
let mut position = 0u64;
let mut start_timestamp = None;
let mut end_offset = start_offset;
let mut end_timestamp = 0;
- while position < messages_size {
- let Some(header) = read_batch_header(messages_path, position,
messages_size) else {
+ let mut expected_offset = start_offset;
+ let mut scratch = Vec::new();
+ while let Some(messages) = messages.as_ref()
+ && position < messages_size
+ {
+ let Some(header) = read_batch_header(messages, position,
messages_size) else {
break;
};
let extent = position.saturating_add(header.total_size() as
u64);
if extent > messages_size {
break;
}
+ // The FILENAME is the only trustworthy anchor once the index
is
+ // gone, and `read_batch_header` checks a length, not a
checksum.
+ // A torn header claiming an offset below `start_offset` would
+ // underflow the message count the caller derives; one
claiming a
+ // jump above becomes this partition's counter, and the next
+ // prepare stamps a `base_offset` diverged from every peer. So
+ // the chain has to be contiguous from the filename onward, and
+ // the batch has to verify before its header is believed.
+ if header.base_offset != expected_offset
+ || !batch_verifies(messages, position, &header, &mut
scratch)
+ {
+ break;
+ }
if header.message_count > 0 {
end_offset = header
.base_offset
.saturating_add(u64::from(header.message_count) - 1);
end_timestamp = header.base_timestamp;
start_timestamp.get_or_insert(header.base_timestamp);
+ expected_offset = end_offset.saturating_add(1);
}
position = extent;
}
@@ -496,16 +514,34 @@ async fn recover_segment_bounds(
/// The batch command header at `position` in the messages file, or `None`
/// when the header does not fit / decode (`position` past the file, header
/// truncated, or garbage bytes).
+/// Whether the batch at `position` decodes and passes its own
`batch_checksum`.
+///
+/// The index-less recovery walk trusts nothing else: without an index the only
+/// anchors are the filename and the payload's self-description, and a torn
+/// header is exactly what that walk exists to survive.
+fn batch_verifies(
+ messages: &fs::File,
+ position: u64,
+ header: &SendMessages2Header,
+ scratch: &mut Vec<u8>,
+) -> bool {
+ scratch.clear();
+ scratch.resize(header.total_size(), 0);
+ if messages.read_exact_at(scratch, position).is_err() {
+ return false;
+ }
+ decode_batch_slice(scratch).is_ok()
+}
+
fn read_batch_header(
- messages_path: &str,
+ messages: &fs::File,
position: u64,
messages_size: u64,
) -> Option<SendMessages2Header> {
if position.checked_add(COMMAND_HEADER_SIZE as u64)? > messages_size {
return None;
}
- let file = fs::File::open(messages_path).ok()?;
let mut header_bytes = [0u8; COMMAND_HEADER_SIZE];
- file.read_exact_at(&mut header_bytes, position).ok()?;
+ messages.read_exact_at(&mut header_bytes, position).ok()?;
SendMessages2Header::decode(&header_bytes).ok()
}
diff --git a/core/server-ng/src/server_error.rs
b/core/server-ng/src/server_error.rs
index 4345e8f83..1aa14d62a 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -110,16 +110,23 @@ pub enum ServerNgError {
#[source]
source: std::io::Error,
},
- // Refuses boot rather than treating the group as fresh or reading through
- // to a superseded view: mirrors the metadata plane's
- // `RecoveryError::SuperblockUnreadable` policy. Quarantining just the one
- // partition is future work; skipping it here could hand the reconciler a
- // reason to re-create it empty.
+ // Quarantines the one partition rather than treating the group as fresh or
+ // reading through to a superseded view: mirrors the metadata plane's
+ // `RecoveryError::SuperblockUnreadable` policy, minus the boot refusal,
+ // because one unreadable partition directory must not strand every healthy
+ // group on the shard.
#[error(
- "partition superblock at {dir} holds no record this build can trust \
- (version {version:?}); refusing boot"
+ "partition superblock at {dir} is present but its format version \
+ {version} is unrecognized by this build (a downgrade, or a corrupt \
+ version field)"
)]
- PartitionSuperblockUnreadable { dir: PathBuf, version: Option<u16> },
+ PartitionSuperblockVersionUnknown { dir: PathBuf, version: u16 },
+ #[error(
+ "partition superblock at {dir} is present but a copy holds bytes that \
+ do not verify (bit-rot or a checksum failure), so its latest \
+ generation cannot be established"
+ )]
+ PartitionSuperblockUnverifiable { dir: PathBuf },
#[error(
"partition superblock at {dir} was checksum-clean but did not decode; \
refusing boot rather than infer a stale view"
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index ab6129528..dd54e8d7c 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -864,10 +864,39 @@ const SEGMENT_SIZE_CEILING_BYTES: u64 = 1 << 30;
/// The most one segment can overshoot its size cap: rotation checks the cap
/// AFTER appending, so a segment closes at most one maximum-size batch past
it.
-/// A batch is bounded by the per-message payload ceiling plus its 256-byte
-/// command header; anything larger is refused at ingest.
-const SEGMENT_SIZE_OVERSHOOT_BYTES: u64 = iggy_common::MAX_PAYLOAD_SIZE as u64
- + server_common::send_messages2::COMMAND_HEADER_SIZE as u64;
+///
+/// Derived from the BUS frame cap, not `MAX_PAYLOAD_SIZE`: server-ng never
+/// enforces the latter (its only enforcement sites are the legacy server and
the
+/// SDK batch types), so the largest appendable batch is whatever the message
bus
+/// will frame. This tracks the shipped `message_bus.max_message_size`
default; an
+/// operator raising that is caught by the config validator, which requires
+/// `partition.transfer_artifact_bytes_max` to cover `system.segment.size` plus
+/// the configured bus cap.
+const SEGMENT_SIZE_OVERSHOOT_BYTES: u64 = 64 * 1024 * 1024;
+
+/// Default alloc ceiling for ONE received state-transfer artifact.
+///
+/// Mirrors `[partition] transfer_artifact_bytes_max`. Free const so the config
+/// crate's copy can be pinned to it by a `const _: () = assert!(..)` at the
+/// server-ng build edge, the way every other runtime default is.
+pub const PARTITION_ARTIFACT_LEN_DEFAULT: u64 =
+ SEGMENT_SIZE_CEILING_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES;
+
+/// Default per-shard resident budget for served segment payloads
+/// (`[partition] transfer_served_cache_bytes_max`). Pinned like
+/// [`PARTITION_ARTIFACT_LEN_DEFAULT`].
+pub const SERVED_SEGMENT_CACHE_BYTES_DEFAULT: u64 =
+ SEGMENT_SIZE_CEILING_BYTES * CONCURRENT_SERVED_SEGMENTS;
+
+/// Distinct max-size segments the served-payload budget holds at once.
+///
+/// TWO, not the receiver's in-flight cap of four: the budget is PER SHARD and
+/// shard count defaults to core count, so each segment here multiplies by the
+/// core count during a whole-node rejoin, on top of page cache and the receive
+/// side's own in-flight artifacts. Two keeps one pull's segment resident
while a
+/// second rotates through; running under the budget costs re-reads, not
+/// failures, and operators serving many concurrent rejoins raise the knob.
+const CONCURRENT_SERVED_SEGMENTS: u64 = 2;
/// Shard-wide cache of segment payloads loaded to serve partition chunks,
/// content-addressed by `(namespace, manifest checksum)` so every requester
@@ -910,31 +939,23 @@ impl ServedSegmentCache {
/// whole segment to serve one 256 KiB chunk. That is the 4096:1 read
/// amplification this cache exists to prevent, plus an offer eviction per
/// failed re-verify feeding the hard-failure backoff.
- const RESIDENT_BYTES_DEFAULT: u64 =
- SEGMENT_SIZE_CEILING_BYTES * Self::CONCURRENT_SERVED_SEGMENTS;
-
- /// Distinct max-size segments the budget holds at once. Matches the
- /// receiver-side in-flight cap, since that is how many distinct segments
one
- /// requester can pull concurrently.
- const CONCURRENT_SERVED_SEGMENTS: u64 = 4;
-
- /// Sweeps a payload survives without serving a chunk.
+ /// Drop every payload that has served nothing for `idle_sweeps_max`
sweeps.
+ ///
+ /// The budget comes from the caller because the two clocks differ: this
+ /// sweep runs on the raw 10 ms consensus tick while the offers these
+ /// payloads back expire on `retry_ticks * MULTIPLE`. Counting bare sweeps
+ /// gave a payload ~100 ms against an offer's ~10 s, so one dropped chunk
+ /// frame -- whose only re-drive is the 1 s stall sweep -- evicted the
+ /// payload and made the resume re-read and re-hash the whole segment to
+ /// serve the next 256 KiB. The trade in the other direction: an abandoned
+ /// pull now pins its resident payload for the full offer window.
///
- /// The sweep runs on the raw consensus tick while the offers it serves
- /// expire on `retry_ticks * MULTIPLE`, so a one-sweep lifetime meant every
- /// chunk gap longer than a tick re-read and re-hashed the whole segment --
- /// 4096 times over for a 1 GiB segment served in 256 KiB chunks. A payload
- /// must outlive at least one stall-retry interval, which this multiple
sets
- /// against the same clock the offer's own expiry uses.
- const IDLE_SWEEPS_MAX: u64 = STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE as u64;
-
- /// Drop every payload that has served nothing for
[`Self::IDLE_SWEEPS_MAX`]
- /// sweeps. Runs from the same place offers expire: without it, one rejoin
- /// leaves a permanent high-water of resident bytes (nothing else releases
- /// the cache once the pulls stop).
- fn expire_idle(&mut self) {
+ /// Runs from the same place offers expire: without it, one rejoin leaves a
+ /// permanent high-water of resident bytes (nothing else releases the cache
+ /// once the pulls stop).
+ fn expire_idle(&mut self, idle_sweeps_max: u64) {
self.sweeps += 1;
- let floor = self.sweeps.saturating_sub(Self::IDLE_SWEEPS_MAX);
+ let floor = self.sweeps.saturating_sub(idle_sweeps_max);
let stale: Vec<(u64, u64)> = self
.entries
.iter()
@@ -1258,12 +1279,12 @@ where
/// Live `[partition] transfer_served_cache_bytes_max`: the byte budget for
/// segment payloads this shard keeps resident to serve chunk requests.
- /// Defaults to [`ServedSegmentCache::RESIDENT_BYTES_DEFAULT`]; server-ng
+ /// Defaults to [`SERVED_SEGMENT_CACHE_BYTES_DEFAULT`]; server-ng
/// overrides it at bootstrap.
served_segment_cache_bytes_max: Cell<u64>,
/// Live `[partition] transfer_artifact_bytes_max`: the alloc ceiling for
one
- /// RECEIVED artifact. Defaults to `PARTITION_ARTIFACT_LEN_DEFAULT`;
+ /// RECEIVED artifact. Defaults to [`PARTITION_ARTIFACT_LEN_DEFAULT`];
/// server-ng overrides it at bootstrap.
partition_artifact_len_max: Cell<u64>,
@@ -1393,8 +1414,8 @@ where
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
- served_segment_cache_bytes_max:
Cell::new(ServedSegmentCache::RESIDENT_BYTES_DEFAULT),
- partition_artifact_len_max:
Cell::new(Self::PARTITION_ARTIFACT_LEN_DEFAULT),
+ served_segment_cache_bytes_max:
Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
+ partition_artifact_len_max:
Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
@@ -1646,8 +1667,8 @@ where
metadata_transfer: RefCell::new(None),
state_transfer_offers: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
- served_segment_cache_bytes_max:
Cell::new(ServedSegmentCache::RESIDENT_BYTES_DEFAULT),
- partition_artifact_len_max:
Cell::new(Self::PARTITION_ARTIFACT_LEN_DEFAULT),
+ served_segment_cache_bytes_max:
Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
+ partition_artifact_len_max:
Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
@@ -5360,16 +5381,15 @@ where
futures::future::join_all(chunk).await;
}
- // Counted ONCE for the whole sweep, then tracked locally as arms
land: a
- // per-namespace count is a full scan per partition, so with
per-partition
- // groups the sweep is O(P^2) exactly when every group is re-arming at
- // once (node-wide view change or rejoin). Capped arms reschedule on
the
- // flat retry interval, so the losers stay phase-locked and repeat that
- // sweep every interval for the whole rejoin -- at a thousand
partitions,
- // past the 10 ms tick budget on the scan alone. A slot freed
mid-sweep is
- // seen on the next tick, which is the same latency a capped arm
already
- // accepts.
- let mut transfers_inflight = self.partition_transfers_inflight();
+ // Counted at most ONCE per sweep and only if a re-arm actually fires,
+ // then tracked locally as arms land. Counting per namespace is a full
+ // scan per partition, so with per-partition groups the sweep would be
+ // O(P^2) exactly when every group is re-arming at once (node-wide view
+ // change or rejoin) -- and counting eagerly every tick pays that scan
on
+ // every quiet tick too, since the re-arm branch is rare. A slot freed
+ // mid-sweep is seen on the next tick, the same latency a capped arm
+ // already accepts.
+ let mut transfers_inflight: Option<usize> = None;
for namespace in namespaces {
let Some(partition) = partitions.get_by_ns(&namespace) else {
@@ -5518,15 +5538,17 @@ where
}
};
if let Some(peer) = rearm_peer {
+ // Counted here, before the `&mut partition` below exists: the
+ // scan takes shared borrows of every partition.
+ let inflight =
+ *transfers_inflight.get_or_insert_with(||
self.partition_transfers_inflight());
let Some(partition) = partitions.get_mut_by_ns(&namespace)
else {
continue;
};
partition.consensus().begin_state_transfer_await();
- let armed = self
- .arm_partition_transfer(partition, peer,
transfers_inflight)
- .await;
+ let armed = self.arm_partition_transfer(partition, peer,
inflight).await;
if armed {
- transfers_inflight += 1;
+ transfers_inflight = Some(inflight + 1);
}
}
}
@@ -5830,10 +5852,11 @@ where
ChunkAttempt::Reply(reply) => break reply,
ChunkAttempt::Load { log_path, entry } => {
// Chunked read + incremental hash: this runs on the pump
to
- // answer ONE 256 KiB chunk request, and a whole-file read
- // plus a non-yielding hash over up to 2 GiB holds the core
- // long enough to miss the heartbeat and view-change
- // deadlines of every group it owns.
+ // answer ONE 256 KiB chunk request, so a whole-file read
+ // plus a single hash pass over up to a segment would be
one
+ // long uninterruptible CPU+IO block. The chunking keeps
the
+ // REACTOR moving; this shard's consensus ticks are a
sibling
+ // select arm of the same task and stay frozen either way.
let loaded =
partitions::state_transfer::load_verified_segment_artifact(
&log_path, &entry,
)
@@ -5919,17 +5942,6 @@ where
}
}
- /// Alloc cap per PARTITION artifact: the configured segment ceiling plus
the
- /// one maximum-size batch a segment may overshoot it by (rotation checks
the
- /// cap after appending). The metadata plane's flat 1 GiB cap would
- /// deterministically reject a legal overshooting segment, and the previous
- /// 2 GiB left the receiver holding twice the largest legal artifact --
- /// `mem::take` moves the buffer out of the session, not out of memory, so
it
- /// stays resident through verify + walk + staging write, times the
in-flight
- /// cap, times the shard count.
- const PARTITION_ARTIFACT_LEN_DEFAULT: u64 =
- SEGMENT_SIZE_CEILING_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES;
-
/// Sanity cap across a partition manifest. Segment artifacts spill to
/// disk as they complete, so this bounds corruption, not memory.
const PARTITION_TRANSFER_TOTAL_LEN_MAX: u64 = 1 << 40;
@@ -5989,14 +6001,28 @@ where
.count()
}
- /// Drop every serving-side artifact this shard holds for `namespace`: the
- /// cached offers and the resident segment payloads behind them.
+ /// Drop every trace of `namespace`'s current bytes from the serving side:
+ /// the partition's own offer cache, this shard's cached offers, and the
+ /// resident payloads behind them.
///
- /// Called where the partition's bytes stop being the bytes the offers
- /// describe (a purge). Neither cache can detect that on its own -- offers
- /// are keyed by `commit_op`, payloads by the manifest checksum over the
old
- /// bytes -- so a puller mid-transfer would keep receiving purged data and
- /// keep both expiry clocks reset while doing it.
+ /// Called wherever a partition's segments stop being the bytes an offer
+ /// describes -- retention cleaning, a committed truncate, a purge. None of
+ /// the caches can detect that themselves: the builder cache is keyed on
+ /// `commit_op` (which a metadata-plane truncate never moves), the shard's
+ /// offers on the requester, and the payloads on a checksum over the bytes
+ /// that just went away -- so a puller mid-transfer keeps receiving deleted
+ /// data and keeps both expiry clocks reset while doing it.
+ pub(crate) fn drop_partition_transfer_state(
+ &self,
+ namespace: IggyNamespace,
+ partition: &IggyPartition<B, SB>,
+ ) where
+ B: MessageBus,
+ {
+ partition.clear_state_transfer_offer_cache();
+ self.drop_served_state_for(namespace.inner());
+ }
+
fn drop_served_state_for(&self, namespace: u64) {
self.state_transfer_offers
.borrow_mut()
@@ -6053,6 +6079,11 @@ where
B: MessageBus + 'static,
T: ShardsTable,
{
+ // BEFORE the quarantine: it moves away the segments that are this
+ // partition's only other witness to the offset frontier, and the
+ // rebuild's sole anchor is then the durable record. Advancing form --
+ // the live counter is what the rebuild must not fall below.
+ partition.persist_offset_frontier().await;
match partition.quarantine_partition_dir().await {
Ok(Some(fenced_dir)) => tracing::error!(
shard = self.id,
@@ -6062,14 +6093,25 @@ where
inspection and never read again"
),
Ok(None) => {}
- Err(error) => tracing::error!(
- shard = self.id,
- namespace_raw = namespace.inner(),
- %error,
- "failed to quarantine the fenced partition's segment files;
the rebuild \
- does NOT re-read them -- `build_partition_fresh` plants
segment 0 with \
- `file_exists = false` and truncates whatever remains"
- ),
+ Err(error) => {
+ // NO rebuild: `build_partition_fresh` plants segment 0 with
+ // `file_exists = false`, truncating whatever the failed
+ // quarantine left, so a rebuild here eats the chain one
segment
+ // per attempt. Tombstone and stop -- the bytes stay for an
+ // operator, and the boot path makes the same call. The
+ // partition stays unreachable until it is dealt with; that is
+ // the intended fence, not a wait.
+ tracing::error!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ %error,
+ "failed to quarantine the fenced partition's segment
files; leaving it \
+ tombstoned rather than rebuilding over them"
+ );
+ self.plane.partitions().tombstone(namespace);
+ self.shards_table.remove(&namespace);
+ return;
+ }
}
self.plane.partitions().tombstone(namespace);
self.shards_table.remove(&namespace);
@@ -6854,7 +6896,13 @@ where
/// counter on every chunk it fetches, so only an abandoned or finished
/// transfer ages out.
fn expire_idle_state_transfer_offers(&self) {
- self.served_segment_cache.borrow_mut().expire_idle();
+ // Same clock the offers below age on: `retry_ticks * MULTIPLE` ticks,
+ // and this sweep runs once per tick.
+ let payload_idle_sweeps =
u64::from(self.repair_retry_ticks.get().max(1))
+ * u64::from(STATE_TRANSFER_OFFER_EXPIRY_MULTIPLE);
+ self.served_segment_cache
+ .borrow_mut()
+ .expire_idle(payload_idle_sweeps);
// `max(1)`: the retry interval is operator-configurable, and a zero
would
// make the expiry zero, dropping every offer on the tick after it was
// built and breaking transfers outright.
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 12c11c3bf..883eefe2d 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -277,6 +277,14 @@ impl ShardMetrics {
self.partition_transfer_refusals_total.inc();
}
+ /// Test-only read, mirroring the siblings; the production scrape goes
+ /// through the prometheus registry.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn partition_transfer_refusals_value(&self) -> u64 {
+ self.partition_transfer_refusals_total.get()
+ }
+
/// Bumped when a parked partition frame is answered instead of served
/// because it was addressed to an incarnation this shard no longer holds
/// (delete + recreate recycled the namespace's slab keys). Serving it
would
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index ea9c8d206..c426e526a 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -624,6 +624,14 @@ where
.clean_expired_segments(now, message_expiry, max_bytes)
.await;
if segments > 0 {
+ // Any unlink invalidates what this shard is SERVING:
+ // the offer names files that are gone and the payload
+ // cache can answer from RAM without touching disk, so
a
+ // puller would install deleted messages. Neither cache
+ // can notice on its own -- one is keyed on the
+ // partition's commit_op, which retention does not
move,
+ // the other on a checksum over the deleted bytes.
+ self.drop_partition_transfer_state(namespace,
partition);
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
@@ -645,6 +653,11 @@ where
let (segments, messages) =
partition.remove_sealed_segments_up_to(up_to_offset).await;
if segments > 0 {
+ // See the cleaner arm: a truncate commits on the
+ // METADATA plane, so this partition's commit_op never
+ // moves and the cached offer stays a hit over unlinked
+ // files.
+ self.drop_partition_transfer_state(namespace,
partition);
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
@@ -677,8 +690,7 @@ where
// disk, so a puller would install purged data.
Both
// are keyed on pre-purge content, so neither can
// notice on its own.
- partition.clear_state_transfer_offer_cache();
- self.drop_served_state_for(namespace.inner());
+ self.drop_partition_transfer_state(namespace,
partition);
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
@@ -700,6 +712,9 @@ where
%error,
"purge-partition failed to reset partition;
fencing it for rebuild"
);
+ // Fenced, but the caches still describe the
+ // pre-purge bytes until the rebuild lands.
+ self.drop_partition_transfer_state(namespace,
partition);
self.fence_partition_for_rebuild(namespace,
partition).await;
}
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 7a8b3ba67..64aecc440 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -368,45 +368,11 @@ impl Simulator {
/// mesh construction caps it at `u16`).
#[allow(clippy::cast_possible_truncation)]
pub fn init_partition(&mut self, namespace: IggyNamespace) {
- for (i, replica) in self.replicas.iter_mut().enumerate() {
+ for (i, replica) in self.replicas.iter().enumerate() {
if self.crashed.contains(&(i as u8)) {
continue;
}
- let shard_count =
u32::try_from(replica.shards.len()).expect("shard count fits u32");
- let owner = calculate_shard_assignment(&namespace, shard_count);
- // One store per group, minted on first materialisation and reused
on
- // every later one, so the recorded view survives a replica
restart.
- let superblock = Rc::clone(
- replica
- .partition_superblocks
- .borrow_mut()
- .entry(namespace)
- .or_default(),
- );
- let recovered_state = superblock
- .read_latest_sync()
- .and_then(|bytes| VsrState::try_from(bytes.as_slice()).ok());
- replica.shards[usize::from(owner)].init_partition(
- namespace,
- Some(superblock),
- recovered_state,
- );
- // Commit the namespace before stamping the rows: a partition the
- // metadata plane never heard of is a shape production cannot
- // produce, and the shard refuses to serve client traffic whose
- // routing-row epoch it cannot match against a committed
- // `created_revision`.
- let streams = replica.shards[0].plane.metadata().mux_stm.streams();
- streams.seed_namespace(namespace, namespace.inner());
- let epoch = streams
- .created_revision_for_namespace(namespace)
- .expect("namespace committed by the seed above");
- for shard in &replica.shards {
- shard.shards_table().insert(
- namespace,
- PartitionLocation::new(ShardId::new(owner), epoch),
- );
- }
+ materialise_partition(replica, namespace);
}
}
@@ -847,6 +813,25 @@ impl Simulator {
pump_tasks,
};
+ // Re-materialise every group this replica had before the crash, as a
+ // rebooted server-ng re-opens every partition directory it owns. This
+ // is what makes the carried-forward superblock load-bearing: the group
+ // recovers the `(view, log_view)` it recorded instead of re-entering
+ // view 0.
+ // SORTED: `HashMap` iteration order is seeded per process, and
+ // materialisation order is observable (shard init order, routing-row
+ // stamps), so replay would stop being byte-identical.
+ let mut materialised: Vec<IggyNamespace> = self.replicas[idx]
+ .partition_superblocks
+ .borrow()
+ .keys()
+ .copied()
+ .collect();
+ materialised.sort_unstable_by_key(IggyNamespace::inner);
+ for namespace in materialised {
+ materialise_partition(&self.replicas[idx], namespace);
+ }
+
// Reconnect to the network and mark the replica live again.
self.network
.process_enable(ProcessId::Replica(replica_index));
@@ -956,6 +941,46 @@ impl Simulator {
}
}
+/// Materialises `namespace` on its hash-owning shard of one replica and stamps
+/// the routing row on every shard of that replica.
+///
+/// Shared by [`SimCluster::init_partition`] and the restart path: a rebooted
+/// server-ng re-opens every partition directory it owns, so the sim has to
+/// re-materialise too, otherwise the superblock a restart carries forward is
+/// never read back and the recovered-view branch is dead code.
+fn materialise_partition(replica: &SimReplica, namespace: IggyNamespace) {
+ let shard_count = u32::try_from(replica.shards.len()).expect("shard count
fits u32");
+ let owner = calculate_shard_assignment(&namespace, shard_count);
+ // One store per group, minted on first materialisation and reused on every
+ // later one, so the recorded view survives a replica restart.
+ let superblock = Rc::clone(
+ replica
+ .partition_superblocks
+ .borrow_mut()
+ .entry(namespace)
+ .or_default(),
+ );
+ let recovered_state = superblock
+ .read_latest_sync()
+ .and_then(|bytes| VsrState::try_from(bytes.as_slice()).ok());
+ replica.shards[usize::from(owner)].init_partition(namespace,
Some(superblock), recovered_state);
+ // Commit the namespace before stamping the rows: a partition the metadata
+ // plane never heard of is a shape production cannot produce, and the shard
+ // refuses to serve client traffic whose routing-row epoch it cannot match
+ // against a committed `created_revision`.
+ let streams = replica.shards[0].plane.metadata().mux_stm.streams();
+ streams.seed_namespace(namespace, namespace.inner());
+ let epoch = streams
+ .created_revision_for_namespace(namespace)
+ .expect("namespace committed by the seed above");
+ for shard in &replica.shards {
+ shard.shards_table().insert(
+ namespace,
+ PartitionLocation::new(ShardId::new(owner), epoch),
+ );
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
