You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/05 15:29:24 UTC
(iggy) branch partition_superblock updated (5d588ff38 -> 9494dd6cc)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a change to branch partition_superblock
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 5d588ff38 Merge branch 'master' into partition_superblock
omit 56b961285 feat(python): expose remaining Topic/TopicDetails fields and partitions (#3623)
new 9494dd6cc address review comments
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (5d588ff38)
\
N -- N -- N refs/heads/partition_superblock (9494dd6cc)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
Cargo.lock | 1 +
bdd/python/uv.lock | 4 +-
core/bench/src/main.rs | 14 +
core/binary_protocol/src/consensus/header.rs | 25 +-
core/common/src/types/partition/mod.rs | 2 +-
core/consensus/src/state_transfer.rs | 10 +
core/integration/src/bench_utils.rs | 62 +-
.../tests/cluster/partition_state_transfer.rs | 93 ++-
.../tests/server/partition_view_durability_vsr.rs | 55 +-
.../server/scenarios/purge_delete_scenario.rs | 83 +--
core/partitions/Cargo.toml | 1 +
core/partitions/src/iggy_partition.rs | 111 ++-
core/partitions/src/journal.rs | 47 +-
core/partitions/src/state_transfer.rs | 759 ++++++++++++++++-----
core/server-ng/src/bootstrap.rs | 75 +-
core/server-ng/src/partition_helpers.rs | 70 +-
core/server-ng/src/segment_recovery.rs | 117 +++-
core/server-ng/src/server_error.rs | 59 ++
core/shard/src/lib.rs | 617 +++++++++++++----
core/shard/src/router.rs | 23 +-
core/simulator/src/lib.rs | 36 +-
examples/python/uv.lock | 2 +-
foreign/python/Cargo.toml | 2 +-
foreign/python/apache_iggy.pyi | 221 +-----
foreign/python/pyproject.toml | 2 +-
foreign/python/src/client.rs | 109 ++-
foreign/python/src/consumer.rs | 32 +-
foreign/python/src/lib.rs | 5 +-
foreign/python/src/topic.rs | 276 +-------
foreign/python/tests/test_topic.py | 235 +------
foreign/python/uv.lock | 4 +-
31 files changed, 1862 insertions(+), 1290 deletions(-)
(iggy) 01/01: address review comments
Posted by gk...@apache.org.
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
commit 9494dd6cc8b713fa30f113c34d17001d22dbc4af
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Wed Aug 5 17:29:14 2026 +0200
address review comments
---
Cargo.lock | 1 +
core/bench/src/main.rs | 14 +
core/binary_protocol/src/consensus/header.rs | 25 +-
core/consensus/src/state_transfer.rs | 10 +
core/integration/src/bench_utils.rs | 62 +-
.../tests/cluster/partition_state_transfer.rs | 93 ++-
.../tests/server/partition_view_durability_vsr.rs | 55 +-
.../server/scenarios/purge_delete_scenario.rs | 83 +--
core/partitions/Cargo.toml | 1 +
core/partitions/src/iggy_partition.rs | 111 ++-
core/partitions/src/journal.rs | 47 +-
core/partitions/src/state_transfer.rs | 759 ++++++++++++++++-----
core/server-ng/src/bootstrap.rs | 75 +-
core/server-ng/src/partition_helpers.rs | 70 +-
core/server-ng/src/segment_recovery.rs | 117 +++-
core/server-ng/src/server_error.rs | 59 ++
core/shard/src/lib.rs | 617 +++++++++++++----
core/shard/src/router.rs | 23 +-
core/simulator/src/lib.rs | 36 +-
19 files changed, 1762 insertions(+), 496 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 3cd7dccbc..afa8bc7a7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -9546,6 +9546,7 @@ dependencies = [
"bytes",
"compio",
"consensus",
+ "futures",
"iggy_binary_protocol",
"iggy_common",
"journal",
diff --git a/core/bench/src/main.rs b/core/bench/src/main.rs
index 6556a5ac6..70e06be02 100644
--- a/core/bench/src/main.rs
+++ b/core/bench/src/main.rs
@@ -33,11 +33,25 @@ use tracing::{error, info};
use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
use utils::cpu_name::append_cpu_name_lowercase;
+/// Which SDK framing this binary was compiled with.
+///
+/// One binary name, two wire dialects, and the mismatch is asymmetric: a
+/// default-features bench against a vsr server HANGS rather than fails, because
+/// the server reads a full 256-byte header before validating anything and the
+/// client's own response timeout is itself vsr-gated. Printing the flavor in the
+/// always-on banner turns that from a silent hang into a one-second diagnosis.
+const SDK_FRAMING: &str = if cfg!(feature = "vsr") {
+ "vsr (server-ng Register handshake)"
+} else {
+ "legacy (classic server framing)"
+};
+
#[tokio::main]
async fn main() -> Result<(), IggyError> {
let standard_font = FIGlet::standard().unwrap();
let figure = standard_font.convert("Iggy Bench");
println!("{}", figure.unwrap());
+ println!("SDK framing: {SDK_FRAMING}");
let mut args = IggyBenchArgs::parse();
args.validate();
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index 52b9fb6ee..6dd003ddb 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1413,9 +1413,25 @@ pub struct StateTransferTargetHeader {
/// Serving primary's applied frontier (`commit_min`) when the descriptor
/// was built. The receiver's tail repair targets past this.
pub commit_op: u64,
+ /// Serving replica's `commit_max` when the descriptor was built.
+ ///
+ /// A receiver refuses an offer from a replica that knows LESS than it does:
+ /// without this the descriptor carried no proof of the sender's own
+ /// progress, and a phantom view-0 primary (a group whose directory vanished
+ /// boots `init()` rather than `init_as_backup()`, comes up Normal at view 0,
+ /// and an empty log is trivially caught up) could hand a data-holding
+ /// rejoiner an empty offer that unlinks its chain.
+ pub commit_max: u64,
pub namespace: u64,
pub available: u8,
- pub reserved: [u8; 95],
+ /// Set on an `available == 0` refusal that means "not right now" rather than
+ /// "this node is broken": the requester then re-arms on a flat interval
+ /// instead of charging its consecutive-failure count, whose exponential
+ /// backoff climbs to 1024x the retry interval and is reset only by a
+ /// completed install. A serving primary momentarily behind its own frontier
+ /// is the common case under produce load.
+ pub unavailable_transient: u8,
+ pub reserved: [u8; 86],
}
const _: () = {
assert!(size_of::<StateTransferTargetHeader>() == HEADER_SIZE);
@@ -1423,7 +1439,7 @@ const _: () = {
offset_of!(StateTransferTargetHeader, nonce)
== offset_of!(StateTransferTargetHeader, reserved_frame) + size_of::<[u8; 66]>()
);
- assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 95]>() == HEADER_SIZE);
+ assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8; 86]>() == HEADER_SIZE);
};
impl ConsensusHeader for StateTransferTargetHeader {
@@ -1450,6 +1466,11 @@ impl ConsensusHeader for StateTransferTargetHeader {
"available must be 0 or 1".to_string(),
));
}
+ if self.unavailable_transient > 1 {
+ return Err(ConsensusError::InvalidField(
+ "unavailable_transient must be 0 or 1".to_string(),
+ ));
+ }
// Unavailable is a bare refusal; a manifest body on it would be
// ambiguous (which offer would the chunks belong to?). An
// `available == 1` body is left unbounded here on purpose: it carries
diff --git a/core/consensus/src/state_transfer.rs b/core/consensus/src/state_transfer.rs
index 12fe6a200..fc6c28645 100644
--- a/core/consensus/src/state_transfer.rs
+++ b/core/consensus/src/state_transfer.rs
@@ -94,6 +94,16 @@ impl ChunkProgress for ArtifactProgress {
}
fn extend_from_chunk(&mut self, payload: &[u8]) {
+ // Reserved on the FIRST chunk of this artifact, not when the manifest is
+ // accepted: reserving every artifact up front is an eager address-space
+ // commit of the whole manifest, and a receiver only ever pulls one
+ // artifact at a time. Exact rather than geometric -- `entry.len` already
+ // passed the caller's per-kind caps, and doubling to gigabyte sizes would
+ // copy roughly twice the bytes at a ~1.5x transient peak.
+ if self.buf.is_empty() {
+ #[allow(clippy::cast_possible_truncation)]
+ self.buf.reserve_exact(self.entry.len as usize);
+ }
self.buf.extend_from_slice(payload);
}
}
diff --git a/core/integration/src/bench_utils.rs b/core/integration/src/bench_utils.rs
index 9519e2165..966d6a25b 100644
--- a/core/integration/src/bench_utils.rs
+++ b/core/integration/src/bench_utils.rs
@@ -109,30 +109,30 @@ pub fn run_bench_and_wait_for_finish(
let mut child = command.spawn().unwrap();
let deadline = Instant::now() + BENCH_WAIT_TIMEOUT;
- let result = loop {
+ // A timeout does NOT panic here: doing so jumped over the capture dump and
+ // the temp-file cleanup below, so every timed-out run leaked both files and
+ // printed only stderr -- while iggy-bench writes its progress to stdout,
+ // the one capture that explains a hang. The verdict is the assert at the end.
+ let mut timed_out = false;
+ let status = loop {
match child.try_wait().unwrap() {
- Some(status) => break status,
+ Some(status) => break Some(status),
None if Instant::now() >= deadline => {
let _ = child.kill();
let _ = child.wait();
- if let Some(stderr_file_path) = &stderr_file_path {
- eprintln!(
- "Iggy bench stderr:\n{}",
- fs::read_to_string(stderr_file_path).unwrap_or_default()
- );
- }
- panic!(
- "iggy-bench did not finish within {BENCH_WAIT_TIMEOUT:?}; if the server \
- runs in vsr mode, make sure iggy-bench was built with --features vsr \
- (the SDK framing is chosen at compile time)"
- );
+ timed_out = true;
+ break None;
}
None => thread::sleep(Duration::from_millis(200)),
}
};
- // Cleanup
- if let Ok(output) = child.wait_with_output() {
+ // Only for a child that exited on its own: a killed-and-reaped one has no
+ // output left to collect, and `wait_with_output` on it would just fail.
+ if !timed_out {
+ let output = child
+ .wait_with_output()
+ .expect("failed to get output from iggy-bench");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
if let Some(stderr_file_path) = &stderr_file_path {
@@ -154,23 +154,17 @@ pub fn run_bench_and_wait_for_finish(
.write_all(stdout.as_bytes())
.unwrap();
}
- } else {
- panic!("Failed to get output from iggy-bench");
}
- if panicking() {
- if let Some(stdout_file_path) = &stdout_file_path {
- eprintln!(
- "Iggy bench stdout:\n{}",
- fs::read_to_string(stdout_file_path).unwrap()
- );
- }
-
- if let Some(stderr_file_path) = &stderr_file_path {
- eprintln!(
- "Iggy bench stderr:\n{}",
- fs::read_to_string(stderr_file_path).unwrap()
- );
+ let failed = timed_out || status.is_none_or(|status| !status.success());
+ if failed || panicking() {
+ for (stream, path) in [("stdout", &stdout_file_path), ("stderr", &stderr_file_path)] {
+ if let Some(path) = path {
+ eprintln!(
+ "Iggy bench {stream}:\n{}",
+ fs::read_to_string(path).unwrap_or_default()
+ );
+ }
}
}
@@ -181,7 +175,13 @@ pub fn run_bench_and_wait_for_finish(
fs::remove_file(stderr_file_path).unwrap();
}
- assert!(result.success());
+ assert!(
+ !timed_out,
+ "iggy-bench did not finish within {BENCH_WAIT_TIMEOUT:?}; if the server \
+ runs in vsr mode, make sure iggy-bench was built with --features vsr \
+ (the SDK framing is chosen at compile time)"
+ );
+ assert!(status.is_some_and(|status| status.success()));
}
pub fn get_random_path() -> String {
diff --git a/core/integration/tests/cluster/partition_state_transfer.rs b/core/integration/tests/cluster/partition_state_transfer.rs
index e27080ad2..55ac09306 100644
--- a/core/integration/tests/cluster/partition_state_transfer.rs
+++ b/core/integration/tests/cluster/partition_state_transfer.rs
@@ -62,7 +62,10 @@ const KILL_GATE_POLL: Duration = Duration::from_millis(10);
const CONVERSION_MARKER: &str = "partition repair floor unreachable; converting to state transfer";
const INSTALL_MARKER: &str = "partition state transfer installed";
-const SERVING_MARKER: &str = "serving partition state transfer";
+/// Logged once per transfer, at the last byte of the last artifact. The
+/// descriptor-time "serving" line only proves a request ARRIVED; this proves the
+/// pull ran.
+const FULLY_SERVED_MARKER: &str = "partition state transfer fully served";
const ABANDON_MARKER: &str =
"partition state transfer stalled past its retry budget; abandoning with a backed-off re-arm";
@@ -197,13 +200,10 @@ async fn given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
await_marker(harness, 2, CONVERSION_MARKER).await;
await_marker(harness, 2, INSTALL_MARKER).await;
// The serving side proves the pull actually ran (not a vacuous install).
- let served_somewhere = [0, 1]
- .iter()
- .any(|&node| harness.node(node).stdout_contains(SERVING_MARKER));
- assert!(
- served_somewhere,
- "some survivor must have served the partition transfer"
- );
+ // Retried like every other marker check here: stdout goes through the
+ // 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;
}
#[iggy_harness(
@@ -215,11 +215,13 @@ async fn given_evicted_ring_when_node_restarts_with_data_should_state_transfer_p
system.segment.size = "8MiB"
)
)]
-// The 8 MiB segment cap makes the 64 MiB bulky seed span ~8 sealed
-// segments, so this spec also exercises the multi-artifact manifest, the
-// per-artifact spill, and the staged-segment reuse scan when the transfer
-// re-arms against the surviving primary -- the other specs are all
-// single-segment under the default 1 GiB cap.
+// The 8 MiB segment cap makes the 64 MiB bulky seed span several sealed
+// segments, so unlike the other specs (single-segment under the default 1 GiB
+// cap) this one installs from a MULTI-ARTIFACT manifest, one spill per artifact.
+// The installed segment count at the end is what asserts that. The re-armed
+// pull also runs the staged-segment reuse scan, but how much it can adopt
+// depends on how many artifacts completed before the kill, so nothing here
+// asserts on reuse.
async fn given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partition(
harness: &mut TestHarness,
) {
@@ -267,14 +269,23 @@ async fn given_transfer_peer_dies_when_stalled_should_abandon_and_recover_partit
// 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;
+
+ // The manifest the install consumed carried one artifact per sealed
+ // segment, and each was spilled and renamed separately: the seeded 64 MiB
+ // over an 8 MiB segment cap cannot land as one file.
+ let data_path = harness.node(2).data_path();
let deadline = Instant::now() + TRANSFER_BUDGET;
loop {
- if harness.node(2).stdout_contains(INSTALL_MARKER) {
+ let installed = segment_log_count(&data_path);
+ if installed > 1 {
return;
}
assert!(
Instant::now() < deadline,
- "node 2 never installed a partition transfer with node 0 dead"
+ "node 2 installed {installed} segment file(s); a 64 MiB transfer under an \
+ 8 MiB segment cap must arrive as a multi-artifact manifest"
);
sleep(MARKER_POLL).await;
}
@@ -384,6 +395,21 @@ async fn poll_count(client: &IggyClient, count: u32) -> Result<u32, IggyError> {
Ok(polled.messages.len() as u32)
}
+/// [`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;
+ while !nodes
+ .iter()
+ .any(|&node| harness.node(node).stdout_contains(marker))
+ {
+ assert!(
+ Instant::now() < deadline,
+ "none of {nodes:?} logged {marker:?} within {TRANSFER_BUDGET:?}"
+ );
+ sleep(MARKER_POLL).await;
+ }
+}
+
async fn await_marker(harness: &TestHarness, node: usize, marker: &str) {
let deadline = Instant::now() + TRANSFER_BUDGET;
while !harness.node(node).stdout_contains(marker) {
@@ -395,13 +421,19 @@ async fn await_marker(harness: &TestHarness, node: usize, marker: &str) {
}
}
-/// Total `.log` bytes outside the metadata directory: transferred segment
-/// payload. Walked rather than path-derived so the test does not hard-code
-/// the `streams/<s>/topics/<t>/partitions/<p>` layout.
+/// Total transferred segment payload under `root`. Walked rather than
+/// path-derived so the test does not hard-code the
+/// `streams/<s>/topics/<t>/partitions/<p>` layout.
+///
+/// Matches the segment file NAME shape, not the `.log` extension alone: the
+/// server's own text log sits under the same data root (~15 KiB on a local run,
+/// about a third of the floor this feeds), and the child inherits `RUST_LOG`, so
+/// an extension-only sum let a debug run satisfy the caller with zero
+/// transferred segment bytes.
fn total_partition_log_bytes(root: &Path) -> u64 {
let mut total = 0;
let _ = walk(root, &mut |path| {
- if path.extension().is_some_and(|extension| extension == "log")
+ if is_segment_log(path)
&& let Ok(metadata) = std::fs::metadata(path)
{
total += metadata.len();
@@ -411,6 +443,29 @@ fn total_partition_log_bytes(root: &Path) -> u64 {
total
}
+/// Number of segment files under `root`; more than one proves the install came
+/// from a multi-artifact manifest, each artifact spilled and renamed in turn.
+fn segment_log_count(root: &Path) -> usize {
+ let mut count = 0;
+ let _ = walk(root, &mut |path| {
+ if is_segment_log(path) {
+ count += 1;
+ }
+ false
+ });
+ count
+}
+
+/// A segment `.log`, named for its 20-digit zero-padded base offset (see
+/// `partitions::state_transfer`'s path builders).
+fn is_segment_log(path: &Path) -> bool {
+ path.extension().is_some_and(|extension| extension == "log")
+ && path
+ .file_stem()
+ .and_then(|stem| stem.to_str())
+ .is_some_and(|stem| stem.len() == 20 && stem.bytes().all(|byte| byte.is_ascii_digit()))
+}
+
fn find_consumer_offset_file(root: &Path) -> Option<PathBuf> {
walk(root, &mut |path| {
path.parent()
diff --git a/core/integration/tests/server/partition_view_durability_vsr.rs b/core/integration/tests/server/partition_view_durability_vsr.rs
index 334465a89..6e6e42a77 100644
--- a/core/integration/tests/server/partition_view_durability_vsr.rs
+++ b/core/integration/tests/server/partition_view_durability_vsr.rs
@@ -52,6 +52,9 @@ const MESSAGES_COUNT: u32 = 10;
/// then the restarted node's probe and repair), and CI runners are slow. 60s bounds
/// the worst case without hanging the suite.
const CONVERGE_TIMEOUT: Duration = Duration::from_secs(60);
+/// Boot line reporting the `(view, log_view)` a partition group restored from its
+/// superblock: the recovered replica's own account of what it read back.
+const RESTORED_VIEW_MARKER: &str = "restored partition view from its superblock";
const POLL_INTERVAL: Duration = Duration::from_millis(250);
#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
@@ -152,20 +155,54 @@ async fn given_advanced_partition_view_when_survivor_restarts_should_recover_vie
.start()
.expect("restart the survivor (node 2)");
- // The reformed cluster must serve the pre-crash batch again, proving the
- // partition group is live under its recovered view.
- wait_until_partition_serves(harness, &[0, 1, 2]).await;
+ // NODE 2 specifically must serve the pre-crash batch: the helper returns on
+ // the first node that answers, and nodes 0 and 1 were not restarted, so
+ // asking the whole set would never contact the node under test.
+ wait_until_partition_serves(harness, &[2]).await;
- // The advanced view survived the survivor's own restart: re-read its partition
- // superblock once resettled and confirm it did not regress toward a fresh 0.
- let view_after = wait_for_advanced_partition_view(harness, 2).await;
+ // 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.
+ let restored = restored_partition_view(harness, 2)
+ .expect("node 2 must log the partition view it restored from its superblock");
assert!(
- view_after.view >= view_before.view && view_after.log_view >= view_before.log_view,
- "the durable partition view must survive node 2's restart without regressing: \
- before={view_before:?}, after={view_after:?}"
+ 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,
+ restored.1
);
}
+/// `(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.
+fn restored_partition_view(harness: &TestHarness, node: usize) -> Option<(u32, u32)> {
+ let log = harness.node(node).stdout_plain();
+ log.lines()
+ .filter(|line| line.contains(RESTORED_VIEW_MARKER))
+ .filter_map(|line| {
+ // Leading space so the `view` key cannot match inside `log_view`.
+ let view = field(line, " view=")?;
+ let log_view = field(line, " log_view=")?;
+ Some((view, log_view))
+ })
+ .max()
+}
+
+/// Value of a space-prefixed `key=<u32>` tracing field.
+fn field(line: &str, key: &str) -> Option<u32> {
+ let start = line.find(key)? + key.len();
+ line[start..]
+ .split(|character: char| !character.is_ascii_digit())
+ .next()
+ .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
diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
index 5abb681cc..77c9c53e2 100644
--- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
+++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
@@ -984,45 +984,50 @@ pub async fn run_purge_topic(harness: &mut TestHarness, restart_server: bool) {
#[cfg(feature = "vsr")]
await_segment_layout(&partition_path, &[0]).await;
- // --- Verify consumer offsets cleared ---
- let consumer_offset = client
- .get_consumer_offset(&consumer, &stream_ident, &topic_ident, Some(PARTITION_ID))
- .await
- .unwrap();
- assert!(
- consumer_offset.is_none(),
- "Consumer offset must be cleared after purge"
- );
-
- let group_offset = client
- .get_consumer_offset(
- &group_consumer_ref,
- &stream_ident,
- &topic_ident,
- Some(PARTITION_ID),
- )
- .await
- .unwrap();
- assert!(
- group_offset.is_none(),
- "Consumer group offset must be cleared after purge"
- );
-
- // --- Verify offset files deleted from disk ---
- let consumer_files: Vec<_> = read_dir(&consumers_dir)
- .map(|e| e.filter_map(|e| e.ok().map(|e| e.file_name())).collect())
- .unwrap_or_default();
- let group_files: Vec<_> = read_dir(&groups_dir)
- .map(|e| e.filter_map(|e| e.ok().map(|e| e.file_name())).collect())
- .unwrap_or_default();
- assert!(
- consumer_files.is_empty(),
- "Consumer offset files must be deleted after purge, found: {consumer_files:?}"
- );
- assert!(
- group_files.is_empty(),
- "Consumer group offset files must be deleted after purge, found: {group_files:?}"
- );
+ // --- Verify consumer offsets cleared (memory + disk) ---
+ // POLLED, not asserted instantly: in the restart cells 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. The purge contract is that the offsets
+ // GO AWAY, not that they are gone in the same frame as a boot-planted
+ // layout, so converge on the cleared state.
+ let offsets_deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+ loop {
+ let consumer_offset = client
+ .get_consumer_offset(&consumer, &stream_ident, &topic_ident, Some(PARTITION_ID))
+ .await
+ .unwrap();
+ let group_offset = client
+ .get_consumer_offset(
+ &group_consumer_ref,
+ &stream_ident,
+ &topic_ident,
+ Some(PARTITION_ID),
+ )
+ .await
+ .unwrap();
+ let consumer_files: Vec<_> = read_dir(&consumers_dir)
+ .map(|e| e.filter_map(|e| e.ok().map(|e| e.file_name())).collect())
+ .unwrap_or_default();
+ let group_files: Vec<_> = read_dir(&groups_dir)
+ .map(|e| e.filter_map(|e| e.ok().map(|e| e.file_name())).collect())
+ .unwrap_or_default();
+ if consumer_offset.is_none()
+ && group_offset.is_none()
+ && consumer_files.is_empty()
+ && group_files.is_empty()
+ {
+ break;
+ }
+ assert!(
+ std::time::Instant::now() < offsets_deadline,
+ "consumer offsets must be cleared after purge: consumer={consumer_offset:?} \
+ group={group_offset:?} consumer_files={consumer_files:?} group_files={group_files:?}"
+ );
+ tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+ }
// --- Verify partition reset: single empty segment at offset 0 ---
assert_fresh_empty_partition(&partition_path);
diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml
index 5524ecf17..0acc918ba 100644
--- a/core/partitions/Cargo.toml
+++ b/core/partitions/Cargo.toml
@@ -43,6 +43,7 @@ bytemuck = { workspace = true }
bytes = { workspace = true }
compio = { workspace = true }
consensus = { workspace = true }
+futures = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
journal = { workspace = true }
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 95fb0ecbc..5761aee4e 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -181,11 +181,13 @@ where
/// refused); tail repair takes over at install. See
/// [`PartitionTransferSession`].
pub transfer: Option<PartitionTransferSession>,
- /// Consecutive transfer stall rounds, across session re-mints. NOT in the
- /// session: three of four metadata arming sites re-minted their session,
- /// so a per-session counter bounded nothing and a permanent failure
- /// cycled abandon -> repair -> refusal -> re-arm at zero forever. Reset
- /// only by [`Self::note_transfer_progress`].
+ /// Consecutive transfer stall rounds WITHIN one recovery attempt. NOT in the
+ /// session: three of four metadata arming sites re-minted their session, so
+ /// a per-session counter bounded nothing and a permanent failure cycled
+ /// abandon -> repair -> refusal -> re-arm at zero forever. Reset by
+ /// [`Self::note_transfer_progress`] and by
+ /// [`Self::note_transfer_rearm_scheduled`]; livelock across attempts is
+ /// bounded by [`Self::transfer_failures`] and its exponential backoff.
transfer_attempts: u32,
/// CONSECUTIVE transfer failures of any class (decode, spill, install,
/// peer-unavailable, stall exhaustion). Deliberately NOT keyed on the
@@ -199,19 +201,24 @@ where
/// consensus ticks elapse. Owned by the shard tick sweep; while one is
/// pending, the repair-refusal trigger must not arm concurrently.
pub transfer_rearm: Option<PendingTransferRearm>,
+ /// Memoized segment-payload checksum state, keyed by segment base offset.
+ /// Sealed segments are immutable, so their stamp never changes; the active
+ /// segment extends its own hasher over the bytes it gained. Without this,
+ /// EVERY offer build re-reads and re-hashes all retained bytes on the pump
+ /// -- and a committing primary advances `commit_op` each round, so the offer
+ /// cache alone never saves the pass. Swept against the live chain at build
+ /// time; cleared wherever segment files are unlinked-and-recreated (purge,
+ /// install, converge).
+ pub(crate) segment_checksum_cache:
+ RefCell<std::collections::HashMap<u64, crate::state_transfer::SegmentChecksumMemo>>,
+ /// Receiving-side memo of the last staged-segment reuse scan, so a peer
+ /// rotation against the same segment set does not re-read and re-walk every
+ /// staged file. See [`crate::state_transfer::ReuseScanMemo`].
+ pub(crate) reuse_scan_memo: RefCell<Option<crate::state_transfer::ReuseScanMemo>>,
/// Serving-side offer cache, keyed by the `commit_op` it was built at, so
/// simultaneous rejoiners share one manifest instead of re-reading every
/// segment per requester. Invalidated by `purge` (same commit frontier,
/// different bytes) and released by the shard's offer-expiry sweep.
- /// Memoized segment-payload checksums keyed by `(start_offset, size)`.
- /// Sealed segments are immutable, so their stamp never changes; the
- /// active segment's growth changes `size` and thus the key. Without
- /// this, EVERY offer build re-reads and re-hashes all retained bytes on
- /// the pump -- and a committing primary advances `commit_op` each round,
- /// so the offer cache alone never saves the pass. Swept against the
- /// live chain at build time; cleared wherever segment files are
- /// unlinked-and-recreated (purge, install, converge).
- pub(crate) segment_checksum_cache: RefCell<std::collections::HashMap<(u64, u64), u64>>,
pub(crate) transfer_offer_cache:
RefCell<Option<Rc<crate::state_transfer::PartitionStateTransferOffer>>>,
}
@@ -356,6 +363,7 @@ where
transfer_failures: 0,
transfer_rearm: None,
segment_checksum_cache: RefCell::new(std::collections::HashMap::new()),
+ reuse_scan_memo: RefCell::new(None),
transfer_offer_cache: RefCell::new(None),
};
if single_replica {
@@ -563,6 +571,19 @@ where
self.transfer_failures = 0;
}
+ /// A fresh re-arm is scheduled: the stall budget starts over for it.
+ ///
+ /// Carrying an exhausted budget into the next attempt left every later
+ /// session with a single retry-interval window to land its first response --
+ /// against a re-arm backoff climbing to 1024x, and a serving side that
+ /// hashes retained bytes before it can answer, so a slow first response is
+ /// ordinary rather than a stall. The budget bounds consecutive stalls within
+ /// one attempt; `transfer_failures` and its backoff are what bound livelock
+ /// across attempts.
+ pub const fn note_transfer_rearm_scheduled(&mut self) {
+ self.transfer_attempts = 0;
+ }
+
/// Read-only view of the stall budget, for diagnostics. The counters
/// themselves are private: they are the anti-livelock argument, and the
/// docs promise exactly one resetter each -- a `pub` field would let any
@@ -3199,6 +3220,15 @@ where
/// (see `rotate_segment`); falls back to the config-derived path for
/// in-memory partitions with no directory.
///
+ /// Both files are opened through `SegmentStorage::new` with
+ /// `file_exists = false`, which TRUNCATES them. That is load-bearing, not
+ /// incidental: this offset may already have an `.index` on disk (a crash
+ /// between the state-transfer install's index-rename and log-rename loops
+ /// leaves final-name indexes with no logs, and the boot sweep only reaches
+ /// the ones still orphaned at startup). The `partitions`-side writers with
+ /// the same names do NOT truncate, so a recreate path that opened them
+ /// directly would read index entries from a previous generation.
+ ///
/// # Errors
/// If the segment's log / index file cannot be created.
pub(crate) async fn install_empty_segment(
@@ -3284,7 +3314,12 @@ where
/// `PurgeTopic` advances the committed generation and triggers a fresh pass).
///
/// # Errors
- /// If the replacement segment's log / index file cannot be created.
+ /// If the replacement segment's log / index file cannot be created. Every
+ /// segment is already drained by then, so an `Err` leaves a partition with
+ /// no serviceable chain: the caller must FENCE it (quarantine + retire for
+ /// the reconciler to rebuild), exactly as the state-transfer install's
+ /// `ConvergeFailed` arm does, or the next append panics on
+ /// `active_segment()`.
pub async fn purge(
&mut self,
config: &PartitionsConfig,
@@ -3331,6 +3366,25 @@ where
}
}
+ // An in-flight state transfer was pulling the PRE-purge state: its
+ // staged segments hold data this purge just deleted, and letting the
+ // session complete renames it back in -- durably, because the install
+ // takes `max(offer generation, applied)` and this purge already stamped
+ // the newer generation, so the reconciler's purge gate never re-fires
+ // and the resurrected data outlives the process. Drop the session,
+ // cancel the scheduled re-arm, release the transfer stage so the
+ // ordinary triggers can arm a fresh one, and sweep the staged bytes.
+ self.transfer = None;
+ self.transfer_rearm = None;
+ let consensus = self.consensus();
+ if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
+ consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
+ }
+ self.reuse_scan_memo.borrow_mut().take();
+ if let Some(partition_dir) = self.partition_dir.clone() {
+ crate::state_transfer::sweep_staging_except(&partition_dir, &[]).await;
+ }
+
// Recreate a fresh empty segment at offset 0 with real writers.
let start_offset = 0u64;
self.install_empty_segment(config, start_offset).await?;
@@ -3508,12 +3562,12 @@ where
// a visible stall beats invisible loss.
let durable_end = self.recovered_durable_offset;
// Recovered bytes and an installed frontier both "stand in"
- // below the floor; a window connecting to either is whole.
- let stand_in = match (durable_end, self.installed_frontier) {
- (Some(durable), Some(frontier)) => Some(durable.saturating_add(1).max(frontier)),
- (Some(durable), None) => Some(durable.saturating_add(1)),
- (None, frontier) => frontier,
- };
+ // below the floor; a window connecting to either is whole. `None`
+ // orders below every `Some`, so the join covers all four
+ // combinations.
+ let stand_in = durable_end
+ .map(|durable| durable.saturating_add(1))
+ .max(self.installed_frontier);
let connected = match (session.first_batch_offset, stand_in) {
(Some(first), Some(bound)) => first <= bound,
(Some(first), None) => first == 0,
@@ -3599,7 +3653,11 @@ where
/// (`floor >= to_op`) counts as complete: there is nothing left that
/// could arrive and change the floor verdict.
fn repaired_window_is_complete(&self, floor: u64, to_op: u64) -> bool {
- ((floor + 1)..=to_op).all(|op| self.log.journal().inner.header_by_op(op).is_some())
+ self.log
+ .journal()
+ .inner
+ .repaired_window_shape(floor, to_op)
+ .complete
}
/// Whether the served repair window `(floor, to_op]` arrived complete and
@@ -3612,13 +3670,8 @@ where
if floor >= to_op {
return false;
}
- ((floor + 1)..=to_op).all(|op| {
- self.log
- .journal()
- .inner
- .header_by_op(op)
- .is_some_and(|header| header.operation != Operation::SendMessages)
- })
+ let shape = self.log.journal().inner.repaired_window_shape(floor, to_op);
+ shape.complete && !shape.holds_messages
}
/// Journal a repaired `SendMessages` prepare, preserving its embedded
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index b737b4a8f..a24dc6b79 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -24,7 +24,7 @@ use server_common::{
use std::io;
use std::{
cell::{Cell, UnsafeCell},
- collections::{BTreeMap, HashMap, VecDeque},
+ collections::{BTreeMap, HashMap, HashSet, VecDeque},
};
use tracing::warn;
@@ -47,6 +47,16 @@ pub struct RetainedBatchMeta {
pub message_count: u32,
}
+/// What one pass over the journal headers found for a repair window.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct RepairedWindowShape {
+ /// Every op in the window is resident. An EMPTY window is complete: nothing
+ /// left can arrive and change the floor verdict.
+ pub complete: bool,
+ /// At least one resident op in the window is a `SendMessages`.
+ pub holds_messages: bool,
+}
+
/// Lookup key for querying messages from the journal.
///
/// `ceiling` is the inclusive commit-frontier bound: the resident journal holds
@@ -655,6 +665,41 @@ where
headers.iter().find(|header| header.op == op).copied()
}
+ /// Presence and message-carrying shape of the repair window `(floor, to_op]`
+ /// in ONE pass over the header vec.
+ ///
+ /// [`Self::header_by_op`] is a linear scan with no index, so asking it
+ /// op-by-op over a window is O(window x headers): on the floor-refusal path
+ /// the replica is gap-stopped, so nothing evicts and the header vec grows
+ /// with the live tail, and the default 4096-op window over ~100k resident
+ /// headers is on the order of 4e8 comparisons -- synchronous, on the shard
+ /// pump, per repair round. Long enough to miss heartbeat and view-change
+ /// deadlines for every group on the core and turn one rejoin into an
+ /// election storm.
+ ///
+ /// The evicted ring is deliberately NOT consulted, matching the op-by-op
+ /// form: consulting it would change the floor-refusal verdict.
+ pub fn repaired_window_shape(&self, floor: u64, to_op: u64) -> RepairedWindowShape {
+ let headers = unsafe { &*self.headers.get() };
+ let expected = to_op.saturating_sub(floor);
+ let mut present: HashSet<u64> = HashSet::with_capacity(headers.len());
+ let mut holds_messages = false;
+ for header in headers
+ .iter()
+ .filter(|header| header.op > floor && header.op <= to_op)
+ {
+ if header.operation == Operation::SendMessages {
+ holds_messages = true;
+ }
+ present.insert(header.op);
+ }
+ RepairedWindowShape {
+ // In-window ops only, deduplicated, so a count match IS coverage.
+ complete: present.len() as u64 == expected,
+ holds_messages,
+ }
+ }
+
/// Headers for the contiguous op run `from_op ..= commit_max`, in op order,
/// stopping at the first missing op. A replication gap must not be skipped:
/// the caller advances `commit_min` strictly by one, so a hole would break
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 9b84dedc4..d11ecd136 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -27,29 +27,42 @@
//! protocol from `core/consensus`; everything in this module is the
//! partition-specific payload handling on either end.
+use crate::messages_writer::MessagesWriter;
+use crate::offset_storage::{delete_persisted_offset, persist_offset};
+use crate::segment::Segment;
+use crate::types::PartitionsConfig;
+use crate::{IggyIndexWriter, IggyPartition};
use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
use consensus::le_cursor::{LeCursor, Truncated, split_verified_trailer};
-use consensus::{ArtifactProgress, state_artifact_checksum};
+use consensus::state_manifest::artifact_kind;
+use consensus::{ArtifactProgress, Sequencer as _, state_artifact_checksum};
+use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize};
+use journal::superblock::SuperblockStore;
+use message_bus::MessageBus;
+use server_common::SegmentStorage;
+use server_common::send_messages2::decode_batch_slice;
use std::fmt;
use std::mem::size_of;
use std::path::{Path, PathBuf};
+use std::rc::Rc;
+use std::sync::atomic::Ordering;
/// Framing marker for the consumer-offsets wire artifact, "ICO1".
-pub const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1";
+pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1";
/// Version byte following the magic.
///
/// Any layout change bumps this, INCLUDING appended fields: the decoder
/// deliberately fails closed on unknown versions and on trailing bytes,
/// because a v2 field can change the meaning of fields v1 already read.
-pub const CONSUMER_OFFSETS_VERSION: u8 = 1;
+pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1;
/// Per-section entry ceiling for the consumer-offsets artifact.
///
/// A corruption guard, not a target: it bounds the allocation `decode`
/// makes from a length field a peer sent, exactly like the manifest's own
/// entry ceiling.
-pub const CONSUMER_OFFSETS_ENTRIES_MAX: u32 = 1 << 20;
+pub(crate) const CONSUMER_OFFSETS_ENTRIES_MAX: u32 = 1 << 20;
/// One in-flight partition state transfer on the receiving replica.
///
@@ -187,6 +200,45 @@ pub struct StagedSegmentMeta {
pub index_staging: PathBuf,
}
+/// Memoized streaming checksum state for one segment file: the hasher fed
+/// exactly `hashed_len` of its bytes, plus the stamp at that length.
+///
+/// Keyed per segment rather than per `(base offset, size)` pair so the ACTIVE
+/// segment extends its own hasher as it grows, instead of missing the memo on
+/// every byte it gained and re-reading from byte zero. Every path that plants a
+/// new file at an existing base offset (purge, install, converge) clears the
+/// whole map, which is what makes "same base offset, longer file, same leading
+/// bytes" hold.
+pub(crate) struct SegmentChecksumMemo {
+ hashed_len: u64,
+ checksum: u64,
+ hasher: consensus::state_manifest::StateArtifactHasher,
+}
+
+impl SegmentChecksumMemo {
+ fn new() -> Self {
+ Self {
+ hashed_len: 0,
+ checksum: 0,
+ hasher: consensus::state_manifest::StateArtifactHasher::new(),
+ }
+ }
+}
+
+/// The result of the last staged-segment reuse scan.
+///
+/// A scan reads every length-matching staged file whole, verifies it, and
+/// re-walks every batch -- sequentially, on the pump. Peer rotation and stall
+/// re-arms mint a fresh session against the same segment set, so without this
+/// the full cost is re-paid per arm. `digest` covers every `SEGMENT_LOG`
+/// manifest entry, so a hit means the new offer expects byte-identical staged
+/// files; any write to a staging file, or any unlink of one, drops the memo, so
+/// a hit can never describe bytes that were replaced meanwhile.
+pub(crate) struct ReuseScanMemo {
+ digest: u64,
+ adopted: Vec<(u32, StagedSegmentMeta)>,
+}
+
/// The consumer-offset artifact: both offset maps plus the applied purge
/// generation, at the offer's `commit_op`.
///
@@ -195,7 +247,7 @@ pub struct StagedSegmentMeta {
/// generation and the reconciler would immediately re-wipe it, costing a
/// full extra transfer.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
-pub struct ConsumerOffsetsWire {
+pub(crate) struct ConsumerOffsetsWire {
pub purge_generation: u64,
/// The origin group's message-offset frontier: the offset the NEXT
/// append will mint, `0` for a partition that never appended. Segments
@@ -315,7 +367,7 @@ impl ConsumerOffsetsWire {
// Ascending-strict doubles as the duplicate reject and makes the
// encoding canonical: one table, one byte sequence.
if previous.is_some_and(|previous| id <= previous) {
- return Err(ConsumerOffsetsWireError::DuplicateConsumerId { section, id });
+ return Err(ConsumerOffsetsWireError::NonAscendingId { section, id });
}
previous = Some(id);
entries.push((id, offset));
@@ -347,7 +399,10 @@ pub enum ConsumerOffsetsWireError {
count: u32,
max: u32,
},
- DuplicateConsumerId {
+ /// Ids in a section are not strictly ascending: a duplicate, or an
+ /// out-of-order entry. Both are the same encoder bug and both break the
+ /// canonical form the encoding promises.
+ NonAscendingId {
section: &'static str,
id: u32,
},
@@ -386,9 +441,10 @@ impl fmt::Display for ConsumerOffsetsWireError {
f,
"consumer-offsets artifact {section} count {count} exceeds the {max} ceiling"
),
- Self::DuplicateConsumerId { section, id } => write!(
+ Self::NonAscendingId { section, id } => write!(
f,
- "consumer-offsets artifact {section} id {id} is duplicated or out of order"
+ "consumer-offsets artifact {section} id {id} does not ascend \
+ (duplicate, or out of order)"
),
}
}
@@ -515,7 +571,7 @@ mod tests {
};
assert_eq!(
ConsumerOffsetsWire::decode(&duplicate.encode()),
- Err(ConsumerOffsetsWireError::DuplicateConsumerId {
+ Err(ConsumerOffsetsWireError::NonAscendingId {
section: "consumers",
id: 5,
})
@@ -528,7 +584,7 @@ mod tests {
};
assert_eq!(
ConsumerOffsetsWire::decode(&unordered.encode()),
- Err(ConsumerOffsetsWireError::DuplicateConsumerId {
+ Err(ConsumerOffsetsWireError::NonAscendingId {
section: "groups",
id: 4,
})
@@ -553,7 +609,7 @@ mod tests {
/// What a full validation walk over one segment payload derived.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct SegmentWalkStats {
+pub(crate) struct SegmentWalkStats {
pub end_offset: u64,
pub start_timestamp: u64,
pub end_timestamp: u64,
@@ -621,12 +677,10 @@ impl std::error::Error for SegmentWalkError {}
/// # Errors
/// [`SegmentWalkError`] on the first invalid byte; nothing is partially
/// trusted.
-pub fn walk_segment_payload(
+pub(crate) fn walk_segment_payload(
base_offset: u64,
bytes: &[u8],
) -> Result<(SegmentWalkStats, Vec<u8>), SegmentWalkError> {
- use server_common::send_messages2::decode_batch_slice;
-
if bytes.is_empty() {
return Err(SegmentWalkError::Empty);
}
@@ -790,6 +844,18 @@ pub enum PartitionTransferUnavailable {
/// In-memory / simulated partition: nothing on disk to serve.
NoPartitionDir,
RepairInProgress,
+ /// Primary-by-index of a group that has committed nothing: an empty group
+ /// is trivially "caught up", so this is the only thing separating a real
+ /// primary from a view-0 phantom whose directory vanished.
+ NothingCommitted,
+ /// More retained segments than the manifest can carry entries for.
+ ManifestTooLarge {
+ entries: usize,
+ max: usize,
+ },
+ /// The segment chain changed while the offer's checksum passes ran, so the
+ /// stamps no longer describe the bytes the offer addresses.
+ SegmentSetChanged,
FlushFailed(iggy_common::IggyError),
SegmentUnreadable {
start_offset: u64,
@@ -797,12 +863,45 @@ pub enum PartitionTransferUnavailable {
},
}
+impl PartitionTransferUnavailable {
+ /// Whether the refusal says "not right now" rather than "this node is
+ /// broken". A requester charges its consecutive-failure count (and the
+ /// exponential re-arm backoff behind it) only for the latter: a primary
+ /// that is momentarily behind its own frontier is the common case under
+ /// produce load, and charging it pins the backoff at its ceiling while
+ /// nothing else recovers the partition.
+ #[must_use]
+ pub const fn transient(&self) -> bool {
+ match self {
+ Self::NotCaughtUpPrimary
+ | Self::RepairInProgress
+ | Self::NothingCommitted
+ | Self::SegmentSetChanged => true,
+ Self::NoPartitionDir
+ | Self::ManifestTooLarge { .. }
+ | Self::FlushFailed(_)
+ | Self::SegmentUnreadable { .. } => false,
+ }
+ }
+}
+
impl fmt::Display for PartitionTransferUnavailable {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::NotCaughtUpPrimary => write!(f, "not the caught-up primary of this group"),
Self::NoPartitionDir => write!(f, "partition has no on-disk directory"),
Self::RepairInProgress => write!(f, "partition is itself mid-repair"),
+ Self::NothingCommitted => write!(
+ f,
+ "primary by index at view 0 with nothing committed; refusing to serve an empty offer"
+ ),
+ Self::ManifestTooLarge { entries, max } => write!(
+ f,
+ "offer needs {entries} manifest entries, past the {max} ceiling"
+ ),
+ Self::SegmentSetChanged => {
+ write!(f, "segment chain changed while the offer was being built")
+ }
Self::FlushFailed(source) => {
write!(f, "flushing the committed prefix failed: {source}")
}
@@ -841,6 +940,13 @@ pub enum PartitionInstallError {
commit_op: u64,
commit_min: u64,
},
+ /// The offer's offset frontier is at or below the offsets this replica
+ /// already holds durably, so installing it would unlink data the offer
+ /// does not carry (and, at frontier 0, fork the offset space).
+ OfferRewindsDurableData {
+ offer_next_offset: u64,
+ durable_end: u64,
+ },
Offsets(ConsumerOffsetsWireError),
/// Duplicate base offset in the staged set.
DuplicateSegment {
@@ -887,6 +993,14 @@ impl fmt::Display for PartitionInstallError {
f,
"transfer frontier {commit_op} is below the local commit frontier {commit_min}"
),
+ Self::OfferRewindsDurableData {
+ offer_next_offset,
+ durable_end,
+ } => write!(
+ f,
+ "offer frontier {offer_next_offset} is at or below the recovered durable end \
+ {durable_end}; installing it would drop offsets this replica already holds"
+ ),
Self::Offsets(source) => write!(f, "consumer-offsets artifact rejected: {source}"),
Self::DuplicateSegment { start_offset } => {
write!(f, "duplicate staged segment at base offset {start_offset}")
@@ -918,30 +1032,101 @@ impl From<ConsumerOffsetsWireError> for PartitionInstallError {
}
}
-use crate::messages_writer::MessagesWriter;
-use crate::offset_storage::{delete_persisted_offset, persist_offset};
-use crate::segment::Segment;
-use crate::types::PartitionsConfig;
-use crate::{IggyIndexWriter, IggyPartition};
-use consensus::Sequencer as _;
-use consensus::state_manifest::artifact_kind;
-use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize};
-use journal::superblock::SuperblockStore;
-use message_bus::MessageBus;
-use server_common::SegmentStorage;
-use std::rc::Rc;
-use std::sync::atomic::Ordering;
+/// 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`).
+pub const STAGING_SUFFIX: &str = ".staging";
-/// Staging-file names inside the partition directory. The `.staging`
-/// extension is provably invisible to boot recovery, which filters on
-/// `extension == "log"`.
+/// Staging-file names inside the partition directory.
fn staging_paths(partition_dir: &str, start_offset: u64) -> (PathBuf, PathBuf) {
(
- PathBuf::from(format!("{partition_dir}/{start_offset:0>20}.log.staging")),
- PathBuf::from(format!("{partition_dir}/{start_offset:0>20}.index.staging")),
+ PathBuf::from(format!(
+ "{partition_dir}/{start_offset:0>20}.log{STAGING_SUFFIX}"
+ )),
+ PathBuf::from(format!(
+ "{partition_dir}/{start_offset:0>20}.index{STAGING_SUFFIX}"
+ )),
)
}
+/// Move every segment file in `partition_dir` aside into `<dir>.fenced.<n>/`,
+/// returning the directory used.
+///
+/// The partition directory itself STAYS, and so do its two superblock slots:
+/// they hold the group's only durable `(view, log_view)`, and moving them would
+/// make the rebuild read an empty directory -- no `restore_partition_view`,
+/// `consensus.init()` instead of `init_as_backup()`, no replica-identity guard --
+/// so the group would re-enter view 0 after acting in view N and could answer a
+/// retransmitted DVC with `(0, 0)`, letting a quorum adopt a log shorter than the
+/// committed prefix.
+///
+/// Nothing reclaims the fenced copies: they are evidence for an operator,
+/// bounded to 1000 per partition by the suffix search, and never read again
+/// (recovery keys on `.log` files inside the partition directory, and the fenced
+/// subdirectory is not one).
+///
+/// # Errors
+/// The underlying `std::io::Error`; the caller logs and lets the rebuild
+/// re-hydrate whatever remains in place.
+pub async fn quarantine_segment_files(partition_dir: &str) -> std::io::Result<String> {
+ 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;
+ }
+ 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?;
+ let entries = std::fs::read_dir(partition_dir)?;
+ for entry in entries.flatten() {
+ let path = entry.path();
+ let quarantined = path.to_str().is_some_and(|path| {
+ [".log", ".index", STAGING_SUFFIX]
+ .iter()
+ .any(|suffix| path.ends_with(suffix))
+ });
+ if !quarantined {
+ continue;
+ }
+ let Some(name) = path.file_name() else {
+ continue;
+ };
+ compio::fs::rename(&path, &PathBuf::from(&target).join(name)).await?;
+ }
+ fsync_dir(partition_dir).await?;
+ Ok(target)
+}
+
+/// Unlink every staging file in `partition_dir` except `keep`.
+///
+/// Best-effort disk hygiene shared by the reuse scan and the install: a file
+/// that survives is swept at the next boot, so a failed unlink is not worth
+/// failing either caller for. The CONVERGE sweep is deliberately not this
+/// function -- it deletes the live chain as well and must propagate its
+/// errors.
+pub(crate) async fn sweep_staging_except(partition_dir: &str, keep: &[&Path]) {
+ let Ok(entries) = std::fs::read_dir(partition_dir) else {
+ return;
+ };
+ for entry in entries.flatten() {
+ let path = entry.path();
+ let is_staging = path
+ .to_str()
+ .is_some_and(|path| path.ends_with(STAGING_SUFFIX));
+ if is_staging && !keep.contains(&path.as_path()) {
+ let _ = compio::fs::remove_file(&path).await;
+ }
+ }
+}
+
fn final_paths(partition_dir: &str, start_offset: u64) -> (String, String) {
(
format!("{partition_dir}/{start_offset:0>20}.log"),
@@ -949,6 +1134,22 @@ fn final_paths(partition_dir: &str, start_offset: u64) -> (String, String) {
)
}
+/// Consumer-offset files written concurrently while installing a transfer.
+/// Each is an open + write + optional fsync, so the width trades reactor queue
+/// depth against how long one partition monopolises it; matches the tick's
+/// superblock pre-pass.
+const OFFSET_PERSIST_CONCURRENCY: usize = 16;
+
+/// One consumer-offset file the install is about to write. Collected before any
+/// write is issued so the offset maps and the persisted-offset tracker are never
+/// borrowed across a batch's await.
+struct PlannedOffsetWrite {
+ kind: ConsumerKind,
+ id: u32,
+ path: String,
+ value: u64,
+}
+
/// fsync the partition directory so a rename made durable stays durable.
/// Async so the wait parks the task instead of the whole shard reactor;
/// every other future on the pump keeps running through it.
@@ -977,6 +1178,7 @@ where
/// # Errors
/// [`PartitionTransferUnavailable`]; the requester falls back to journal
/// repair or retries after the next trigger.
+ #[allow(clippy::too_many_lines)]
pub async fn state_transfer_offer(
&mut self,
config: &PartitionsConfig,
@@ -992,6 +1194,16 @@ where
if self.repair.is_some() {
return Err(PartitionTransferUnavailable::RepairInProgress);
}
+ // Primary-by-index at view 0 over an empty log passes every gate above
+ // yet knows nothing: a group whose directory is absent boots through
+ // `consensus.init()`, comes up Normal at view 0, and an empty group is
+ // trivially "caught up". Its offer would be zero segments at frontier
+ // 0, which makes a receiver holding real data unlink its own chain.
+ // Nothing committed also means nothing worth serving, so refuse and
+ // let the requester rotate to a replica that has state.
+ if self.consensus().commit_max() == 0 && self.installed_frontier.is_none() {
+ return Err(PartitionTransferUnavailable::NothingCommitted);
+ }
self.flush_committed_messages(config)
.await
.map_err(PartitionTransferUnavailable::FlushFailed)?;
@@ -1002,76 +1214,98 @@ where
return Ok(Rc::clone(cached));
}
- // Under the write lock so GC (`remove_sealed_segments_up_to`, also
- // write-locked) cannot unlink a file between enumeration and read.
+ // Enumerate under the write lock so GC (`remove_sealed_segments_up_to`,
+ // also write-locked) cannot unlink a file between enumeration and read.
+ // The checksum passes below run with the lock RELEASED: they are the
+ // expensive part, and the same mutex serializes
+ // `append_send_messages_to_journal` and `commit_messages_inner`, so
+ // holding it across a multi-GiB first pass stalls this partition's
+ // produce and commit for the whole pass. The chain is re-validated
+ // under the lock afterwards.
let write_lock = self.write_lock.clone();
- let _guard = write_lock.lock().await;
- let mut segments = Vec::with_capacity(self.log.segments().len());
- for (segment, storage) in self.log.segments().iter().zip(self.log.storages()) {
- let size = segment.size.as_bytes_u64();
- if size == 0 {
- continue;
- }
- let (log_path, _) = storage.segment_and_index_paths();
- let Some(log_path) = log_path else {
- return Err(PartitionTransferUnavailable::SegmentUnreadable {
- start_offset: segment.start_offset,
- source: std::io::Error::other("segment holds bytes but no backing file"),
- });
- };
- // Checksum pass over exactly `segment.size` bytes: the file can
- // be LONGER after a failed-index-save rewind, and the size
- // counter is what readers bound by. Memoized by
- // `(start_offset, size)` -- sealed segments are immutable and
- // the active one re-keys as it grows -- so repeat builds (every
- // committed round moves `commit_op` and misses the offer cache)
- // hash only new bytes. A miss hashes chunked, one await per
- // chunk, resident footprint one chunk; still on the pump, so
- // the memo is what keeps a large retained history from stalling
- // this core's consensus ticks on every rejoin round.
- let memo_key = (segment.start_offset, size);
- let memoized = self.segment_checksum_cache.borrow().get(&memo_key).copied();
- let checksum = if let Some(checksum) = memoized {
- checksum
- } else {
- {
- let checksum =
- hash_segment_prefix(&log_path, size)
- .await
- .map_err(|source| PartitionTransferUnavailable::SegmentUnreadable {
- start_offset: segment.start_offset,
- source,
- })?;
- self.segment_checksum_cache
- .borrow_mut()
- .insert(memo_key, checksum);
- checksum
+ // Sampled with the chain: a purge inside the hash window below both
+ // truncates every file and restarts the offset space, so a size that
+ // grew back past its planned length would pass the size re-check while
+ // the stamps describe post-purge bytes at pre-purge offsets.
+ let planned_purge_generation = self.applied_purge_generation;
+ let planned: Vec<(u64, u64, String)> = {
+ let _guard = write_lock.lock().await;
+ let mut planned = Vec::with_capacity(self.log.segments().len());
+ for (segment, storage) in self.log.segments().iter().zip(self.log.storages()) {
+ let size = segment.size.as_bytes_u64();
+ if size == 0 {
+ continue;
}
- };
+ let (log_path, _) = storage.segment_and_index_paths();
+ let Some(log_path) = log_path else {
+ return Err(PartitionTransferUnavailable::SegmentUnreadable {
+ start_offset: segment.start_offset,
+ source: std::io::Error::other("segment holds bytes but no backing file"),
+ });
+ };
+ planned.push((segment.start_offset, size, log_path));
+ }
+ planned
+ };
+ // One manifest entry per planned segment plus the offsets table. The
+ // manifest encoder ASSERTS its entry ceiling and that assert survives
+ // release builds, so a partition retaining more segments than the
+ // ceiling would panic this shard the moment a peer asked it to serve.
+ // A small configured `segment_size` makes a chain that long ordinary,
+ // so refuse the request instead of tripping the assert.
+ let manifest_entries = planned.len() + 1;
+ let manifest_entries_max = consensus::state_manifest::STATE_MANIFEST_ENTRIES_MAX as usize;
+ if manifest_entries > manifest_entries_max {
+ return Err(PartitionTransferUnavailable::ManifestTooLarge {
+ entries: manifest_entries,
+ max: manifest_entries_max,
+ });
+ }
+ let mut segments = Vec::with_capacity(planned.len());
+ for (start_offset, size, log_path) in &planned {
+ let checksum = self
+ .segment_checksum(*start_offset, *size, log_path)
+ .await?;
segments.push(SegmentArtifactSource {
entry: consensus::StateArtifact {
kind: artifact_kind::SEGMENT_LOG,
- frontier: segment.start_offset,
- len: size,
+ frontier: *start_offset,
+ len: *size,
checksum,
},
log_path: log_path.clone(),
});
}
- // Sweep memo entries whose segment left the chain (GC) or whose
- // active-size key was superseded, so the map tracks the live chain.
+ // Re-validate the chain under the lock: the passes above yielded, so GC
+ // could have unlinked a sealed segment or a purge could have planted a
+ // fresh file at a planned path. Every stamp would then describe bytes
+ // the offer no longer addresses, so refuse and let the requester ask
+ // again against the chain that exists now.
{
- let live: std::collections::HashSet<(u64, u64)> = self
+ let _guard = write_lock.lock().await;
+ let live: std::collections::HashMap<u64, u64> = self
.log
.segments()
.iter()
.map(|segment| (segment.start_offset, segment.size.as_bytes_u64()))
.collect();
+ // Append-only within a segment instance, so a live size BELOW the
+ // planned one means the file was replaced rather than extended.
+ let changed = planned_purge_generation != self.applied_purge_generation
+ || planned.iter().any(|(start_offset, size, _)| {
+ live.get(start_offset)
+ .is_none_or(|live_size| live_size < size)
+ });
+ if changed {
+ return Err(PartitionTransferUnavailable::SegmentSetChanged);
+ }
+ // Sweep memo entries whose segment left the chain (GC), so the map
+ // tracks the live chain rather than growing with history.
self.segment_checksum_cache
.borrow_mut()
- .retain(|key, _| live.contains(key));
+ .retain(|start_offset, _| live.contains_key(start_offset));
}
let offsets_wire = self.offsets_wire_snapshot();
@@ -1090,30 +1324,97 @@ where
Ok(offer)
}
- /// Move this partition's directory aside as `<dir>.fenced.<n>`, for the
- /// shard's `ConvergeFailed` fence: the converge already failed to sweep
- /// it, so whatever it holds (partial chains, undeletable strays) must
- /// not feed the reconciler's rebuild or resurrect at boot. Picks the
- /// first free suffix rather than a timestamp so repeated fences of the
- /// same group cannot collide.
+ /// The artifact stamp over the first `size` bytes of a segment file,
+ /// extending the memoized hasher rather than re-reading what it already
+ /// covered.
+ ///
+ /// Sealed segments hit the memo outright. The active one pays for its delta
+ /// only, which is what keeps a committing primary off a full re-read of the
+ /// retained history every round: `commit_op` advances per round, so the
+ /// offer cache misses even when nothing else changed.
+ ///
+ /// # Errors
+ /// [`PartitionTransferUnavailable::SegmentUnreadable`] when the file is
+ /// unreadable or shorter than `size`.
+ async fn segment_checksum(
+ &self,
+ start_offset: u64,
+ size: u64,
+ log_path: &str,
+ ) -> Result<u64, PartitionTransferUnavailable> {
+ // Taken OUT of the map for the read: the hash awaits, and a half-fed
+ // hasher left visible could be extended twice by a second build.
+ let memo = self
+ .segment_checksum_cache
+ .borrow_mut()
+ .remove(&start_offset);
+ let mut memo = match memo {
+ Some(memo) if memo.hashed_len == size => {
+ let checksum = memo.checksum;
+ self.segment_checksum_cache
+ .borrow_mut()
+ .insert(start_offset, memo);
+ return Ok(checksum);
+ }
+ 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
+ // file at an existing base offset clears the whole map, so a
+ // shrunk size means the two have drifted.
+ shrunk => {
+ debug_assert!(
+ shrunk.is_none(),
+ "segment {start_offset} shrank to {size} bytes below its memo"
+ );
+ SegmentChecksumMemo::new()
+ }
+ };
+ hash_segment_range(log_path, memo.hashed_len, size, &mut memo.hasher)
+ .await
+ .map_err(|source| PartitionTransferUnavailable::SegmentUnreadable {
+ start_offset,
+ source,
+ })?;
+ memo.hashed_len = size;
+ memo.checksum = memo.hasher.finish();
+ let checksum = memo.checksum;
+ self.segment_checksum_cache
+ .borrow_mut()
+ .insert(start_offset, memo);
+ Ok(checksum)
+ }
+
+ /// Move this partition's SEGMENT files aside into `<dir>.fenced.<n>/`, for
+ /// the shard's `ConvergeFailed` fence: the converge already failed to sweep
+ /// them, so whatever they hold (partial chains, undeletable strays) must not
+ /// feed the reconciler's rebuild or resurrect at boot. Picks the first free
+ /// suffix rather than a timestamp so repeated fences of the same group
+ /// cannot collide, and returns the directory it used so the fence site can
+ /// name it.
+ ///
+ /// The partition directory itself STAYS, and so do its two superblock
+ /// slots: they hold this group's only durable `(view, log_view)`, and moving
+ /// them would make the rebuild read an empty directory -- no
+ /// `restore_partition_view`, `consensus.init()` instead of
+ /// `init_as_backup()`, no replica-identity guard -- so the group would
+ /// re-enter view 0 after acting in view N and could answer a retransmitted
+ /// DVC with `(0, 0)`, letting a quorum adopt a log shorter than the
+ /// committed prefix.
+ ///
+ /// Nothing reclaims the fenced copies: they are evidence for an operator,
+ /// bounded to 1000 per partition by the suffix search and never read again
+ /// (boot recovery keys on `.log` files inside the partition directory, and
+ /// the fenced subdirectory is not one).
///
/// # Errors
- /// The underlying rename's `std::io::Error`; the caller logs and lets
- /// the rebuild re-hydrate the directory in place.
- pub async fn quarantine_partition_dir(&self) -> std::io::Result<()> {
+ /// The underlying `std::io::Error`; the caller logs and lets the rebuild
+ /// re-hydrate whatever remains in place.
+ pub async fn quarantine_partition_dir(&self) -> std::io::Result<Option<String>> {
let Some(dir) = self.partition_dir.clone() else {
- return Ok(());
+ return Ok(None);
};
- for attempt in 0..1000 {
- let target = format!("{dir}.fenced.{attempt}");
- if compio::fs::metadata(&target).await.is_ok() {
- continue;
- }
- return compio::fs::rename(&dir, &target).await;
- }
- Err(std::io::Error::other(
- "a thousand fenced copies of this partition already exist",
- ))
+ quarantine_segment_files(&dir).await.map(Some)
}
/// Release the cached offer once no requester holds one (the shard's
@@ -1189,6 +1490,10 @@ where
let Some(partition_dir) = self.partition_dir.clone() else {
return Err(SpillError::NoPartitionDir);
};
+ // This write replaces whatever the reuse memo recorded for this path,
+ // so the memo cannot outlive it: a later scan against an older offer
+ // must re-read the file rather than trust a walk of the old bytes.
+ self.reuse_scan_memo.borrow_mut().take();
// Artifact-level integrity FIRST, exactly as the metadata plane
// verifies every artifact before decoding: the walk's per-batch
// checksums prove batch bodies, not that these are the bytes the
@@ -1282,6 +1587,11 @@ where
/// immutable, so on a retry or peer re-target typically only the active
/// segment and the offsets artifact re-pull. Staging strays matching no
/// entry are swept.
+ ///
+ /// A scan against a segment set this partition already scanned short-
+ /// circuits through [`ReuseScanMemo`]: rotating to another peer would
+ /// otherwise re-read and re-walk every staged file, up to 2 GiB each,
+ /// sequentially, on the pump.
pub async fn reuse_staged_segments(
&self,
manifest: &[consensus::StateArtifact],
@@ -1289,6 +1599,34 @@ where
let Some(partition_dir) = self.partition_dir.clone() else {
return Vec::new();
};
+ let digest = segment_manifest_digest(manifest);
+ // Cloned out of the borrow: the file re-checks below await.
+ let memoized = self
+ .reuse_scan_memo
+ .borrow()
+ .as_ref()
+ .filter(|memo| memo.digest == digest)
+ .map(|memo| memo.adopted.clone());
+ if let Some(adopted) = memoized {
+ // The memo proves the bytes were validated; only their continued
+ // presence needs re-checking (an install or a converge between the
+ // two scans unlinks them).
+ let mut intact = true;
+ for (_, meta) in &adopted {
+ let log_matches = matches!(
+ compio::fs::metadata(&meta.log_staging).await,
+ Ok(metadata) if metadata.len() == meta.size
+ );
+ if !log_matches || compio::fs::metadata(&meta.index_staging).await.is_err() {
+ intact = false;
+ break;
+ }
+ }
+ if intact {
+ return adopted;
+ }
+ self.reuse_scan_memo.borrow_mut().take();
+ }
let mut adopted = Vec::new();
let mut matched_paths = Vec::new();
for (index, entry) in manifest.iter().enumerate() {
@@ -1317,16 +1655,13 @@ where
adopted.push((index as u32, meta));
}
}
- // Sweep strays: anything `.staging` that no adopted meta claims.
- if let Ok(entries) = std::fs::read_dir(&partition_dir) {
- for dir_entry in entries.flatten() {
- let path = dir_entry.path();
- let is_staging = path.to_str().is_some_and(|path| path.ends_with(".staging"));
- if is_staging && !matched_paths.contains(&path) {
- let _ = compio::fs::remove_file(&path).await;
- }
- }
- }
+ // Sweep strays: anything staged that no adopted meta claims.
+ let keep: Vec<&Path> = matched_paths.iter().map(PathBuf::as_path).collect();
+ sweep_staging_except(&partition_dir, &keep).await;
+ *self.reuse_scan_memo.borrow_mut() = Some(ReuseScanMemo {
+ digest,
+ adopted: adopted.clone(),
+ });
adopted
}
@@ -1367,6 +1702,26 @@ where
});
}
let offsets_wire = ConsumerOffsetsWire::decode(offsets_bytes)?;
+ // Anti-rewind against LOCAL DURABLE BYTES, not just the commit
+ // frontier: the partition journal is memory-only and
+ // `restore_partition_view` restores view/log_view alone, so `commit_min`
+ // is 0 after every restart however much data sits on disk -- the
+ // `StaleTransfer` refusal above is inert on exactly the canonical
+ // rejoin. An offer whose frontier is at or below the recovered durable
+ // end would unlink a chain that already covers those offsets, and at
+ // frontier 0 restart the offset space, forking every future batch stamp
+ // from the rest of the group. A purge is the one legitimate rewind, and
+ // the artifact carries the generation that proves one happened.
+ let purge_advances = offsets_wire.purge_generation > self.applied_purge_generation;
+ if !purge_advances
+ && let Some(durable_end) = self.recovered_durable_offset
+ && offsets_wire.next_offset <= durable_end
+ {
+ return Err(PartitionInstallError::OfferRewindsDurableData {
+ offer_next_offset: offsets_wire.next_offset,
+ durable_end,
+ });
+ }
staged.sort_unstable_by_key(|meta| meta.start_offset);
for pair in staged.windows(2) {
if pair[1].start_offset == pair[0].start_offset {
@@ -1389,6 +1744,10 @@ where
// race every other segment-vec mutator takes this lock against.
let write_lock = self.write_lock.clone();
let _guard = write_lock.lock().await;
+ // Captured before `staged` moves: the convergence may only claim an
+ // offset frontier when the offer itself proved nothing is retained
+ // below it.
+ let staged_was_empty = staged.is_empty();
let outcome = self
.apply_checked_install(config, commit_op, staged, &offsets_wire, &partition_dir)
.await;
@@ -1403,9 +1762,13 @@ where
// that resumed minting at 0 would fork its batch stamps from
// the group. A convergence failure outranks the install error:
// the partition cannot serve and the caller must fence it.
- self.converge_to_empty_after_failed_install(config, offsets_wire.next_offset)
- .await
- .map_err(|source| PartitionInstallError::ConvergeFailed { source })?;
+ self.converge_to_empty_after_failed_install(
+ config,
+ offsets_wire.next_offset,
+ staged_was_empty,
+ )
+ .await
+ .map_err(|source| PartitionInstallError::ConvergeFailed { source })?;
}
outcome
}
@@ -1428,19 +1791,11 @@ where
// the reuse-scan sweeps too, and boot sweeps ALL of `.staging`
// (`segment_recovery::sweep_staging_files`), so a transfer abandoned
// for good leaks at most until the next restart.
- let keep: Vec<&PathBuf> = staged
+ let keep: Vec<&Path> = staged
.iter()
- .flat_map(|meta| [&meta.log_staging, &meta.index_staging])
+ .flat_map(|meta| [meta.log_staging.as_path(), meta.index_staging.as_path()])
.collect();
- if let Ok(entries) = std::fs::read_dir(partition_dir) {
- for entry in entries.flatten() {
- let path = entry.path();
- let is_staging = path.to_str().is_some_and(|path| path.ends_with(".staging"));
- if is_staging && !keep.iter().any(|kept| **kept == path) {
- let _ = compio::fs::remove_file(&path).await;
- }
- }
- }
+ sweep_staging_except(partition_dir, &keep).await;
// The install recreates segment files at paths it unlinks (the staged
// chain can reuse the same base offsets), so an in-flight poll's
@@ -1451,6 +1806,10 @@ where
// re-resolve by path and see the fresh files.
self.log.invalidate_sealed_read_state();
self.segment_checksum_cache.borrow_mut().clear();
+ // Every staging file this install does not rename away is gone by the
+ // time it returns, and the ones it does rename stop being staging
+ // files, so no memo entry can survive it.
+ self.reuse_scan_memo.borrow_mut().take();
// Unlink the old segment chain oldest-first (a crash mid-loop leaves
// the NEWEST suffix, which is contiguous) and drop the in-memory
@@ -1680,6 +2039,13 @@ where
// it ever fires the table was dropped and the flag must say so.
offsets_written = false;
}
+ // Both maps are populated first (no await, so nothing borrows across
+ // one), then the files are written in capped batches. One await per
+ // file put a rejoin carrying thousands of consumers on the pump for
+ // thousands of sequential open + write + optional fsync round trips;
+ // the tick's superblock pre-pass sets the precedent for the width.
+ let mut planned: Vec<PlannedOffsetWrite> =
+ Vec::with_capacity(offsets_wire.consumers.len() + offsets_wire.groups.len());
if let Some(dir) = self.consumer_offsets_path.clone() {
for (id, offset) in &offsets_wire.consumers {
let value = clamp(*offset);
@@ -1687,16 +2053,12 @@ where
entry.offset.store(value, Ordering::Release);
let path = entry.path.clone();
self.consumer_offsets.pin().insert(*id as usize, entry);
- if persist_offset(&path, value, self.consumer_offset_enforce_fsync)
- .await
- .is_ok()
- {
- self.persisted_offsets
- .borrow_mut()
- .insert((ConsumerKind::Consumer, *id), value);
- } else {
- offsets_written = false;
- }
+ planned.push(PlannedOffsetWrite {
+ kind: ConsumerKind::Consumer,
+ id: *id,
+ path,
+ value,
+ });
}
}
if let Some(dir) = self.consumer_group_offsets_path.clone() {
@@ -1707,13 +2069,27 @@ where
entry.offset.store(value, Ordering::Release);
let path = entry.path.clone();
self.consumer_group_offsets.pin().insert(group_id, entry);
- if persist_offset(&path, value, self.consumer_offset_enforce_fsync)
+ planned.push(PlannedOffsetWrite {
+ kind: ConsumerKind::ConsumerGroup,
+ id: *id,
+ path,
+ value,
+ });
+ }
+ }
+ let enforce_fsync = self.consumer_offset_enforce_fsync;
+ for batch in planned.chunks(OFFSET_PERSIST_CONCURRENCY) {
+ let writes = batch.iter().map(|write| async move {
+ let written = persist_offset(&write.path, write.value, enforce_fsync)
.await
- .is_ok()
- {
+ .is_ok();
+ (written, write.kind, write.id, write.value)
+ });
+ for (written, kind, id, value) in futures::future::join_all(writes).await {
+ if written {
self.persisted_offsets
.borrow_mut()
- .insert((ConsumerKind::ConsumerGroup, *id), value);
+ .insert((kind, id), value);
} else {
offsets_written = false;
}
@@ -1819,12 +2195,19 @@ where
&mut self,
config: &PartitionsConfig,
minted_next_offset: u64,
+ staged_was_empty: bool,
) -> Result<(), iggy_common::IggyError> {
while let Some((_, mut storage)) = self.log.retire_front() {
let _ = storage.shutdown();
}
self.log.journal().inner.clear_all();
self.log.journal_mut().info = crate::log::JournalInfo::default();
+ // Every segment and staging file this partition had is about to be
+ // unlinked, so neither memo can describe anything real afterwards. The
+ // checksum map's own doc promises the clear happens here; without it the
+ // promise rested on the caller clearing it first.
+ self.segment_checksum_cache.borrow_mut().clear();
+ self.reuse_scan_memo.borrow_mut().take();
// Sweep EVERY segment file, not the in-memory count's worth: after
// a late failure the renamed-in new chain is on disk while the
@@ -1885,7 +2268,14 @@ where
self.dirty_offset.store(end, Ordering::Relaxed);
self.should_increment_offset = minted_next_offset > 0;
self.recovered_durable_offset = None;
- self.installed_frontier = Some(minted_next_offset);
+ // The frontier claims "everything below me is represented here", and the
+ // repair floor check accepts any floor at or below it. Nothing was
+ // installed, so the claim holds ONLY when the offer itself proved the
+ // origin retains nothing below its frontier -- an empty staged set. With
+ // segments staged and the install failed, this replica holds zero bytes
+ // of a range it would otherwise declare whole, and `set_commit_floor`
+ // would lift `commit_min` over ops it cannot serve.
+ self.installed_frontier = staged_was_empty.then_some(minted_next_offset);
self.stats.zero_out_all();
self.stats.increment_segments_count(1);
self.repair = None;
@@ -1934,24 +2324,31 @@ impl std::error::Error for SpillError {}
/// the reactor many times per segment.
const OFFER_HASH_CHUNK_LEN: usize = 1 << 20;
-/// `XxHash3_64` over the first `len` bytes of `path`, read in
+/// Feed bytes `[from, to)` of `path` into `hasher`, read in
/// [`OFFER_HASH_CHUNK_LEN`] chunks with one reactor yield per chunk. Errors
-/// on a file shorter than `len`: the segment accounts bytes the disk does
+/// on a file shorter than `to`: the segment accounts bytes the disk does
/// not hold.
-async fn hash_segment_prefix(path: &str, len: u64) -> std::io::Result<u64> {
+async fn hash_segment_range(
+ path: &str,
+ from: u64,
+ to: u64,
+ hasher: &mut consensus::state_manifest::StateArtifactHasher,
+) -> std::io::Result<()> {
+ if from >= to {
+ return Ok(());
+ }
let file = compio::fs::File::open(path).await?;
- let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
- let mut position = 0u64;
+ let mut position = from;
// One buffer for the whole pass; `BufResult` hands it back per read
// precisely so the alloc + memset are not paid per chunk. Re-allocated
// (not resized) when the tail chunk is shorter: compio reads into a
- // Vec's CAPACITY, and a shrunken `len` over the old 1 MiB capacity made
+ // Vec's CAPACITY, and a shrunken length over the old 1 MiB capacity made
// `read_exact_at` demand a full megabyte at EOF.
#[allow(clippy::cast_possible_truncation)]
- let mut buf = vec![0u8; OFFER_HASH_CHUNK_LEN.min(len as usize)];
- while position < len {
+ let mut buf = vec![0u8; OFFER_HASH_CHUNK_LEN.min((to - from) as usize)];
+ while position < to {
#[allow(clippy::cast_possible_truncation)]
- let want = OFFER_HASH_CHUNK_LEN.min((len - position) as usize);
+ let want = OFFER_HASH_CHUNK_LEN.min((to - position) as usize);
if buf.len() != want {
buf = vec![0u8; want];
}
@@ -1959,13 +2356,69 @@ async fn hash_segment_prefix(path: &str, len: u64) -> std::io::Result<u64> {
buf = returned;
read.map_err(|source| {
std::io::Error::other(format!(
- "reading segment bytes at {position} of {len} failed: {source}"
+ "reading segment bytes at {position} of {to} failed: {source}"
))
})?;
hasher.update(&buf);
position += want as u64;
}
- Ok(hasher.finish())
+ Ok(())
+}
+
+/// Read the first `entry.len` bytes of a served segment file and re-verify them
+/// against the manifest entry, in [`OFFER_HASH_CHUNK_LEN`] chunks with one
+/// 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. `None` means the file is shorter than the entry
+/// or no longer hashes to it (GC unlinked and recreated it, or the bytes rotted):
+/// the caller evicts the offer and has the requester restart, which is the only
+/// way the load-time re-verification this exists for can be acted on.
+///
+/// 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.
+pub async fn load_verified_segment_artifact(
+ log_path: &str,
+ entry: &consensus::StateArtifact,
+) -> Option<Vec<u8>> {
+ let file = compio::fs::File::open(log_path).await.ok()?;
+ if file.metadata().await.ok()?.len() < entry.len {
+ return None;
+ }
+ let mut hasher = consensus::state_manifest::StateArtifactHasher::new();
+ #[allow(clippy::cast_possible_truncation)]
+ let mut bytes = Vec::with_capacity(entry.len as usize);
+ let mut position = 0u64;
+ while position < entry.len {
+ #[allow(clippy::cast_possible_truncation)]
+ let want = OFFER_HASH_CHUNK_LEN.min((entry.len - position) as usize);
+ let compio::BufResult(read, chunk) = file.read_exact_at(vec![0u8; want], position).await;
+ read.ok()?;
+ hasher.update(&chunk);
+ bytes.extend_from_slice(&chunk);
+ position += want as u64;
+ }
+ (hasher.finish() == entry.checksum).then_some(bytes)
+}
+
+/// Stamp over every `SEGMENT_LOG` entry of a manifest, keying
+/// [`ReuseScanMemo`]. Equal digests mean the two offers expect byte-identical
+/// staged files, so a scan already done for one answers the other; the offsets
+/// 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();
+ for entry in manifest
+ .iter()
+ .filter(|entry| entry.kind == artifact_kind::SEGMENT_LOG)
+ {
+ hasher.update(&entry.frontier.to_le_bytes());
+ hasher.update(&entry.len.to_le_bytes());
+ hasher.update(&entry.checksum.to_le_bytes());
+ }
+ hasher.finish()
}
async fn write_staging_file(path: &Path, payload: Vec<u8>) -> std::io::Result<()> {
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 23ecd9108..563111417 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -25,8 +25,8 @@ use crate::dispatch::{
};
use crate::http;
use crate::partition_helpers::{
- configure_consumer_offsets, ensure_initial_segment, open_partition_superblock,
- restore_partition_view, validate_namespace_bounds,
+ build_partition_fresh, configure_consumer_offsets, ensure_initial_segment,
+ open_partition_superblock, restore_partition_view, validate_namespace_bounds,
};
use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerNgError, ShardJoinFailure, ShardJoinFailureKind};
@@ -1721,17 +1721,70 @@ async fn build_shard_for_thread(
for (stream_id, topic_id, partition_stats, partition_metadata) in owned {
validate_namespace_bounds(config, stream_id, topic_id, partition_metadata.id)?;
let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
- let partition = load_partition(
+ let partition = match load_partition(
config,
namespace,
- partition_stats,
+ Arc::clone(&partition_stats),
&partition_metadata,
topology.cluster_id,
topology.self_replica_id,
topology.replica_count,
Rc::clone(&bus),
)
- .await?;
+ .await
+ {
+ Ok(partition) => partition,
+ // ONE damaged local chain must not take the node down. The shapes
+ // this refuses are exactly what a failed state-transfer quarantine
+ // leaves behind, so fence that group the same way the runtime path
+ // does -- move its segment files aside, keeping the superblock so it
+ // cannot re-enter view 0 -- and materialise it fresh. The ordinary
+ // rejoin path (repair, then state transfer on a refused floor)
+ // recovers its data from a peer.
+ Err(ServerNgError::PartitionChainRefused { dir, reason, .. }) => {
+ let partition_dir = dir.to_string_lossy().into_owned();
+ error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ partition_dir,
+ %reason,
+ "refusing the recovered segment chain; fencing this partition and \
+ rebuilding it empty for the rejoin path"
+ );
+ match partitions::state_transfer::quarantine_segment_files(&partition_dir).await {
+ Ok(fenced_dir) => error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ fenced_dir,
+ "quarantined the refused segment files; they are kept for inspection"
+ ),
+ Err(error) => error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ partition_dir,
+ %error,
+ "failed to quarantine the refused segment files; the rebuild will \
+ re-read them and refuse again"
+ ),
+ }
+ // The refused load already folded its segment counts in.
+ partition_stats.zero_out_all();
+ build_partition_fresh(
+ config,
+ namespace,
+ partition_stats,
+ topology.cluster_id,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ )
+ .await?
+ }
+ Err(error) => return Err(error),
+ };
partitions.insert(namespace, partition);
shards_table.insert(
namespace,
@@ -2175,9 +2228,15 @@ async fn load_partition(
let partition_dir = config
.system
.get_partition_path(stream_id, topic_id, partition_id);
- let (superblock, recovered_state) =
- open_partition_superblock(&partition_dir, cluster_id, self_replica_id, replica_count)
- .await?;
+ let (superblock, recovered_state) = open_partition_superblock(
+ &partition_dir,
+ ReplicaIdentity {
+ cluster: cluster_id,
+ replica_id: self_replica_id,
+ replica_count,
+ },
+ )
+ .await?;
if let Some(state) = recovered_state.as_ref() {
restore_partition_view(&mut consensus, state);
}
diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs
index 273a0c055..d722d637e 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -34,7 +34,7 @@ use iggy_common::{
};
use journal::superblock::{PingPongSuperblock, SuperblockContents};
use message_bus::IggyMessageBus;
-use metadata::impls::recovery::IdentityField;
+use metadata::impls::recovery::{IdentityField, ReplicaIdentity};
use partitions::{IggyIndexWriter, IggyPartition, MessagesWriter, Segment};
use server_common::SegmentStorage;
use server_common::fs_utils::remove_dir_all;
@@ -43,7 +43,7 @@ use std::path::{Path, PathBuf};
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::Ordering;
-use tracing::{error, warn};
+use tracing::{error, info, warn};
/// Validate that a namespace fits within the static caps declared in
/// `config.extra.namespace`.
@@ -347,6 +347,11 @@ pub async fn ensure_initial_segment(
.system
.get_index_path(stream_id, topic_id, partition_id, 0);
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
+ // this with a stale `.index` at offset 0 on disk. The `partitions`-side
+ // writers with the same names do NOT truncate, so opening them directly
+ // instead would read index entries from a previous generation.
let storage = SegmentStorage::new(
&messages_path,
&index_path,
@@ -449,9 +454,7 @@ pub async fn ensure_initial_segment(
/// variants when a record exists but cannot be trusted.
pub(crate) async fn open_partition_superblock(
partition_dir: &str,
- cluster_id: u128,
- self_replica_id: u8,
- replica_count: u8,
+ identity: ReplicaIdentity,
) -> Result<(Rc<PingPongSuperblock>, Option<VsrState>), ServerNgError> {
let io_error = |source| ServerNgError::PartitionSuperblockIo {
dir: PathBuf::from(partition_dir),
@@ -490,20 +493,20 @@ pub(crate) async fn open_partition_superblock(
found,
})
};
- if state.cluster != cluster_id {
- return mismatch(IdentityField::Cluster, cluster_id, state.cluster);
+ if state.cluster != identity.cluster {
+ return mismatch(IdentityField::Cluster, identity.cluster, state.cluster);
}
- if state.replica_id != self_replica_id {
+ if state.replica_id != identity.replica_id {
return mismatch(
IdentityField::ReplicaId,
- self_replica_id.into(),
+ identity.replica_id.into(),
state.replica_id.into(),
);
}
- if state.replica_count != replica_count {
+ if state.replica_count != identity.replica_count {
return mismatch(
IdentityField::ReplicaCount,
- replica_count.into(),
+ identity.replica_count.into(),
state.replica_count.into(),
);
}
@@ -519,6 +522,16 @@ pub(crate) fn restore_partition_view(
consensus: &mut VsrConsensus<Rc<IggyMessageBus>>,
state: &VsrState,
) {
+ // The one line proving the durable record was READ BACK, not merely written:
+ // the group's whole anti-regression guarantee rests on this call running, and
+ // a replica that came back at view 0 is otherwise indistinguishable from one
+ // that resumed correctly until it votes.
+ info!(
+ namespace_raw = consensus.namespace(),
+ view = state.view,
+ log_view = state.log_view,
+ "restored partition view from its superblock"
+ );
consensus.set_view(state.view);
consensus.set_log_view(state.log_view);
consensus.mark_superblock_durable(state.view, state.log_view);
@@ -612,9 +625,15 @@ pub async fn build_partition_fresh(
let partition_dir = config
.system
.get_partition_path(stream_id, topic_id, partition_id);
- let (superblock, recovered_state) =
- open_partition_superblock(&partition_dir, cluster_id, self_replica_id, replica_count)
- .await?;
+ let (superblock, recovered_state) = open_partition_superblock(
+ &partition_dir,
+ ReplicaIdentity {
+ cluster: cluster_id,
+ replica_id: self_replica_id,
+ replica_count,
+ },
+ )
+ .await?;
if let Some(state) = recovered_state.as_ref() {
restore_partition_view(&mut consensus, state);
}
@@ -744,16 +763,23 @@ mod tests {
root.path().join("partition").to_string_lossy().into_owned()
}
+ const fn test_identity() -> ReplicaIdentity {
+ ReplicaIdentity {
+ cluster: CLUSTER,
+ replica_id: REPLICA,
+ replica_count: REPLICAS,
+ }
+ }
+
#[compio::test]
async fn given_fresh_partition_dir_when_superblock_opened_should_yield_no_state() {
let root = tempfile::tempdir().expect("tempdir");
// A not-yet-materialized directory must open as fresh, not error: the
// helper creates it, since a follower can reach load before its first
// segment write.
- let (_store, recovered) =
- open_partition_superblock(&partition_dir(&root), CLUSTER, REPLICA, REPLICAS)
- .await
- .expect("open a fresh partition superblock");
+ let (_store, recovered) = open_partition_superblock(&partition_dir(&root), test_identity())
+ .await
+ .expect("open a fresh partition superblock");
assert!(
recovered.is_none(),
"an empty superblock is a fresh group, never an error"
@@ -764,7 +790,7 @@ mod tests {
async fn given_recorded_view_when_superblock_reopened_should_recover_state() {
let root = tempfile::tempdir().expect("tempdir");
let dir = partition_dir(&root);
- let (store, recovered) = open_partition_superblock(&dir, CLUSTER, REPLICA, REPLICAS)
+ let (store, recovered) = open_partition_superblock(&dir, test_identity())
.await
.expect("first open");
assert!(recovered.is_none());
@@ -775,7 +801,7 @@ mod tests {
.expect("record the advanced view");
drop(store);
- let (_store, recovered) = open_partition_superblock(&dir, CLUSTER, REPLICA, REPLICAS)
+ let (_store, recovered) = open_partition_superblock(&dir, test_identity())
.await
.expect("reopen after a restart");
@@ -790,7 +816,7 @@ mod tests {
async fn given_foreign_cluster_record_when_superblock_opened_should_refuse_boot() {
let root = tempfile::tempdir().expect("tempdir");
let dir = partition_dir(&root);
- let (store, _) = open_partition_superblock(&dir, CLUSTER, REPLICA, REPLICAS)
+ let (store, _) = open_partition_superblock(&dir, test_identity())
.await
.expect("first open");
let foreign = VsrState {
@@ -803,7 +829,7 @@ mod tests {
.expect("record a foreign identity");
drop(store);
- let refused = open_partition_superblock(&dir, CLUSTER, REPLICA, REPLICAS).await;
+ let refused = open_partition_superblock(&dir, test_identity()).await;
match refused {
Err(ServerNgError::PartitionSuperblockIdentityMismatch { field, .. }) => {
diff --git a/core/server-ng/src/segment_recovery.rs b/core/server-ng/src/segment_recovery.rs
index 5a8f996b7..297b92afa 100644
--- a/core/server-ng/src/segment_recovery.rs
+++ b/core/server-ng/src/segment_recovery.rs
@@ -27,17 +27,20 @@
//! recovery panic). This module is the server-ng-owned loader, reading the same
//! 24-byte format its writer emits.
-use crate::server_error::ServerNgError;
+use crate::server_error::{PartitionChainRefusal, ServerNgError};
use configs::server_ng::ServerNgConfig;
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 std::fs;
use std::os::unix::fs::FileExt;
+use std::path::PathBuf;
use tracing::{error, warn};
const LOG_EXTENSION: &str = "log";
+const INDEX_EXTENSION: &str = "index";
/// A persisted segment recovered from disk: its metadata plus the storage
/// handles (readers/writers) opened over its `.log` / `.index` files.
@@ -67,7 +70,7 @@ pub async fn load_persisted_segments(
let partition_path = config
.system
.get_partition_path(stream_id, topic_id, partition_id);
- sweep_staging_files(&partition_path);
+ sweep_scratch_files(&partition_path);
let mut start_offsets = collect_segment_start_offsets(&partition_path)?;
start_offsets.sort_unstable();
@@ -171,7 +174,13 @@ pub async fn load_persisted_segments(
last.segment.sealed = false;
}
- ensure_contiguous_chain(&recovered, stream_id, topic_id, partition_id)?;
+ ensure_contiguous_chain(
+ &recovered,
+ &partition_path,
+ stream_id,
+ topic_id,
+ partition_id,
+ )?;
Ok(recovered)
}
@@ -181,12 +190,27 @@ pub async fn load_persisted_segments(
/// an operator copy) would otherwise splice a hole or an overlap into the
/// chain and push `current_offset` past data this replica does not hold.
/// Refuse loudly instead of serving a holed log.
+///
+/// The refusal names the partition and its directory so the caller can fence
+/// THAT group rather than abort the node's boot: the shapes it rejects are
+/// exactly what a failed quarantine leaves behind, and one damaged local chain
+/// must not take the whole node down.
fn ensure_contiguous_chain(
recovered: &[RecoveredSegment],
+ partition_path: &str,
stream_id: usize,
topic_id: usize,
partition_id: usize,
) -> Result<(), ServerNgError> {
+ let refused = |reason| {
+ Err(ServerNgError::PartitionChainRefused {
+ dir: PathBuf::from(partition_path),
+ stream_id,
+ topic_id,
+ partition_id,
+ reason,
+ })
+ };
for pair in recovered.windows(2) {
let previous = &pair[0].segment;
let next = &pair[1].segment;
@@ -196,51 +220,80 @@ fn ensure_contiguous_chain(
// more chain is exactly what a failed converge rebuild leaves behind.
// Skipping it here was the guard's blind spot.
if previous.size == IggyByteSize::default() {
- error!(
- stream_id,
- topic_id,
- partition_id,
- empty_start = previous.start_offset,
- next_start = next.start_offset,
- "recovered chain holds an empty non-tail segment; refusing to serve past it"
- );
- return Err(IggyError::CannotReadPartitions.into());
+ return refused(PartitionChainRefusal::EmptyNonTailSegment {
+ empty_start: previous.start_offset,
+ next_start: next.start_offset,
+ });
}
if next.start_offset != previous.end_offset + 1 {
- error!(
- stream_id,
- topic_id,
- partition_id,
- previous_start = previous.start_offset,
- previous_end = previous.end_offset,
- next_start = next.start_offset,
- "recovered segment chain is not contiguous; refusing to serve a holed log"
- );
- return Err(IggyError::CannotReadPartitions.into());
+ return refused(PartitionChainRefusal::Hole {
+ previous_start: previous.start_offset,
+ previous_end: previous.end_offset,
+ next_start: next.start_offset,
+ });
}
}
Ok(())
}
-/// Unlink every `*.staging` spill file in the partition directory. Boot is
-/// the one sweep that always runs: the install-time and reuse-time sweeps
-/// only fire on the NEXT transfer attempt, so a transfer abandoned for good
-/// would otherwise leak a full partition copy across restarts. Staging files
-/// are pure scratch (never a rename source until an install owns them), so
-/// unlinking is always safe here.
-fn sweep_staging_files(partition_path: &str) {
+/// Unlink the partition directory's scratch leftovers: every `*.staging` spill
+/// file, and every `.index` with no `.log` beside it.
+///
+/// Boot is the one sweep that always runs. The install-time and reuse-time
+/// staging sweeps only fire on the NEXT transfer attempt, so a transfer
+/// abandoned for good would otherwise leak a full partition copy across
+/// restarts; staging files are pure scratch (never a rename source until an
+/// install owns them), so unlinking is always safe.
+///
+/// Orphaned indexes come from the state-transfer install, which renames ALL
+/// indexes to their final names, fsyncs the directory, and only then renames the
+/// logs -- a crash in that window is GUARANTEED to leave final-name `.index`
+/// files with no `.log`. Recovery keys on `.log` stems, so nothing else ever
+/// looks at them again: they are invisible to it and to the size stats, and
+/// without this they are a permanent leak at offsets the partition may never
+/// revisit. Unlinking rather than keeping them is safe because every path that
+/// 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;
};
+ let mut swept = Vec::new();
+ let mut orphan_candidates = Vec::new();
+ let mut log_stems = std::collections::HashSet::new();
for entry in entries.flatten() {
let path = entry.path();
- let is_staging = path.to_str().is_some_and(|path| path.ends_with(".staging"));
- if is_staging && let Err(error) = fs::remove_file(&path) {
+ let Some(as_str) = path.to_str() else {
+ continue;
+ };
+ if as_str.ends_with(STAGING_SUFFIX) {
+ swept.push(path);
+ continue;
+ }
+ match path.extension().and_then(|extension| extension.to_str()) {
+ Some(LOG_EXTENSION) => {
+ if let Some(stem) = path.file_stem().and_then(|stem| stem.to_str()) {
+ log_stems.insert(stem.to_owned());
+ }
+ }
+ Some(INDEX_EXTENSION) => orphan_candidates.push(path),
+ _ => {}
+ }
+ }
+ swept.extend(orphan_candidates.into_iter().filter(|path| {
+ !path
+ .file_stem()
+ .and_then(|stem| stem.to_str())
+ .is_some_and(|stem| log_stems.contains(stem))
+ }));
+ for path in swept {
+ if let Err(error) = fs::remove_file(&path) {
warn!(
partition_path,
path = %path.display(),
%error,
- "failed to sweep a stale staging file at boot"
+ "failed to sweep a stale scratch file at boot"
);
}
}
diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs
index d0d06a8be..317596069 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -140,6 +140,22 @@ pub enum ServerNgError {
expected: u128,
found: u128,
},
+ // Per-partition, not fatal: the boot path fences this one group (quarantines
+ // its segment files and materialises it fresh) instead of taking the node
+ // down for one damaged local chain. The shapes it reports are exactly what a
+ // failed state-transfer quarantine leaves behind, and the rebuild recovers
+ // the data from a peer.
+ #[error(
+ "partition {stream_id}/{topic_id}/{partition_id} at {dir} recovered an \
+ unusable segment chain: {reason}"
+ )]
+ PartitionChainRefused {
+ dir: PathBuf,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ reason: PartitionChainRefusal,
+ },
#[error(
"shard {shard_id} aborted while waiting for shard-0 to broadcast the metadata \
factory bundle; shard 0 dropped its sender (most likely it failed to recover)"
@@ -233,6 +249,49 @@ pub enum ServerNgError {
ShardJoinFailures { failures: Vec<ShardJoinFailure> },
}
+/// Why a recovered segment chain cannot be served.
+///
+/// Both shapes mean the same thing operationally -- the local files do not form
+/// a chain this replica can serve -- but they are distinguished because they
+/// point at different causes: an empty non-tail segment is a failed rebuild's
+/// orphan pairing, a hole is a stray or half-unlinked file.
+#[derive(Debug)]
+pub enum PartitionChainRefusal {
+ EmptyNonTailSegment {
+ empty_start: u64,
+ next_start: u64,
+ },
+ Hole {
+ previous_start: u64,
+ previous_end: u64,
+ next_start: u64,
+ },
+}
+
+impl std::fmt::Display for PartitionChainRefusal {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::EmptyNonTailSegment {
+ empty_start,
+ next_start,
+ } => write!(
+ f,
+ "segment {empty_start} is empty yet {next_start} follows it, so the \
+ chain cannot be served past it"
+ ),
+ Self::Hole {
+ previous_start,
+ previous_end,
+ next_start,
+ } => write!(
+ f,
+ "segment {previous_start} ends at offset {previous_end} but the next \
+ starts at {next_start}, leaving a hole"
+ ),
+ }
+ }
+}
+
/// Per-shard outcome captured by [`crate::bootstrap::ShardHandles::join_all`]
/// when a shard either returned `Err` or panicked.
///
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index b7d291b0c..6eee83032 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -843,9 +843,7 @@ struct MetadataTransferSession {
/// chunk-serve time, so n retained gigabytes never pin n resident gigabytes.
enum ServedOffer {
Metadata(Rc<metadata::StateTransferOffer>),
- Partition {
- offer: Rc<partitions::state_transfer::PartitionStateTransferOffer>,
- },
+ Partition(Rc<partitions::state_transfer::PartitionStateTransferOffer>),
}
/// Shard-wide cache of segment payloads loaded to serve partition chunks,
@@ -860,35 +858,49 @@ struct ServedSegmentCache {
entries: HashMap<(u64, u64), CachedSegmentPayload>,
resident_bytes: u64,
use_seq: u64,
- /// `use_seq` as of the previous idle sweep; entries at or below it have
- /// served nothing for a full sweep interval.
- last_sweep_seq: u64,
+ /// Idle sweeps run so far; entries carry the reading at their last use, so
+ /// their age is measured on the OFFER clock rather than the raw tick.
+ sweeps: u64,
}
-/// One resident payload plus its last-use sequence (the LRU key).
+/// One resident payload, its last-use sequence (the LRU key), and the sweep
+/// reading at that use (the age key).
struct CachedSegmentPayload {
payload: Rc<Vec<u8>>,
last_use: u64,
+ last_use_sweep: u64,
}
impl ServedSegmentCache {
- /// Byte budget across all resident segment payloads on ONE shard (the
- /// process-wide bound is this times the shard count). LRU pressure from
- /// new inserts plus the idle sweep below reclaim it; a config knob can
- /// follow if operators need to trade it against page cache.
+ /// Byte budget across all resident segment payloads on ONE shard, so the
+ /// process-wide bound is this times the shard count. LRU pressure from new
+ /// inserts plus the idle sweep below reclaim it; a single segment larger than
+ /// the budget still loads (the serve could not proceed otherwise) and owns
+ /// the budget until it ages out. A config knob can follow if operators need
+ /// to trade it against page cache.
const RESIDENT_BYTES_MAX: u64 = 1 << 30;
- /// Drop every payload not touched since the previous idle sweep. 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).
+ /// Sweeps a payload survives without serving a chunk.
+ ///
+ /// 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) {
- let floor = self.last_sweep_seq;
- self.last_sweep_seq = self.use_seq;
+ self.sweeps += 1;
+ let floor = self.sweeps.saturating_sub(Self::IDLE_SWEEPS_MAX);
let stale: Vec<(u64, u64)> = self
.entries
.iter()
- .filter(|(_, cached)| cached.last_use <= floor)
+ .filter(|(_, cached)| cached.last_use_sweep <= floor)
.map(|(&key, _)| key)
.collect();
for key in stale {
@@ -902,8 +914,11 @@ impl ServedSegmentCache {
fn get(&mut self, namespace: u64, checksum: u64) -> Option<Rc<Vec<u8>>> {
self.use_seq += 1;
+ let use_seq = self.use_seq;
+ let sweeps = self.sweeps;
let cached = self.entries.get_mut(&(namespace, checksum))?;
- cached.last_use = self.use_seq;
+ cached.last_use = use_seq;
+ cached.last_use_sweep = sweeps;
Some(Rc::clone(&cached.payload))
}
@@ -931,6 +946,7 @@ impl ServedSegmentCache {
CachedSegmentPayload {
payload,
last_use: self.use_seq,
+ last_use_sweep: self.sweeps,
},
) {
self.resident_bytes = self
@@ -959,6 +975,48 @@ struct ServedStateTransfer {
fully_served: bool,
}
+/// One `StateTransferTarget` descriptor: the offer if there is one, plus what
+/// the serving replica knows about its own progress.
+///
+/// The progress fields ride along even on a refusal, so a receiver can tell a
+/// peer that is momentarily behind from one that knows less than it does.
+struct TransferDescriptor<'a> {
+ /// `Some((manifest, commit_op))` when the peer can serve.
+ offer: Option<(&'a [consensus::StateArtifact], u64)>,
+ /// Serving replica's view and commit frontier at build time.
+ view: u32,
+ commit_max: u64,
+ /// A refusal the requester should retry soon WITHOUT charging its
+ /// consecutive-failure count. Always false when `offer` is `Some`.
+ transient: bool,
+}
+
+impl<'a> TransferDescriptor<'a> {
+ const fn available(offer: &'a [consensus::StateArtifact], commit_op: u64) -> Self {
+ Self {
+ offer: Some((offer, commit_op)),
+ view: 0,
+ commit_max: 0,
+ transient: false,
+ }
+ }
+
+ const fn unavailable(transient: bool) -> Self {
+ Self {
+ offer: None,
+ view: 0,
+ commit_max: 0,
+ transient,
+ }
+ }
+
+ const fn serving(mut self, view: u32, commit_max: u64) -> Self {
+ self.view = view;
+ self.commit_max = commit_max;
+ self
+ }
+}
+
/// What `on_request_state_chunk` decided inside its offers borrow; the wire
/// sends run after the borrow drops.
enum ChunkReply {
@@ -1715,13 +1773,15 @@ where
}
ReconcileOp::ConfirmRemove { namespace } => {
// Tombstone bit set + shards_table row removed synchronously
- // by the reconciler before this op was enqueued, so no
- // in-flight frame can reach the partition between `remove`
- // and the drop here. Teardown already unlinked the on-disk
- // hierarchy via `delete_partitions_from_disk`, so the
- // partition drops inline: its compio file handles close
- // through io_uring without blocking, and no fsync is wanted
- // on data that is already gone.
+ // before this op was enqueued -- by the reconciler on a real
+ // delete, by `fence_partition_for_rebuild` when a partition
+ // is retired for rebuild -- so no in-flight frame can reach
+ // the partition between `remove` and the drop here. On the
+ // delete path teardown already unlinked the on-disk hierarchy
+ // via `delete_partitions_from_disk`, so the partition drops
+ // inline: its compio file handles close through io_uring
+ // without blocking, and no fsync is wanted on data that is
+ // already gone.
let removed = partitions.remove(&namespace);
partitions.untombstone(&namespace);
// A topic created then deleted before its `InsertOwned`
@@ -1758,13 +1818,25 @@ where
}
}
-/// The next replica to try after a transfer against `failed_peer` failed:
-/// walk the ring past the failed peer, skipping this replica. A cluster of
-/// two has no alternative and retries the same peer.
-const fn next_transfer_peer(self_id: u8, failed_peer: u8, replica_count: u8) -> u8 {
+/// The next replica to try after a transfer against `failed_peer` failed.
+///
+/// Prefers the view's primary: it is the only replica that can pass the serving
+/// side's caught-up-primary gate, so rotating by ring index alone can spend a
+/// full backoff round on a backup that must refuse -- and, worse, can land on a
+/// phantom view-0 primary of an empty group. Falls back to walking the ring past
+/// the failed peer, skipping this replica; a cluster of two has no alternative
+/// and retries the same peer.
+///
+/// `failed_peer` and `primary` are both bounded by `replica_count` at their
+/// ingress, which is what keeps the `+ 1` here from wrapping a peer id of 255
+/// onto replica 0.
+const fn next_transfer_peer(self_id: u8, failed_peer: u8, replica_count: u8, primary: u8) -> u8 {
if replica_count <= 1 {
return failed_peer;
}
+ if primary != self_id && primary != failed_peer {
+ return primary;
+ }
let mut candidate = (failed_peer + 1) % replica_count;
if candidate == self_id {
candidate = (candidate + 1) % replica_count;
@@ -2898,9 +2970,25 @@ where
/// `-p iggy-server-ng` build excludes the `simulator` feature and this
/// method; `cargo build --workspace` compiles it in but with no
/// production caller.
+ /// `superblock` is this group's durable `(view, log_view)` store. Passing
+ /// `None` keeps the storeless branch, where the persist gate marks every view
+ /// durable without writing anything -- fine for specs that never restart a
+ /// replica, but it means the gate itself, its write-failure fence, and view
+ /// recovery are all unexercised. A caller that hands one in (the simulator,
+ /// which retains the store across a replica rebuild) gets the production
+ /// contract: a recorded view is restored before the group joins, and a failed
+ /// write withholds every view-scoped send.
+ ///
+ /// `recovered_state` is that store's last record, read by the caller (the
+ /// store's read is async and this is not), mirroring how `new_shard` takes the
+ /// metadata plane's.
#[cfg(any(test, feature = "simulator"))]
- pub fn init_partition(&self, namespace: IggyNamespace)
- where
+ pub fn init_partition(
+ &self,
+ namespace: IggyNamespace,
+ superblock: Option<Rc<SB>>,
+ recovered_state: Option<consensus::VsrState>,
+ ) where
B: MessageBus + Clone,
{
let partitions = self.plane.partitions();
@@ -2908,7 +2996,7 @@ where
return;
}
- let consensus = VsrConsensus::with_clock(
+ let mut consensus = VsrConsensus::with_clock(
self.partition_consensus.cluster_id,
self.partition_consensus.self_replica_id,
self.partition_consensus.replica_count,
@@ -2917,15 +3005,25 @@ where
LocalPipeline::new(),
self.partition_consensus.clock.clone(),
);
+ // Recorded view first, exactly as the two boot paths order it: restoring
+ // after `init` would advertise a view older than the recorded one.
+ if let Some(state) = recovered_state {
+ consensus.set_view(state.view);
+ consensus.set_log_view(state.log_view);
+ consensus.mark_superblock_durable(state.view, state.log_view);
+ }
consensus.init();
let stats = Arc::new(PartitionStats::default());
- let partition = IggyPartition::with_in_memory_storage(
+ let mut partition = IggyPartition::with_in_memory_storage(
stats,
consensus,
partitions.config().segment_size,
partitions.config().enforce_fsync,
);
+ if let Some(superblock) = superblock {
+ partition.set_superblock(superblock);
+ }
partitions.insert(namespace, partition);
}
@@ -3217,8 +3315,14 @@ where
peer = header.replica,
"adopted a live view while awaiting transfer; requesting partition state transfer"
);
- self.arm_partition_transfer(partition, header.replica, transfers_inflight)
- .await;
+ // The announcing replica becomes `session.peer`, which the re-arm
+ // path feeds to `next_transfer_peer`'s ring arithmetic, so an id
+ // outside the cluster must not get that far.
+ if self.peer_is_known(header.replica, "StartView") {
+ let _ = self
+ .arm_partition_transfer(partition, header.replica, transfers_inflight)
+ .await;
+ }
return;
}
// A commit walk during Fetching can advance commit_min past the
@@ -3627,16 +3731,26 @@ where
});
let header = *msg.header();
let planes = self.plane.inner();
- // `header.namespace == 0`: pre-upgrade metadata WAL entries were
- // journaled before prepares stamped `consensus.namespace()`, and
- // repair ships stored bytes verbatim -- without this acceptance a
- // mixed-version metadata repair re-ships the same 0-stamped entries
- // forever. Safe to claim for the metadata plane: partition ops never
- // reach this arm with a metadata operation code, and namespace 0's
- // partition-side gates journal nothing for foreign operations.
+ // Legacy acceptance: pre-upgrade metadata WAL entries were journaled
+ // before prepares stamped `consensus.namespace()`, and repair ships
+ // stored bytes verbatim, so without it a mixed-version metadata repair
+ // re-ships the same 0-stamped entries forever.
+ //
+ // Fenced rather than left to operation classification: namespace 0 is
+ // also `IggyNamespace::new(0, 0, 0)`, the first partition of a fresh
+ // cluster, and `!is_partition()` is true of every operation code below
+ // `SendMessages`, so the first sub-160 partition operation ever added
+ // would route partition-0 repair frames into the METADATA journal. The
+ // claim therefore also requires a metadata repair actually in flight and
+ // no partition materialised under that namespace on this shard; a
+ // legacy frame arriving without both is dropped, which stalls visibly
+ // instead of storing partition bytes as metadata.
+ let legacy_metadata_claim = header.namespace == 0
+ && !header.operation.is_partition()
+ && self.metadata_repair.borrow().is_some()
+ && !planes.1.0.contains(&IggyNamespace::from_raw(0));
if let Some(ref consensus) = planes.0.consensus
- && (consensus.namespace() == header.namespace
- || (header.namespace == 0 && !header.operation.is_partition()))
+ && (consensus.namespace() == header.namespace || legacy_metadata_claim)
{
let session = *self.metadata_repair.borrow();
let Some(session) = session else {
@@ -3850,9 +3964,18 @@ where
attempts = partition.transfer_attempts(),
"partition repair floor unreachable; converting to state transfer"
);
- partition.consensus().begin_state_transfer_await();
- self.arm_partition_transfer(partition, header.replica, transfers_inflight)
- .await;
+ // Same reason as the StartView arm: this id becomes
+ // `session.peer` and later reaches the peer rotation.
+ if self.peer_is_known(header.replica, "RepairRangeReply") {
+ partition.consensus().begin_state_transfer_await();
+ let _ = self
+ .arm_partition_transfer(
+ partition,
+ header.replica,
+ transfers_inflight,
+ )
+ .await;
+ }
} else {
// The refusal cleared the repair session, so falling
// through would log "repair complete" right after
@@ -4093,11 +4216,13 @@ where
target: u8,
nonce: u128,
namespace: u64,
- offer: Option<(&[consensus::StateArtifact], u64)>,
+ descriptor: TransferDescriptor<'_>,
) where
B: MessageBus,
{
- let manifest = offer.map(|(entries, _)| consensus::encode_state_manifest(entries));
+ let manifest = descriptor
+ .offer
+ .map(|(entries, _)| consensus::encode_state_manifest(entries));
let total_size =
size_of::<StateTransferTargetHeader>() + manifest.as_ref().map_or(0, Vec::len);
let mut msg = Message::<StateTransferTargetHeader>::new(total_size);
@@ -4111,7 +4236,13 @@ where
h.nonce = nonce;
h.namespace = namespace;
h.size = total_size as u32;
- if let Some((_, commit_op)) = offer {
+ // The serving replica's own progress travels with every descriptor,
+ // available or not: it is what lets a receiver refuse an offer from
+ // a replica that knows less than it does.
+ h.view = descriptor.view;
+ h.commit_max = descriptor.commit_max;
+ h.unavailable_transient = u8::from(descriptor.transient);
+ if let Some((_, commit_op)) = descriptor.offer {
h.available = 1;
h.commit_op = commit_op;
}
@@ -4223,7 +4354,8 @@ where
header.replica,
header.nonce,
header.namespace,
- Some((&offer.manifest(), offer.commit_op)),
+ TransferDescriptor::available(&offer.manifest(), offer.commit_op)
+ .serving(consensus.view(), consensus.commit_max()),
)
.await;
return;
@@ -4246,7 +4378,8 @@ where
header.replica,
header.nonce,
header.namespace,
- Some((&offer.manifest(), offer.commit_op)),
+ TransferDescriptor::available(&offer.manifest(), offer.commit_op)
+ .serving(consensus.view(), consensus.commit_max()),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
@@ -4276,7 +4409,8 @@ where
header.replica,
header.nonce,
header.namespace,
- None,
+ TransferDescriptor::unavailable(false)
+ .serving(consensus.view(), consensus.commit_max()),
)
.await;
}
@@ -4701,7 +4835,7 @@ where
header.replica,
header.nonce,
header.namespace,
- None,
+ TransferDescriptor::unavailable(false),
)
.await;
}
@@ -5059,6 +5193,17 @@ 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();
+
for namespace in namespaces {
let Some(partition) = partitions.get_by_ns(&namespace) else {
continue;
@@ -5206,13 +5351,16 @@ where
}
};
if let Some(peer) = rearm_peer {
- let transfers_inflight = self.partition_transfers_inflight();
let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
continue;
};
partition.consensus().begin_state_transfer_await();
- self.arm_partition_transfer(partition, peer, transfers_inflight)
+ let armed = self
+ .arm_partition_transfer(partition, peer, transfers_inflight)
.await;
+ if armed {
+ transfers_inflight += 1;
+ }
}
}
}
@@ -5260,12 +5408,18 @@ where
/// or finished transfer ages out.
/// Serve one partition `RequestStateTransfer`: build (or re-serve) this
/// group's offer and answer with the descriptor.
- #[allow(clippy::future_not_send)]
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
async fn on_partition_request_state_transfer(&self, msg: &Message<RequestStateTransferHeader>)
where
B: MessageBus,
{
let header = *msg.header();
+ // Before anything keyed by the requester: the offer map's documented
+ // bound is the replica count, and an unvalidated id makes it 256 entries
+ // per served group, each pinning an offer for its full expiry.
+ if !self.peer_is_known(header.replica, "RequestStateTransfer") {
+ return;
+ }
let planes = self.plane.inner();
let config = planes.1.0.config().clone();
let Some(partition) = planes
@@ -5277,6 +5431,8 @@ where
};
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
+ let serving_view = partition.consensus().view();
+ let serving_commit_max = partition.consensus().commit_max();
// First-wins per (requester, nonce), exactly as the metadata arm: a
// stall retry reuses the nonce, and rebuilding under it could hand
@@ -5289,7 +5445,7 @@ where
.get_mut(&(header.namespace, header.replica))
.filter(|served| served.nonce == header.nonce)
.and_then(|served| {
- let ServedOffer::Partition { offer, .. } = &served.offer else {
+ let ServedOffer::Partition(offer) = &served.offer else {
return None;
};
served.idle_ticks = 0;
@@ -5308,7 +5464,8 @@ where
header.replica,
header.nonce,
header.namespace,
- Some((&offer.manifest(), offer.commit_op)),
+ TransferDescriptor::available(&offer.manifest(), offer.commit_op)
+ .serving(serving_view, serving_commit_max),
)
.await;
return;
@@ -5331,14 +5488,15 @@ where
header.replica,
header.nonce,
header.namespace,
- Some((&offer.manifest(), offer.commit_op)),
+ TransferDescriptor::available(&offer.manifest(), offer.commit_op)
+ .serving(serving_view, serving_commit_max),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
(header.namespace, header.replica),
ServedStateTransfer {
nonce: header.nonce,
- offer: ServedOffer::Partition { offer },
+ offer: ServedOffer::Partition(offer),
idle_ticks: 0,
fully_served: false,
},
@@ -5347,11 +5505,15 @@ where
Err(reason) => {
// The ACTUAL reason: "not the caught-up primary" is routine
// (the requester re-targets), an unreadable segment is an
- // operator-visible fault on THIS node.
+ // operator-visible fault on THIS node. The requester cannot see
+ // the reason, only whether it was transient, which is what keeps
+ // a routine refusal from charging its failure count.
+ let transient = reason.transient();
tracing::info!(
shard = self.id,
namespace_raw = header.namespace,
requester = header.replica,
+ transient,
%reason,
"cannot serve partition state transfer; requester falls back"
);
@@ -5361,7 +5523,8 @@ where
header.replica,
header.nonce,
header.namespace,
- None,
+ TransferDescriptor::unavailable(transient)
+ .serving(serving_view, serving_commit_max),
)
.await;
}
@@ -5396,6 +5559,10 @@ where
}
let header = *msg.header();
+ // The requester id keys the offer map, whose bound is the replica count.
+ if !self.peer_is_known(header.replica, "RequestStateChunk") {
+ return;
+ }
let planes = self.plane.inner();
let Some(partition) = planes
.1
@@ -5416,7 +5583,7 @@ where
let Some(served) = served else {
break 'attempt ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer));
};
- let ServedOffer::Partition { offer } = &served.offer else {
+ let ServedOffer::Partition(offer) = &served.offer else {
break 'attempt ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer));
};
let last_artifact = offer.artifact_count().saturating_sub(1);
@@ -5460,8 +5627,18 @@ where
let Some(payload) = artifact_bytes.get(start..end) else {
break 'attempt ChunkAttempt::Reply(None);
};
- if artifact == last_artifact && end >= artifact_bytes.len() {
+ if artifact == last_artifact && end >= artifact_bytes.len() && !served.fully_served
+ {
served.fully_served = true;
+ // Once per transfer, at the last byte of the last artifact.
+ // The descriptor log only proves a REQUEST arrived; this is
+ // the serving side's proof that the pull ran to completion.
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ requester = header.replica,
+ "partition state transfer fully served"
+ );
}
served.idle_ticks = 0;
let total_size = size_of::<StateChunkHeader>() + payload.len();
@@ -5483,7 +5660,16 @@ where
match attempt {
ChunkAttempt::Reply(reply) => break reply,
ChunkAttempt::Load { log_path, entry } => {
- if let Some(bytes) = load_partition_artifact(&log_path, &entry).await {
+ // 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.
+ if let Some(bytes) = partitions::state_transfer::load_verified_segment_artifact(
+ &log_path, &entry,
+ )
+ .await
+ {
self.served_segment_cache.borrow_mut().insert(
header.namespace,
entry.checksum,
@@ -5531,7 +5717,7 @@ where
header.replica,
header.nonce,
header.namespace,
- None,
+ TransferDescriptor::unavailable(false),
)
.await;
}
@@ -5570,6 +5756,12 @@ where
const PARTITION_TRANSFERS_INFLIGHT_MAX: usize = 4;
/// Receiving-side transfers currently in flight on this shard.
+ ///
+ /// One scan per call, so callers hoist it: with per-partition groups a
+ /// per-namespace call inside the tick sweep is O(P^2) exactly during a
+ /// node-wide view change or rejoin, and capped arms reschedule on the flat
+ /// retry interval, so the losers stay phase-locked and the sweep repeats
+ /// every interval for the whole rejoin.
fn partition_transfers_inflight(&self) -> usize {
let partitions = self.plane.partitions();
let namespaces: Vec<_> = partitions.namespaces().copied().collect();
@@ -5583,29 +5775,102 @@ where
.count()
}
+ /// Whether a peer-supplied source replica id names a replica of this
+ /// cluster.
+ ///
+ /// `header.replica` arrives unvalidated on every frame, and the partition
+ /// transfer paths turn it into ring arithmetic ([`next_transfer_peer`], where
+ /// id 255 panics in debug and wraps to replica 0 in release -- a silent
+ /// retarget) and into the served-offer map key, whose documented bound is the
+ /// replica count rather than 256 entries per served group. One check at the
+ /// frame's ingress closes both.
+ fn peer_is_known(&self, replica: u8, frame: &'static str) -> bool {
+ let replica_count = self.partition_consensus.replica_count;
+ if replica < replica_count {
+ return true;
+ }
+ tracing::warn!(
+ shard = self.id,
+ frame,
+ replica,
+ replica_count,
+ "dropping a partition frame whose source replica is outside this cluster"
+ );
+ false
+ }
+
+ /// Fence one partition for rebuild: quarantine its segment files, drop it
+ /// from routing, and queue the retirement the reconciler re-materialises
+ /// from committed metadata.
+ ///
+ /// Used wherever a partition is left without a serviceable segment chain (a
+ /// failed state-transfer install whose convergence also failed, a purge that
+ /// could not plant its replacement segment): the next append or poll would
+ /// panic on `active_segment()`'s expect.
+ ///
+ /// `IggyPartitions` mandates external removals go through `ConfirmRemove` --
+ /// a direct `remove()` would invalidate the `&mut` the caller still holds --
+ /// but the tombstone and the routing row drop SYNCHRONOUSLY here, because
+ /// the tombstone is the only gate in `get_mut_by_ns` and the queue does not
+ /// drain until the end of the pump iteration.
+ #[allow(clippy::future_not_send)]
+ async fn fence_partition_for_rebuild(
+ &self,
+ namespace: IggyNamespace,
+ partition: &IggyPartition<B, SB>,
+ ) where
+ B: MessageBus + 'static,
+ T: ShardsTable,
+ {
+ match partition.quarantine_partition_dir().await {
+ Ok(Some(fenced_dir)) => tracing::error!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ fenced_dir,
+ "quarantined the fenced partition's segment files; they are kept for \
+ 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 \
+ re-hydrates whatever they hold and the boot guard refuses a holed chain"
+ ),
+ }
+ self.plane.partitions().tombstone(namespace);
+ self.shards_table.remove(&namespace);
+ self.enqueue_reconcile_op(ReconcileOp::ConfirmRemove { namespace });
+ self.signal_reconcile_wake();
+ }
+
/// Arm a fresh partition transfer session against `peer` and request its
/// descriptor. Every partition arming site goes through here. Drops any
/// repair session (transfer supersedes repair; a transfer-unavailable
- /// fallback re-arms repair fresh) and deliberately does NOT touch
- /// `transfer_attempts` (the budget survives re-mints) or the stage
- /// (callers own the `AwaitingTarget` transition).
+ /// fallback re-arms repair fresh) and leaves the stage to its callers (they
+ /// own the `AwaitingTarget` transition).
///
/// Refuses past [`Self::PARTITION_TRANSFERS_INFLIGHT_MAX`]: the arm
/// converts to a scheduled re-arm (no failure charged -- the local slot
/// shortage is not the peer's fault) and the stage returns to Idle so
/// journal repair keeps the gap visible meanwhile.
- #[allow(clippy::future_not_send)]
+ ///
/// `transfers_inflight` is computed by the caller BEFORE it formed its
/// `&mut partition`: the counting scan takes shared borrows of every
/// partition, and deriving a sibling `&` to the element the caller's
/// protected `&mut` points at is UB under both stacked and tree borrows,
- /// however innocuous the generated code is today.
+ /// however innocuous the generated code is today. Returns whether a session
+ /// was armed, so a caller sweeping many groups can carry the count forward
+ /// instead of re-scanning per group.
+ #[allow(clippy::future_not_send)]
async fn arm_partition_transfer(
&self,
partition: &mut IggyPartition<B, SB>,
peer: u8,
transfers_inflight: usize,
- ) where
+ ) -> bool
+ where
B: MessageBus,
{
if partition.transfer.is_none()
@@ -5621,14 +5886,16 @@ where
if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
}
+ partition.note_transfer_rearm_scheduled();
partition.transfer_rearm = Some(partitions::state_transfer::PendingTransferRearm {
peer,
after_ticks: self.repair_retry_ticks.get(),
});
- return;
+ return false;
}
partition.repair = None;
let nonce = iggy_common::random_id::get_uuid();
+ let armed = partition.transfer.is_none();
partition.transfer = Some(partitions::state_transfer::PartitionTransferSession {
nonce,
peer,
@@ -5639,6 +5906,7 @@ where
});
self.send_request_state_transfer(partition.consensus(), peer, nonce)
.await;
+ armed
}
/// Arm partition journal repair when this replica is lagging its group
@@ -5693,6 +5961,11 @@ where
T: ShardsTable,
{
let header = *msg.header();
+ // The peer id reaches `next_transfer_peer`'s ring arithmetic through the
+ // re-arm below, so it is validated before anything uses it.
+ if !self.peer_is_known(header.replica, "StateTransferTarget") {
+ return;
+ }
let planes = self.plane.inner();
let Some(partition) = planes
.1
@@ -5709,19 +5982,51 @@ where
return;
}
if header.available == 0 {
- // Charged like any other failure: `is_caught_up_primary` on the
- // serving side wants `commit_min == commit_max`, frequently
- // false under produce load, so an uncharged immediate re-arm
- // looped repair -> refusal -> arm -> unavailable at network
- // round-trip rate. The backoff + peer rotation live in the
- // shared re-arm path.
+ // A refusal the peer marked transient (it is momentarily not the
+ // caught-up primary, which `is_caught_up_primary` makes frequent
+ // under produce load) must not charge the consecutive-failure count:
+ // that count is reset only by a completed install, so ten routine
+ // refusals pin the re-arm backoff at its 1024x ceiling while nothing
+ // else recovers the partition -- repair keeps hitting the refused
+ // floor and will not arm while a re-arm is pending. A hard refusal
+ // (unreadable segment, failed flush) still charges.
+ let transient = header.unavailable_transient == 1;
tracing::info!(
shard = self.id,
namespace_raw = header.namespace,
peer = header.replica,
+ transient,
"partition transfer peer cannot serve; backing off before re-arming"
);
- self.abandon_or_rearm_partition_transfer(partition, header.replica)
+ if transient {
+ self.rearm_partition_transfer_after_refusal(partition, header.replica)
+ .await;
+ } else {
+ self.abandon_or_rearm_partition_transfer(partition, header.replica)
+ .await;
+ }
+ return;
+ }
+ // The serving replica's own progress, carried by every descriptor: an
+ // offer from a replica that knows LESS than this one does is the phantom
+ // view-0 primary signature (a group whose directory vanished boots
+ // `init()`, comes up Normal at view 0, and an empty log is trivially
+ // caught up). Installing it would unlink a chain this replica already
+ // holds; nonce match alone cannot tell the two apart.
+ let local_view = partition.consensus().view();
+ let local_commit_max = partition.consensus().commit_max();
+ if header.view < local_view || header.commit_max < local_commit_max {
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ peer = header.replica,
+ serving_view = header.view,
+ serving_commit_max = header.commit_max,
+ local_view,
+ local_commit_max,
+ "refusing a partition transfer offer from a replica behind this one"
+ );
+ self.rearm_partition_transfer_after_refusal(partition, header.replica)
.await;
return;
}
@@ -5776,6 +6081,16 @@ where
return;
}
let reused = partition.reuse_staged_segments(&entries).await;
+ if !reused.is_empty() {
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = header.namespace,
+ peer = header.replica,
+ adopted = reused.len(),
+ artifacts = entries.len(),
+ "adopted staged segments from an earlier transfer attempt"
+ );
+ }
// Re-check the nonce AFTER the await: the staging scan yields, and a
// session re-minted underneath it must not get stamped with this
// (now stale) offer's commit_op and manifest.
@@ -5788,16 +6103,23 @@ where
};
session.target_accepted = true;
session.commit_op = header.commit_op;
+ // No reservation here: the manifest's total is bounded only by
+ // `PARTITION_TRANSFER_TOTAL_LEN_MAX` (1 TiB), so reserving every
+ // artifact up front is an eager address-space commit of the whole
+ // manifest -- times the in-flight cap -- which turns fatal under strict
+ // overcommit, `RLIMIT_AS`, or cgroup accounting, and contradicts the
+ // session's own promise to bound receiver memory to ONE in-flight
+ // artifact. Artifacts adopted by the reuse scan below would also be
+ // reserved and then overwritten with `Staged`, making the retry path's
+ // reservation pure waste. `ArtifactProgress::extend_from_chunk` reserves
+ // on the first chunk of its own artifact instead.
session.artifacts = entries
.iter()
.map(|&entry| {
- // Exact reservation: `entry.len` already passed the per-kind
- // caps, and geometric growth to gigabyte sizes would copy
- // roughly twice the bytes with a ~1.5x transient peak.
- let mut buf = Vec::new();
- #[allow(clippy::cast_possible_truncation)]
- buf.reserve_exact(entry.len as usize);
- TransferArtifact::Pending(consensus::ArtifactProgress { entry, buf })
+ TransferArtifact::Pending(consensus::ArtifactProgress {
+ entry,
+ buf: Vec::new(),
+ })
})
.collect();
session.idle_ticks = 0;
@@ -6041,39 +6363,20 @@ where
) => {
// The partition holds no serviceable segment chain and its
// next append or poll would panic the shard. Fence exactly
- // this group: quarantine the directory (a failed converge
- // sweep can leave strays that `build_partition_fresh` would
- // never clear and the boot contiguity guard would then trip
- // on), then retire the in-memory partition through the
- // reconcile queue -- `IggyPartitions` mandates external
- // removals go through `ConfirmRemove`, and a direct
- // `remove()` here would invalidate the very `&mut` this
- // frame still holds. The reconciler then re-materialises a
- // fresh partition from committed metadata; its first repair
- // floor refusal re-arms a transfer, which re-seeds the
- // offset frontier from the offsets artifact.
+ // this group (a failed converge sweep can leave strays that
+ // `build_partition_fresh` would never clear and the boot
+ // contiguity guard would then trip on). The reconciler
+ // re-materialises a fresh partition from committed metadata;
+ // its first repair floor refusal re-arms a transfer, which
+ // re-seeds the offset frontier from the offsets artifact.
tracing::error!(
shard = self.id,
namespace_raw = namespace,
%error,
- "partition unserviceable after failed install; \
- quarantining its directory and fencing it for rebuild"
+ "partition unserviceable after failed install; fencing it for rebuild"
);
- let quarantined = partition.quarantine_partition_dir().await;
- if let Err(quarantine_error) = quarantined {
- // The directory survives in place; the rebuild below
- // re-hydrates whatever it holds, and the boot guard
- // refuses a holed chain rather than serving it.
- tracing::error!(
- shard = self.id,
- namespace_raw = namespace,
- error = %quarantine_error,
- "failed to quarantine the fenced partition directory"
- );
- }
- let target = IggyNamespace::from_raw(namespace);
- self.enqueue_reconcile_op(ReconcileOp::ConfirmRemove { namespace: target });
- self.signal_reconcile_wake();
+ self.fence_partition_for_rebuild(IggyNamespace::from_raw(namespace), partition)
+ .await;
}
Err(error) => {
tracing::error!(
@@ -6104,22 +6407,73 @@ where
) where
B: MessageBus,
{
- partition.transfer = None;
let failures = partition.record_transfer_failure();
+ let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures);
+ self.schedule_partition_transfer_rearm(partition, peer, failures, after_ticks)
+ .await;
+ }
+
+ /// Re-arm after a refusal the serving peer marked TRANSIENT: schedule the
+ /// next attempt on a flat interval and charge nothing.
+ ///
+ /// "The peer is momentarily not the caught-up primary" is the common case
+ /// under produce load, and `transfer_failures` is reset only by a completed
+ /// install, so charging it turns a transient into a stall measured in re-arm
+ /// ceilings: nothing else recovers the partition meanwhile, since repair
+ /// keeps hitting the refused floor and will not arm while a re-arm is
+ /// pending.
+ #[allow(clippy::future_not_send)]
+ async fn rearm_partition_transfer_after_refusal(
+ &self,
+ partition: &mut IggyPartition<B, SB>,
+ peer: u8,
+ ) where
+ B: MessageBus,
+ {
+ // Deliberately NOT `transfer_rearm_backoff`: a flat interval, so a peer
+ // that spends a minute catching up costs a minute of retries rather than
+ // a climb to the 1024x ceiling.
+ let after_ticks = self.repair_retry_ticks.get();
+ self.schedule_partition_transfer_rearm(partition, peer, 0, after_ticks)
+ .await;
+ }
+
+ /// Drop the session, rotate the peer, and schedule the re-arm; shared by the
+ /// charged and uncharged paths.
+ #[allow(clippy::future_not_send)]
+ async fn schedule_partition_transfer_rearm(
+ &self,
+ partition: &mut IggyPartition<B, SB>,
+ peer: u8,
+ failures: u32,
+ after_ticks: u32,
+ ) where
+ B: MessageBus,
+ {
+ partition.transfer = None;
let consensus = partition.consensus();
if consensus.state_transfer_stage() != consensus::StateTransferStage::Idle {
consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle);
}
- let next_peer = next_transfer_peer(consensus.replica(), peer, consensus.replica_count());
- let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures);
+ let next_peer = next_transfer_peer(
+ consensus.replica(),
+ peer,
+ consensus.replica_count(),
+ consensus.primary_index(consensus.view()),
+ );
tracing::info!(
shard = self.id,
namespace_raw = partition.consensus().namespace(),
failures,
next_peer,
after_ticks,
- "partition transfer failed; scheduling a backed-off re-arm"
+ "partition transfer did not land; scheduling a re-arm"
);
+ // The stall budget belongs to ONE attempt: carried across, an exhausted
+ // count left every later session a single retry-interval window to land
+ // its first response, against a backoff climbing to 1024x. Livelock
+ // across attempts is bounded by `transfer_failures` and that backoff.
+ partition.note_transfer_rearm_scheduled();
partition.transfer_rearm = Some(partitions::state_transfer::PendingTransferRearm {
peer: next_peer,
after_ticks,
@@ -6413,27 +6767,6 @@ where
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
}
-/// Load a served segment artifact and re-verify it against its manifest
-/// entry at load time, so serving-side corruption or a GC unlink surfaces
-/// HERE (offer eviction + requester restart) instead of shipping garbage
-/// the receiver burns a decode budget on. The file may legitimately be
-/// LONGER than the entry (an active segment kept appending after the offer
-/// was built); the artifact is the first `entry.len` bytes.
-#[allow(clippy::future_not_send)]
-async fn load_partition_artifact(
- log_path: &str,
- entry: &consensus::StateArtifact,
-) -> Option<Vec<u8>> {
- let mut bytes = compio::fs::read(log_path).await.ok()?;
- #[allow(clippy::cast_possible_truncation)]
- if (bytes.len() as u64) < entry.len {
- return None;
- }
- #[allow(clippy::cast_possible_truncation)]
- bytes.truncate(entry.len as usize);
- consensus::verify_state_artifact(entry, &bytes).then_some(bytes)
-}
-
/// Re-stamp a stored prepare with the current view before retransmission.
/// After a view change the primary re-sends its uncommitted suffix as its
/// own prepares (VSR), but the journal keeps the original view stamp and
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 90eb71823..91e9591fb 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -675,13 +675,22 @@ where
generation,
"purge-partition reset partition to empty"
),
- Err(error) => tracing::error!(
- shard = self.id,
- namespace_raw = namespace.inner(),
- generation,
- %error,
- "purge-partition failed to reset partition"
- ),
+ Err(error) => {
+ // The purge drained every segment before its first
+ // fallible step, so a failure leaves the partition
+ // with no serviceable chain and the next append
+ // panics on `active_segment()`. Fence this one group
+ // for rebuild, exactly as a failed state-transfer
+ // convergence does.
+ tracing::error!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ generation,
+ %error,
+ "purge-partition failed to reset partition; fencing it for rebuild"
+ );
+ self.fence_partition_for_rebuild(namespace, partition).await;
+ }
}
}
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 90094b893..7a8b3ba67 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -47,7 +47,8 @@ use server_common::Message;
use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
use shard::CONSENSUS_TICK_INTERVAL;
use shard::shards_table::{ShardsTable, calculate_shard_assignment};
-use std::collections::HashSet;
+use std::cell::RefCell;
+use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::rc::Rc;
@@ -85,6 +86,14 @@ pub struct SimReplica {
/// yet the run stays byte-identical on replay. See
/// `VsrConsensus::set_incarnation`.
pub metadata_incarnation: u128,
+ /// One durable superblock per partition group this replica has materialised,
+ /// harness-owned for the same reason as the metadata one: the bytes survive
+ /// the shards being dropped and rebuilt, so a re-materialised group recovers
+ /// the `(view, log_view)` it recorded instead of re-entering view 0. Without
+ /// a store the persist gate marks every view durable without writing, which
+ /// leaves the gate, its write-failure fence, and view recovery all
+ /// unexercised.
+ pub partition_superblocks: RefCell<HashMap<IggyNamespace, Rc<SimSuperblock>>>,
/// Keeps each pump's stop channel alive; dropping one would end that
/// pump gracefully, which is reserved for future shutdown/restart
/// tests (crash uses `DetExecutor::abort` instead).
@@ -325,6 +334,7 @@ impl Simulator {
superblock,
metadata_journal,
metadata_incarnation,
+ partition_superblocks: RefCell::new(HashMap::new()),
_stop_txs: stop_txs,
pump_tasks,
});
@@ -364,7 +374,23 @@ impl Simulator {
}
let shard_count = u32::try_from(replica.shards.len()).expect("shard count fits u32");
let owner = calculate_shard_assignment(&namespace, shard_count);
- replica.shards[usize::from(owner)].init_partition(namespace);
+ // 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
@@ -749,6 +775,11 @@ impl Simulator {
// incarnation and still in flight is ignored. Deterministic, so replay stays
// byte-identical.
let metadata_incarnation = self.replicas[idx].metadata_incarnation + 1;
+ // Partition superblocks carry forward too: a group re-materialised after
+ // the restart must recover its recorded view from the same store, exactly
+ // as a rebooted server-ng partition reads the record in its directory.
+ let partition_superblocks =
+ std::mem::take(&mut *self.replicas[idx].partition_superblocks.borrow_mut());
// Recover the durable VSR state from the retained superblock before the
// rebuild, as production reads it in restore_metadata_consensus.
@@ -811,6 +842,7 @@ impl Simulator {
superblock,
metadata_journal,
metadata_incarnation,
+ partition_superblocks: RefCell::new(partition_superblocks),
_stop_txs: stop_txs,
pump_tasks,
};