This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch partition_superblock
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/partition_superblock by this
push:
new 1a2d35c63 address review comments
1a2d35c63 is described below
commit 1a2d35c637d498b300b9af860d33d7e836ef252e
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Wed Aug 5 19:55:28 2026 +0200
address review comments
---
core/binary_protocol/src/consensus/header.rs | 39 ++-
core/binary_protocol/src/consensus/operation.rs | 15 ++
core/consensus/src/impls.rs | 6 +-
core/consensus/src/state_transfer.rs | 30 ++-
core/integration/src/bench_utils.rs | 38 +--
.../tests/server/partition_view_durability_vsr.rs | 13 +
.../server/scenarios/purge_delete_scenario.rs | 24 +-
core/metadata/src/impls/metadata.rs | 3 +-
core/partitions/src/iggy_partition.rs | 164 ++++++++++++
core/partitions/src/journal.rs | 8 +-
core/partitions/src/state_transfer.rs | 276 ++++++++++++---------
core/server-ng/src/bootstrap.rs | 14 +-
core/shard/src/lib.rs | 259 ++++++++++++-------
13 files changed, 624 insertions(+), 265 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 6dd003ddb..12213676b 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -1413,15 +1413,6 @@ 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,
/// Set on an `available == 0` refusal that means "not right now" rather
than
@@ -1430,8 +1421,27 @@ pub struct StateTransferTargetHeader {
/// 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.
+ ///
+ /// This and `commit_max` below claim the HEAD of what used to be the
+ /// reserved tail, so every pre-existing field keeps its published offset:
+ /// this header ships in the `iggy_binary_protocol` crate, the size assert
+ /// cannot catch an equal-size reshuffle, and nothing on the link carries a
+ /// version signal -- a mid-struct insertion is silent non-interop between
+ /// mixed builds.
pub unavailable_transient: u8,
- pub reserved: [u8; 86],
+ /// Explicit padding so `commit_max` sits 8-aligned without the implicit
+ /// padding `NoUninit` forbids.
+ pub reserved_alignment: [u8; 6],
+ /// 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 reserved: [u8; 80],
}
const _: () = {
assert!(size_of::<StateTransferTargetHeader>() == HEADER_SIZE);
@@ -1439,7 +1449,14 @@ const _: () = {
offset_of!(StateTransferTargetHeader, nonce)
== offset_of!(StateTransferTargetHeader, reserved_frame) +
size_of::<[u8; 66]>()
);
- assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8;
86]>() == HEADER_SIZE);
+ // The pre-existing published offsets. New fields grow into the reserved
+ // tail only; a change that moves one of these is a wire break.
+ assert!(offset_of!(StateTransferTargetHeader, commit_op) == 144);
+ assert!(offset_of!(StateTransferTargetHeader, namespace) == 152);
+ assert!(offset_of!(StateTransferTargetHeader, available) == 160);
+ assert!(offset_of!(StateTransferTargetHeader, unavailable_transient) ==
161);
+ assert!(offset_of!(StateTransferTargetHeader, commit_max) == 168);
+ assert!(offset_of!(StateTransferTargetHeader, reserved) + size_of::<[u8;
80]>() == HEADER_SIZE);
};
impl ConsensusHeader for StateTransferTargetHeader {
diff --git a/core/binary_protocol/src/consensus/operation.rs
b/core/binary_protocol/src/consensus/operation.rs
index 11d15a106..026ff9bca 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -186,6 +186,21 @@ impl Operation {
(*self as u8) >= Self::PARTITION_START
}
+ /// Operations that replicate through the METADATA consensus group and live
+ /// in its WAL.
+ ///
+ /// Wider than [`Self::is_metadata`]: the session ops replicate on the
+ /// metadata plane without being metadata mutations. The single source of
+ /// truth for "does the metadata plane own this op", shared by the plane's
+ /// own applicability predicate and the repair router's legacy-stamp
+ /// acceptance -- the two drifting is how a metadata op ends up offered to
+ /// the partition arm.
+ #[must_use]
+ #[inline]
+ pub const fn is_metadata_plane(&self) -> bool {
+ self.is_metadata() || matches!(self, Self::Register | Self::Logout)
+ }
+
/// Operations clients are allowed to send directly.
#[must_use]
#[inline]
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 17c3c1fa2..dda3fa9a8 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -3046,7 +3046,11 @@ where
// never re-stamped (`restamp_prepare_view` patches only `view`), so
this
// survives view-change retransmits. The header `checksum` and its
`parent`
// chain stay `0`: activating them needs the retransmit path to
re-seal a
- // re-stamped header, a separate change.
+ // re-stamped header, a separate change. Whoever activates it must also
+ // audit every `set_last_prepare_checksum` caller for cross-plane
carry --
+ // the repair router in `shard` drops metadata-plane frames it cannot
+ // journal precisely so one cannot stamp a PARTITION consensus, which
is
+ // inert only while these values are structurally zero.
//
// Metadata plane only. A partition produce prepare already carries a
verified
// `batch_checksum` over the same bytes, so a second full-payload pass
is pure
diff --git a/core/consensus/src/state_transfer.rs
b/core/consensus/src/state_transfer.rs
index fc6c28645..0a1ff3768 100644
--- a/core/consensus/src/state_transfer.rs
+++ b/core/consensus/src/state_transfer.rs
@@ -133,16 +133,22 @@ pub fn next_pending_chunk<T: ChunkProgress>(
/// Append one received chunk; `true` only when bytes actually landed, which
/// is the caller's cue to reset its liveness counters and re-drive progress.
///
-/// Everything else is dropped without side effects: an out-of-range artifact
-/// index, a non-sequential offset (chunks are pulled lockstep, so anything
-/// else is a duplicate or reorder -- the stall retry re-requests from the
-/// current frontier), an overrun past the declared length, and a zero-byte
-/// payload. A zero-byte payload is not progress: it extends nothing and the
-/// same offset is re-requested immediately, and resetting liveness counters
-/// on one is what turned a short rebuilt offer into an unbounded empty-frame
-/// ping-pong on the metadata plane. The serving side refuses to produce
-/// these now; the guard stays because a peer running an older build still
-/// can.
+/// Everything else is dropped without side effects: an artifact that is not
+/// the FIRST incomplete one, a non-sequential offset (chunks are pulled
+/// lockstep, so anything else is a duplicate or reorder -- the stall retry
+/// re-requests from the current frontier), an overrun past the declared
+/// length, and a zero-byte payload. The first-incomplete restriction mirrors
+/// what [`next_pending_chunk`] would have requested: without it a peer that
+/// pushes one byte into EVERY manifest entry makes each artifact's
+/// first-chunk `reserve_exact` fire, committing address space for the sum of
+/// all declared lengths at once -- the whole-manifest reservation the
+/// receiver deliberately refuses to make up front, and a failed `Vec`
+/// reservation aborts the process rather than erroring. A zero-byte payload
+/// is not progress: it extends nothing and the same offset is re-requested
+/// immediately, and resetting liveness counters on one is what turned a
+/// short rebuilt offer into an unbounded empty-frame ping-pong on the
+/// metadata plane. The serving side refuses to produce these now; the guard
+/// stays because a peer running an older build still can.
#[must_use]
pub fn append_chunk<T: ChunkProgress>(
artifacts: &mut [T],
@@ -150,6 +156,10 @@ pub fn append_chunk<T: ChunkProgress>(
offset: u64,
payload: &[u8],
) -> bool {
+ let first_incomplete = artifacts.iter().position(|artifact|
!artifact.complete());
+ if first_incomplete != Some(artifact_index as usize) {
+ return false;
+ }
let Some(artifact) = artifacts.get_mut(artifact_index as usize) else {
return false;
};
diff --git a/core/integration/src/bench_utils.rs
b/core/integration/src/bench_utils.rs
index 966d6a25b..74c2d56bd 100644
--- a/core/integration/src/bench_utils.rs
+++ b/core/integration/src/bench_utils.rs
@@ -20,8 +20,7 @@ use assert_cmd::prelude::CommandCargoExt;
use iggy::prelude::*;
use iggy_common::TransportProtocol;
use std::{
- fs::{self, File, OpenOptions},
- io::Write,
+ fs::{self, File},
process::{Command, Stdio},
thread::{self, panicking},
time::{Duration, Instant},
@@ -127,35 +126,12 @@ pub fn run_bench_and_wait_for_finish(
}
};
- // 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 {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stderr_file_path)
- .unwrap()
- .write_all(stderr.as_bytes())
- .unwrap();
- }
-
- if let Some(stdout_file_path) = &stdout_file_path {
- OpenOptions::new()
- .append(true)
- .create(true)
- .open(stdout_file_path)
- .unwrap()
- .write_all(stdout.as_bytes())
- .unwrap();
- }
- }
-
+ // Nothing to drain, by construction: both branches above redirect the
+ // child's stdout and stderr -- to files, or inherited under
+ // `IGGY_TEST_VERBOSE` -- so no pipe exists for the poll loop to deadlock
+ // against. The old `wait_with_output` capture here could only ever return
+ // empty buffers for the same reason; the captures the failure path prints
+ // are the redirect FILES.
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)] {
diff --git a/core/integration/tests/server/partition_view_durability_vsr.rs
b/core/integration/tests/server/partition_view_durability_vsr.rs
index 6e6e42a77..acbbb8171 100644
--- a/core/integration/tests/server/partition_view_durability_vsr.rs
+++ b/core/integration/tests/server/partition_view_durability_vsr.rs
@@ -165,6 +165,19 @@ async fn
given_advanced_partition_view_when_survivor_restarts_should_recover_vie
// 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.
+ //
+ // Skipped when the harness inherits the node's stdout instead of capturing
+ // it (`IGGY_TEST_VERBOSE`): the log file is then empty, and asserting on
it
+ // would fail spuriously in exactly the mode someone debugging this would
+ // use. The serve check above still ran.
+ let log_captured = !harness.node(2).stdout_plain().is_empty();
+ if !log_captured {
+ eprintln!(
+ "IGGY_TEST_VERBOSE inherits node stdout, so the restored-view
oracle is \
+ unavailable; skipping it"
+ );
+ return;
+ }
let restored = restored_partition_view(harness, 2)
.expect("node 2 must log the partition view it restored from its
superblock");
assert!(
diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
index 77c9c53e2..97cf1c816 100644
--- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs
+++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs
@@ -985,15 +985,21 @@ pub async fn run_purge_topic(harness: &mut TestHarness,
restart_server: bool) {
await_segment_layout(&partition_path, &[0]).await;
// --- Verify consumer offsets cleared (memory + disk) ---
- // POLLED, 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);
+ // Polled ONLY in the restart cells: there the kill can land mid-purge, and
+ // boot then plants the [0] layout itself (fencing a torn chain, or
+ // recovering an already-drained directory) with the offset files still
+ // present -- the layout gate above is satisfied BEFORE the reconciler's
+ // re-purge (the applied generation is not persisted, so a restart
+ // re-purges) clears them. Without a restart the pump clears offsets and
+ // files in the SAME frame that plants the layout, so the instant assert is
+ // correct there and strictly stronger; a poll would hide a regression that
+ // clears them one frame late.
+ let offsets_deadline = std::time::Instant::now()
+ + if restart_server {
+ std::time::Duration::from_secs(10)
+ } else {
+ std::time::Duration::ZERO
+ };
loop {
let consumer_offset = client
.get_consumer_offset(&consumer, &stream_ident, &topic_ident,
Some(PARTITION_ID))
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 01e0916a2..55d4fb042 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1288,8 +1288,7 @@ where
message.header().command(),
Command2::Request | Command2::Prepare | Command2::PrepareOk
));
- let op = message.header().operation();
- op.is_metadata() || matches!(op, Operation::Register |
Operation::Logout)
+ message.header().operation().is_metadata_plane()
}
}
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 5761aee4e..7cf5f6b14 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -3435,6 +3435,19 @@ where
for path in consumer_paths.into_iter().chain(group_paths) {
let _ = delete_persisted_offset(&path).await;
}
+ // Directory fsync so those unlinks stick, mirroring the install path:
a
+ // crash right after the purge otherwise resurrects the offset files at
+ // boot, and while recovery clamps a resurrected offset down to the
+ // rebuilt head, "consumed through 0" is not the intended "no entry at
+ // all" -- that consumer skips the first post-purge message.
+ for dir in self
+ .consumer_offsets_path
+ .clone()
+ .into_iter()
+ .chain(self.consumer_group_offsets_path.clone())
+ {
+ let _ = crate::state_transfer::fsync_dir(&dir).await;
+ }
// The persisted-offset tracker mirrors the files unlinked above; a
// stale entry would make a post-purge auto-commit skip its write and
// lose the offset on restart.
@@ -5361,6 +5374,157 @@ mod tests {
assert_eq!(partition.consensus().commit_min(), 0);
assert!(partition.repair.is_some());
}
+ /// Temp partition directory for the state-transfer fence specs below.
+ async fn transfer_fence_dir(label: &str) -> String {
+ let dir = std::env::temp_dir().join(format!(
+ "iggy-transfer-fence-{label}-{}-{}",
+ std::process::id(),
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .expect("system clock after epoch")
+ .as_nanos(),
+ ));
+ compio::fs::create_dir_all(&dir)
+ .await
+ .expect("create temp partition dir");
+ dir.to_string_lossy().into_owned()
+ }
+
+ fn armed_transfer(peer: u8) ->
crate::state_transfer::PartitionTransferSession {
+ crate::state_transfer::PartitionTransferSession {
+ nonce: 7,
+ peer,
+ commit_op: 12,
+ artifacts: Vec::new(),
+ target_accepted: true,
+ idle_ticks: 0,
+ }
+ }
+
+ /// A purge must not leave a transfer running: its staged segments hold
+ /// PRE-purge data, and completing the install renames it back in durably
+ /// (the install takes `max(offer generation, applied)`, and this purge
+ /// already stamped the newer one, so the reconciler's purge gate never
+ /// re-fires).
+ #[compio::test]
+ async fn
given_armed_transfer_when_purged_should_abandon_session_and_rearm() {
+ let partition_dir = transfer_fence_dir("purge-abandons").await;
+ let mut partition = test_partition();
+ partition.set_partition_dir(partition_dir.clone());
+ partition.transfer = Some(armed_transfer(1));
+ partition.transfer_rearm =
Some(crate::state_transfer::PendingTransferRearm {
+ peer: 2,
+ after_ticks: 5,
+ });
+ partition.consensus().begin_state_transfer_await();
+
+ partition
+ .purge(&repair_config(), 3)
+ .await
+ .expect("purge partition");
+
+ assert!(
+ partition.transfer.is_none(),
+ "purge must drop the in-flight transfer session"
+ );
+ assert!(
+ partition.transfer_rearm.is_none(),
+ "purge must cancel the scheduled re-arm"
+ );
+ assert_eq!(
+ partition.consensus().state_transfer_stage(),
+ consensus::StateTransferStage::Idle,
+ "purge must release the transfer stage so a later trigger can arm"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
+
+ /// An offer whose frontier sits below this replica's own offset counter is
+ /// refused: installing it would rewind the counter, and the next
replicated
+ /// prepare is re-stamped from it, so this replica would persist different
+ /// bytes (and a different `batch_checksum`) than the rest of the group.
+ #[compio::test]
+ async fn
given_offer_below_local_counter_when_installed_should_refuse_rewind() {
+ let partition_dir = transfer_fence_dir("rewind-refused").await;
+ let mut partition = test_partition();
+ partition.set_partition_dir(partition_dir.clone());
+ partition.should_increment_offset = true;
+ partition.offset.store(99, Ordering::Release);
+
+ let behind = crate::state_transfer::ConsumerOffsetsWire {
+ purge_generation: 0,
+ next_offset: 50,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ };
+ let refused = partition
+ .install_state_transfer(&repair_config(), 12, Vec::new(),
&behind.encode())
+ .await;
+ assert!(
+ matches!(
+ refused,
+ Err(
+
crate::state_transfer::PartitionInstallError::OfferRewindsDurableData {
+ offer_next_offset: 50,
+ local_next_offset: 100,
+ }
+ )
+ ),
+ "expected a rewind refusal, got {refused:?}"
+ );
+
+ // A purge at the origin is the one legitimate rewind, and the artifact
+ // carries the generation that proves it: the same offer passes the
fence
+ // once its generation advances.
+ let purged = crate::state_transfer::ConsumerOffsetsWire {
+ purge_generation: 1,
+ next_offset: 0,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ };
+ let accepted = partition
+ .install_state_transfer(&repair_config(), 12, Vec::new(),
&purged.encode())
+ .await;
+ assert!(
+ !matches!(
+ accepted,
+
Err(crate::state_transfer::PartitionInstallError::OfferRewindsDurableData { ..
})
+ ),
+ "a purge-advancing offer must pass the rewind fence, got
{accepted:?}"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
+
+ /// Primary-by-index at view 0 with nothing committed refuses to serve: an
+ /// empty group is trivially "caught up", so this gate is the only thing
+ /// separating a real primary from a phantom whose directory vanished,
whose
+ /// zero-segment offer at frontier 0 would make a data-holding receiver
+ /// unlink its chain.
+ #[compio::test]
+ async fn given_nothing_committed_when_offer_requested_should_refuse() {
+ let partition_dir = transfer_fence_dir("nothing-committed").await;
+ let mut partition = test_partition();
+ partition.set_partition_dir(partition_dir.clone());
+ assert_eq!(partition.consensus().commit_max(), 0);
+
+ let refused = partition.state_transfer_offer(&repair_config()).await;
+ assert!(
+ matches!(
+ refused,
+
Err(crate::state_transfer::PartitionTransferUnavailable::NothingCommitted)
+ ),
+ "expected a NothingCommitted refusal, got {refused:?}"
+ );
+ assert!(
+ refused.is_err_and(|reason| reason.transient()),
+ "the refusal must be transient: the requester rotates rather than \
+ charging its failure count"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
}
#[cfg(test)]
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index a24dc6b79..01487f66b 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -682,7 +682,13 @@ where
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());
+ // At most one entry per in-window op can land, so the window bounds
the
+ // hint: sizing it for every resident header allocated (and memset)
about
+ // a megabyte of table per repair round on the floor-refusal path,
where
+ // the header vec grows with the live tail.
+ #[allow(clippy::cast_possible_truncation)]
+ let capacity = expected.min(headers.len() as u64) as usize;
+ let mut present: HashSet<u64> = HashSet::with_capacity(capacity);
let mut holds_messages = false;
for header in headers
.iter()
diff --git a/core/partitions/src/state_transfer.rs
b/core/partitions/src/state_transfer.rs
index d11ecd136..ec1cbd03f 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -140,14 +140,6 @@ impl TransferArtifact {
Self::Staged(_) => None,
}
}
-
- #[must_use]
- pub const fn complete(&self) -> bool {
- match self {
- Self::Pending(progress) => progress.complete(),
- Self::Staged(_) => true,
- }
- }
}
impl consensus::ChunkProgress for TransferArtifact {
@@ -184,6 +176,9 @@ impl consensus::ChunkProgress for TransferArtifact {
pub struct StagedSegmentMeta {
pub start_offset: u64,
pub end_offset: u64,
+ /// Byte length of the locally rebuilt sparse index sidecar, recorded at
+ /// the walk so the install does not re-stat the renamed file.
+ pub index_size: u64,
/// Payload byte length == the manifest entry's `len` == the final `.log`
/// file size.
pub size: u64,
@@ -681,9 +676,6 @@ pub(crate) fn walk_segment_payload(
base_offset: u64,
bytes: &[u8],
) -> Result<(SegmentWalkStats, Vec<u8>), SegmentWalkError> {
- if bytes.is_empty() {
- return Err(SegmentWalkError::Empty);
- }
let mut position = 0usize;
let mut next_offset = base_offset;
let mut stats: Option<SegmentWalkStats> = None;
@@ -940,12 +932,14 @@ 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).
+ /// The offer's offset frontier is below this replica's own offset
+ /// counter, so installing it would rewind the offset space: the next
+ /// replicated prepare would be re-stamped from the rewound counter and
+ /// persist different bytes (and a different `batch_checksum`) here than
+ /// on the rest of the group.
OfferRewindsDurableData {
offer_next_offset: u64,
- durable_end: u64,
+ local_next_offset: u64,
},
Offsets(ConsumerOffsetsWireError),
/// Duplicate base offset in the staged set.
@@ -995,11 +989,11 @@ impl fmt::Display for PartitionInstallError {
),
Self::OfferRewindsDurableData {
offer_next_offset,
- durable_end,
+ local_next_offset,
} => 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"
+ "offer frontier {offer_next_offset} is below this replica's
own next offset \
+ {local_next_offset}; installing it would rewind the offset
space"
),
Self::Offsets(source) => write!(f, "consumer-offsets artifact
rejected: {source}"),
Self::DuplicateSegment { start_offset } => {
@@ -1101,7 +1095,16 @@ pub async fn quarantine_segment_files(partition_dir:
&str) -> std::io::Result<St
};
compio::fs::rename(&path, &PathBuf::from(&target).join(name)).await?;
}
+ // All three touched directories: the target (its new dirents), the source
+ // (the removals), and the source's parent (the target directory itself is
a
+ // new dirent there). Without the target-side syncs a crash can leave the
+ // moved files linked in neither directory -- only forensics are at stake,
+ // but forensics are the whole point of the copies.
+ fsync_dir(&target).await?;
fsync_dir(partition_dir).await?;
+ if let Some(parent) =
Path::new(partition_dir).parent().and_then(Path::to_str) {
+ fsync_dir(parent).await?;
+ }
Ok(target)
}
@@ -1153,7 +1156,7 @@ struct PlannedOffsetWrite {
/// 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.
-async fn fsync_dir(partition_dir: &str) -> std::io::Result<()> {
+pub(crate) async fn fsync_dir(partition_dir: &str) -> std::io::Result<()> {
compio::fs::File::open(partition_dir)
.await?
.sync_all()
@@ -1199,8 +1202,17 @@ where
// `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.
+ //
+ // A RESTARTED replica holding a full chain matches this shape too
+ // (`commit_max == 0` because the partition journal is memory-only,
+ // `installed_frontier == None` for a recovered non-empty chain). That
+ // is the load-bearing reason this refusal is safe rather than a
+ // wedge: every transfer-arm site presupposes a peer that already
+ // reported commit > 0 (repair floor refusals, StartView adoption), so
+ // nobody ever asks a cluster where everything still reports 0.
+ // Extending the gate with `recovered_durable_offset.is_some()` would
+ // be WRONG: such an offer carries `commit_op = 0`, so the receiver's
+ // floor becomes a no-op while its counter jumps to the frontier.
if self.consensus().commit_max() == 0 &&
self.installed_frontier.is_none() {
return Err(PartitionTransferUnavailable::NothingCommitted);
}
@@ -1370,7 +1382,7 @@ where
SegmentChecksumMemo::new()
}
};
- hash_segment_range(log_path, memo.hashed_len, size, &mut memo.hasher)
+ hash_segment_range(log_path, memo.hashed_len, size, &mut memo.hasher,
None)
.await
.map_err(|source| PartitionTransferUnavailable::SegmentUnreadable {
start_offset,
@@ -1385,27 +1397,10 @@ where
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).
+ /// [`quarantine_segment_files`] over this partition's directory, for the
+ /// shard's `ConvergeFailed` fence -- the safety argument (segment files
+ /// move, superblock slots STAY, copies are unreclaimed operator evidence)
+ /// lives on the free function. `None` for an in-memory partition.
///
/// # Errors
/// The underlying `std::io::Error`; the caller logs and lets the rebuild
@@ -1506,6 +1501,7 @@ where
let (stats, index_bytes) =
walk_segment_payload(entry.frontier,
&bytes).map_err(SpillError::Walk)?;
let (log_staging, index_staging) = staging_paths(&partition_dir,
entry.frontier);
+ let index_size = index_bytes.len() as u64;
// Two writes, not a loop: each moves its buffer into compio's
// owned-buffer API, so a segment-sized payload is never copied.
write_staging_file(&log_staging, bytes)
@@ -1530,6 +1526,7 @@ where
start_offset: entry.frontier,
end_offset: stats.end_offset,
size: entry.len,
+ index_size,
start_timestamp: stats.start_timestamp,
end_timestamp: stats.end_timestamp,
max_timestamp: stats.max_timestamp,
@@ -1546,6 +1543,9 @@ where
/// time) is exactly the work reuse exists to skip. The index write
/// stays: the scan never checks `.index.staging`, and a missing sidecar
/// would hand the install a missing rename source.
+ ///
+ /// No directory fsync here: every sidecar lands in the same directory, so
+ /// the caller fsyncs ONCE after its loop instead of once per adoption.
async fn adopt_staged_segment(
&self,
entry: &consensus::StateArtifact,
@@ -1557,22 +1557,18 @@ where
let (stats, index_bytes) =
walk_segment_payload(entry.frontier,
bytes).map_err(SpillError::Walk)?;
let (log_staging, index_staging) = staging_paths(&partition_dir,
entry.frontier);
+ let index_size = index_bytes.len() as u64;
write_staging_file(&index_staging, index_bytes)
.await
.map_err(|source| SpillError::StagingIo {
path: index_staging.clone(),
source,
})?;
- fsync_dir(&partition_dir)
- .await
- .map_err(|source| SpillError::StagingIo {
- path: PathBuf::from(&partition_dir),
- source,
- })?;
Ok(StagedSegmentMeta {
start_offset: entry.frontier,
end_offset: stats.end_offset,
size: entry.len,
+ index_size,
start_timestamp: stats.start_timestamp,
end_timestamp: stats.end_timestamp,
max_timestamp: stats.max_timestamp,
@@ -1655,6 +1651,15 @@ where
adopted.push((index as u32, meta));
}
}
+ // Every rebuilt sidecar landed in the same directory, so one fsync
+ // covers them all. Its failure discards EVERY adoption: the per-adopt
+ // filter above no longer sees a durability failure, and an undurable
+ // sidecar handed to the install is a missing rename source after a
+ // crash.
+ if !adopted.is_empty() && fsync_dir(&partition_dir).await.is_err() {
+ adopted.clear();
+ matched_paths.clear();
+ }
// 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;
@@ -1702,24 +1707,33 @@ where
});
}
let offsets_wire = ConsumerOffsetsWire::decode(offsets_bytes)?;
- // Anti-rewind against LOCAL DURABLE BYTES, not just the commit
+ // Anti-rewind against the LOCAL OFFSET COUNTER, not 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.
+ // rejoin. The counter is the one signal that is `Some`-equivalent in
+ // EVERY state the offset space has advanced through (recovered bytes,
+ // an installed frontier, a converge after a failed install --
+ // `recovered_durable_offset` is `None` in the last two), and it is
+ // precisely what a rewind corrupts: received prepares are pre-stamp,
+ // `stamp_prepare_for_persistence` overwrites `base_offset` from this
+ // counter and recomputes `batch_checksum` over it, so a rewound
+ // counter persists different bytes and a different checksum on this
+ // replica than on 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
+ let local_next_offset = if self.should_increment_offset {
+ self.offset.load(Ordering::Acquire) + 1
+ } else {
+ 0
+ };
+ if !purge_advances && local_next_offset > 0 &&
offsets_wire.next_offset < local_next_offset
{
return Err(PartitionInstallError::OfferRewindsDurableData {
offer_next_offset: offsets_wire.next_offset,
- durable_end,
+ local_next_offset,
});
}
staged.sort_unstable_by_key(|meta| meta.start_offset);
@@ -1789,7 +1803,7 @@ where
// Sweep staging strays a dead earlier attempt left behind, keeping
// only what THIS install is about to rename. Bounded disk hygiene;
// the reuse-scan sweeps too, and boot sweeps ALL of `.staging`
- // (`segment_recovery::sweep_staging_files`), so a transfer abandoned
+ // (`segment_recovery::sweep_scratch_files`), so a transfer abandoned
// for good leaks at most until the next restart.
let keep: Vec<&Path> = staged
.iter()
@@ -1857,6 +1871,19 @@ where
// contiguous partition and re-triggers transfer for the rest. Not
// fewer fsyncs than this: one-per-pair would lean on intra-directory
// rename ordering POSIX does not grant.
+ //
+ // KNOWN WINDOW, above and here: the old chain's unlinks are already
+ // durable and no staged log has landed yet, and nothing durable names
+ // the offset frontier in between -- a crash there boots to zero
+ // segments and counter 0. Bounded, not silent: the gap check drops
+ // live prepares at sequencer 0 and the repair floor refuses a `None`
+ // stand-in against a nonzero first batch, so the replica takes a clean
+ // full re-transfer instead of serving a hole. One narrow door stays
+ // open until the frontier gets a durable home (the partition
+ // superblock already reserves a field for it):
+ // `repaired_window_is_offsets_only` can accept a complete
+ // offsets-only window with the counter still at 0, after which the
+ // next live append stamps `base_offset` 0 against the group's N.
for meta in &staged {
let (_, index_final) = final_paths(partition_dir,
meta.start_offset);
compio::fs::rename(&meta.index_staging, &index_final)
@@ -1894,23 +1921,32 @@ where
// hydrate pattern; earlier segments are sealed and never written).
for meta in &staged {
let (log_final, index_final) = final_paths(partition_dir,
meta.start_offset);
- let index_len = compio::fs::metadata(&index_final)
- .await
- .map_or(0, |metadata| metadata.len());
- let storage = SegmentStorage::new(
- &log_final,
- &index_final,
- meta.size,
- index_len,
- config.enforce_fsync,
- config.enforce_fsync,
- true,
- )
- .await
- .map_err(|source| PartitionInstallError::SegmentOpen {
- path: log_final.clone(),
- source,
- })?;
+ // Retried once: by this point every rename already landed, so a
+ // failed open converges away a chain that is COMPLETE AND DURABLE
+ // on disk and the re-pull transfers the whole thing again. The
+ // sweep itself is right (a chain the live state does not know
+ // about would resurrect at boot), so one retry against a
+ // transient open failure is the only cheap save available.
+ let open = || {
+ SegmentStorage::new(
+ &log_final,
+ &index_final,
+ meta.size,
+ meta.index_size,
+ config.enforce_fsync,
+ config.enforce_fsync,
+ true,
+ )
+ };
+ let storage = match open().await {
+ Ok(storage) => storage,
+ Err(_) => open()
+ .await
+ .map_err(|source| PartitionInstallError::SegmentOpen {
+ path: log_final.clone(),
+ source,
+ })?,
+ };
let mut segment = Segment::new(meta.start_offset,
config.segment_size);
segment.sealed = true;
segment.start_timestamp = meta.start_timestamp;
@@ -2032,7 +2068,12 @@ where
self.pending_consumer_offset_commits.clear();
self.last_polled_offsets.pin().clear();
- let clamp = |offset: u64| offset.min(next_offset.saturating_sub(1));
+ // `None` when the group's offset space is empty (`next_offset == 0`,
+ // a purged origin): clamping every transferred offset to 0 would tell
+ // each consumer it consumed offset 0 on a partition that never minted
+ // one, so a `Next` poll skips the first message. Dropping the entries
+ // is what "no offsets yet" means.
+ let clamp = |offset: u64| next_offset.checked_sub(1).map(|last|
offset.min(last));
if self.consumer_offsets_path.is_none() ||
self.consumer_group_offsets_path.is_none() {
// Nothing to write the transferred table into: unreachable via
// the server boot paths (they always configure storage), but if
@@ -2048,7 +2089,9 @@ where
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);
+ let Some(value) = clamp(*offset) else {
+ continue;
+ };
let entry = ConsumerOffset::default_for_consumer(*id, &dir);
entry.offset.store(value, Ordering::Release);
let path = entry.path.clone();
@@ -2063,7 +2106,9 @@ where
}
if let Some(dir) = self.consumer_group_offsets_path.clone() {
for (id, offset) in &offsets_wire.groups {
- let value = clamp(*offset);
+ let Some(value) = clamp(*offset) else {
+ continue;
+ };
let group_id = ConsumerGroupId(*id as usize);
let entry =
ConsumerOffset::default_for_consumer_group(group_id, &dir);
entry.offset.store(value, Ordering::Release);
@@ -2131,7 +2176,13 @@ where
// durable bytes on disk. Deliberately NOT `recovered_durable_offset`:
// that field also gates repaired-batch persistence, and overstating
// it would silently drop the `(commit_op, commit_max]` replay window.
- self.installed_frontier = Some(next_offset);
+ // `Some(0)` is filtered out: it reads as a real claim in the
+ // `NothingCommitted` serve gate (`installed_frontier.is_none()`), so a
+ // replica holding zero bytes at frontier 0 would start serving empty
+ // offers -- the phantom shape that gate exists to stop. Behavior is
+ // otherwise unchanged: the repair floor stand-in already treats
+ // `Some(0)` and `None` identically.
+ self.installed_frontier = (next_offset > 0).then_some(next_offset);
self.stats.zero_out_all();
#[allow(clippy::cast_possible_truncation)]
self.stats
@@ -2156,9 +2207,21 @@ where
if commit_op > consensus.commit_min() {
consensus.set_commit_floor(commit_op);
}
- if commit_op > consensus.sequencer().current_sequence() {
- consensus.sequencer().set_sequence(commit_op);
- }
+ // The sequencer is SET, not raised: the install cleared the whole
+ // journal, so ops in `(commit_op, old_sequencer]` -- journaled and
+ // PrepareOk'd before the transfer armed, since a transferring replica
+ // withholds acks -- are ops consensus still claims and the journal can
+ // no longer serve. The primary's retransmit dies in the backup gap
+ // check and repair will not arm (the install leaves
+ // `commit_min == commit_max`), so a DVC from here would advertise an
op
+ // this replica cannot walk. The pipeline is cleared in the same
breath:
+ // its entries are backed by the same erased journal, and
+ // `LocalPipeline::push` asserts op sequentiality in release, so a bare
+ // rewind would turn the silent desync into a shard panic on a replica
+ // promoted mid-transfer. (`last_prepare_checksum` needs nothing: it is
+ // only read as a `parent:` stamp when building a prepare.)
+ consensus.sequencer().set_sequence(commit_op);
+ consensus.pipeline().borrow_mut().clear();
consensus.advance_commit_max(commit_op);
self.observed_view = self.consensus().view();
self.repair = None;
@@ -2169,14 +2232,7 @@ where
offsets_written,
})
}
-}
-/// See `install_state_transfer`'s failure arm.
-impl<B, SB> IggyPartition<B, SB>
-where
- B: MessageBus,
- SB: SuperblockStore,
-{
/// Converge the live partition AND its directory to an empty,
/// honestly-lagging shape after a failed install: no segment files at
/// all (the failure can land anywhere from "old chain unlinked" to
@@ -2220,7 +2276,7 @@ where
.map(|entry| entry.path())
.filter(|path| {
path.to_str().is_some_and(|path| {
- [".log", ".index", ".staging"]
+ [".log", ".index", STAGING_SUFFIX]
.iter()
.any(|extension| path.ends_with(extension))
})
@@ -2275,7 +2331,8 @@ where
// 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.installed_frontier =
+ (staged_was_empty && minted_next_offset >
0).then_some(minted_next_offset);
self.stats.zero_out_all();
self.stats.increment_segments_count(1);
self.repair = None;
@@ -2325,14 +2382,19 @@ impl std::error::Error for SpillError {}
const OFFER_HASH_CHUNK_LEN: usize = 1 << 20;
/// 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 `to`: the segment accounts bytes the disk does
-/// not hold.
+/// [`OFFER_HASH_CHUNK_LEN`] chunks with one reactor yield per chunk, appending
+/// each chunk to `sink` when one is given.
+///
+/// The single chunked reader for both passes over a segment file: the offer
+/// build's checksum extension (no sink) and the serving side's load +
re-verify
+/// (sink collects the artifact). Errors on a file shorter than `to`: the
segment
+/// accounts bytes the disk does not hold.
async fn hash_segment_range(
path: &str,
from: u64,
to: u64,
hasher: &mut consensus::state_manifest::StateArtifactHasher,
+ mut sink: Option<&mut Vec<u8>>,
) -> std::io::Result<()> {
if from >= to {
return Ok(());
@@ -2360,13 +2422,16 @@ async fn hash_segment_range(
))
})?;
hasher.update(&buf);
+ if let Some(sink) = sink.as_deref_mut() {
+ sink.extend_from_slice(&buf);
+ }
position += want as u64;
}
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
+/// against the manifest entry, chunked through [`hash_segment_range`] with one
/// reactor yield per chunk.
///
/// The serving side runs this on the pump to answer a single chunk request, so
@@ -2383,23 +2448,12 @@ 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;
- }
+ hash_segment_range(log_path, 0, entry.len, &mut hasher, Some(&mut bytes))
+ .await
+ .ok()?;
(hasher.finish() == entry.checksum).then_some(bytes)
}
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 8c2bd8861..af72c7d8e 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -2328,7 +2328,19 @@ async fn load_partition(
let current_offset = sized_end.or_else(|| empty_frontier.map(|start| start
- 1));
partition.created_at = partition_metadata.created_at;
partition.recovered_durable_offset = sized_end;
- partition.installed_frontier = empty_frontier;
+ // The OFFSET COUNTER is restored from that file name (above), but the
+ // `installed_frontier` CLAIM deliberately is not: the claim says
"everything
+ // below me is represented here", and
`converge_to_empty_after_failed_install`
+ // refuses to make it when staged segments were dropped -- yet a converge
+ // plants exactly the same empty `{frontier:020}.log` a legitimate empty
+ // install does, so boot provably cannot tell them apart. Re-deriving it
here
+ // would hand the refused claim back: the repair floor stand-in would
accept a
+ // commit floor over ops this replica holds zero bytes for, and the replica
+ // would pass the serve gate and offer that emptiness onward, making a peer
+ // unlink its own chain. Leaving it `None` costs one spurious full
+ // re-transfer on the legitimate empty-install restart; a false caught-up
+ // claim is not recoverable. A durable home for the frontier (the partition
+ // superblock already reserves a field) is what would settle it properly.
let counter = current_offset.unwrap_or(0);
partition.offset.store(counter, Ordering::Release);
partition.dirty_offset.store(counter, Ordering::Relaxed);
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 6eee83032..428d145ca 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -28,9 +28,10 @@ pub use router::CONSENSUS_TICK_INTERVAL;
#[cfg(any(test, feature = "simulator"))]
use consensus::LocalPipeline;
use consensus::{
- CommitOutcome, Consensus, ConsensusClock, MetadataHandle, MuxPlane,
PartitionsHandle, Pipeline,
- Plane, PlaneKind, STATE_TRANSFER_MAX_DECODE_RETRIES,
STATE_TRANSFER_MAX_STALL_RETRIES,
- Sequencer, VsrAction, VsrConsensus, build_deny_reply_from_request_header,
+ ChunkProgress, CommitOutcome, Consensus, ConsensusClock, MetadataHandle,
MuxPlane,
+ PartitionsHandle, Pipeline, Plane, PlaneKind,
STATE_TRANSFER_MAX_DECODE_RETRIES,
+ STATE_TRANSFER_MAX_STALL_RETRIES, Sequencer, VsrAction, VsrConsensus,
+ build_deny_reply_from_request_header,
};
#[cfg(any(test, feature = "simulator"))]
use crossfire::AsyncRxTrait;
@@ -846,6 +847,19 @@ enum ServedOffer {
Partition(Rc<partitions::state_transfer::PartitionStateTransferOffer>),
}
+/// Largest `segment.size` any configuration can set, mirroring
+/// `configs::server_config::validators::SEGMENT_MAX_SIZE_BYTES` (the `configs`
+/// crate is not a dependency here). Both the served-payload budget and the
+/// per-artifact alloc cap are derived from it rather than hand-tuned.
+const SEGMENT_SIZE_CEILING_BYTES: u64 = 1 << 30;
+
+/// The most one segment can overshoot its size cap: rotation checks the cap
+/// AFTER appending, so a segment closes at most one maximum-size batch past
it.
+/// A batch is bounded by the per-message payload ceiling plus its 256-byte
+/// command header; anything larger is refused at ingest.
+const SEGMENT_SIZE_OVERSHOOT_BYTES: u64 = iggy_common::MAX_PAYLOAD_SIZE as u64
+ + server_common::send_messages2::COMMAND_HEADER_SIZE as u64;
+
/// Shard-wide cache of segment payloads loaded to serve partition chunks,
/// content-addressed by `(namespace, manifest checksum)` so every requester
/// pulling the same offer generation shares ONE resident copy (per-requester
@@ -878,7 +892,21 @@ impl ServedSegmentCache {
/// 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;
+ ///
+ /// Sized for CONCURRENT pulls, not one: at exactly one max-size segment
+ /// (`segment.size` defaults to and is capped at 1 GiB) a single receiver
+ /// arming its `PARTITION_TRANSFERS_INFLIGHT_MAX` transfers thrashes the
+ /// cache by itself -- distinct partitions are distinct keys, so the pulls
+ /// evict each other on every chunk, and each miss re-reads and re-hashes a
+ /// whole segment to serve one 256 KiB chunk. That is the 4096:1 read
+ /// amplification this cache exists to prevent, plus an offer eviction per
+ /// failed re-verify feeding the hard-failure backoff.
+ const RESIDENT_BYTES_MAX: u64 = SEGMENT_SIZE_CEILING_BYTES *
Self::CONCURRENT_SERVED_SEGMENTS;
+
+ /// Distinct max-size segments the budget holds at once. Matches the
+ /// receiver-side in-flight cap, since that is how many distinct segments
one
+ /// requester can pull concurrently.
+ const CONCURRENT_SERVED_SEGMENTS: u64 = 4;
/// Sweeps a payload survives without serving a chunk.
///
@@ -924,6 +952,14 @@ impl ServedSegmentCache {
fn insert(&mut self, namespace: u64, checksum: u64, payload: Rc<Vec<u8>>) {
let incoming = payload.len() as u64;
+ // Credited BEFORE the eviction scan: re-inserting an existing key
frees
+ // its own slot, and charging that only afterwards evicted neighbours
to
+ // make room for bytes that were about to be released.
+ if let Some(replaced) = self.entries.remove(&(namespace, checksum)) {
+ self.resident_bytes = self
+ .resident_bytes
+ .saturating_sub(replaced.payload.len() as u64);
+ }
while self.resident_bytes.saturating_add(incoming) >
Self::RESIDENT_BYTES_MAX
&& !self.entries.is_empty()
{
@@ -941,18 +977,16 @@ impl ServedSegmentCache {
}
}
self.use_seq += 1;
- if let Some(replaced) = self.entries.insert(
+ // The key was removed above, so this never replaces an entry whose
bytes
+ // still need crediting back.
+ self.entries.insert(
(namespace, checksum),
CachedSegmentPayload {
payload,
last_use: self.use_seq,
last_use_sweep: self.sweeps,
},
- ) {
- self.resident_bytes = self
- .resident_bytes
- .saturating_sub(replaced.payload.len() as u64);
- }
+ );
self.resident_bytes = self.resident_bytes.saturating_add(incoming);
}
}
@@ -979,7 +1013,10 @@ struct ServedStateTransfer {
/// 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.
+/// peer that is momentarily behind from one that knows less than it does. They
+/// are CONSTRUCTOR arguments rather than an optional builder step: as an
+/// optional step every one of the eight construction sites had to remember it,
+/// and two did not.
struct TransferDescriptor<'a> {
/// `Some((manifest, commit_op))` when the peer can serve.
offer: Option<(&'a [consensus::StateArtifact], u64)>,
@@ -992,29 +1029,28 @@ struct TransferDescriptor<'a> {
}
impl<'a> TransferDescriptor<'a> {
- const fn available(offer: &'a [consensus::StateArtifact], commit_op: u64)
-> Self {
+ const fn available(
+ offer: &'a [consensus::StateArtifact],
+ commit_op: u64,
+ view: u32,
+ commit_max: u64,
+ ) -> Self {
Self {
offer: Some((offer, commit_op)),
- view: 0,
- commit_max: 0,
+ view,
+ commit_max,
transient: false,
}
}
- const fn unavailable(transient: bool) -> Self {
+ const fn unavailable(transient: bool, view: u32, commit_max: u64) -> Self {
Self {
offer: None,
- view: 0,
- commit_max: 0,
+ view,
+ commit_max,
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
@@ -1818,6 +1854,25 @@ where
}
}
+/// The serving replica's `(view, commit_max)` for a descriptor.
+///
+/// Sampled per branch, always AFTER any offer build: the build force-flushes
and
+/// hashes every un-memoized segment (seconds on a first multi-GiB serve) while
+/// reading its `commit_op` post-flush, so a pre-build sample could advertise a
+/// `commit_max` below the descriptor's own `commit_op`. Harmless on the
receiver
+/// (the values are only compared against its own locals) but it makes its gate
+/// refuse, and refusals feed a backoff.
+const fn serving_progress<B, SB>(partition: &IggyPartition<B, SB>) -> (u32,
u64)
+where
+ B: MessageBus,
+ SB: SuperblockStore,
+{
+ (
+ partition.consensus().view(),
+ partition.consensus().commit_max(),
+ )
+}
+
/// 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
@@ -3342,36 +3397,11 @@ where
}
// Same gap-fill as the metadata arm: a journal-less rejoiner that
// adopted the new view still lacks the window's entries; repair from
- // the announcing primary, floor settled by its RangeEvicted.
- let consensus = partition.consensus();
- if consensus.is_normal()
- && consensus.commit_min() < consensus.commit_max()
- && partition.repair.is_none()
- {
- let nonce = iggy_common::random_id::get_uuid();
- let to_op = consensus.commit_max();
- let from_op = consensus.commit_min() + 1;
- let cluster = consensus.cluster();
- let self_id = consensus.replica();
- partition.repair = Some(partitions::RepairSession {
- nonce,
- to_op,
- floor: None,
- peer: header.replica,
- first_batch_offset: None,
- idle_ticks: 0,
- });
- self.send_request_prepares(
- cluster,
- self_id,
- header.replica,
- nonce,
- from_op,
- to_op,
- header.namespace,
- )
+ // the announcing primary, floor settled by its RangeEvicted. The
shared
+ // helper carries one guard more than this site needs
(`is_transferring`,
+ // already covered by the early return above) and logs the arm.
+ self.maybe_request_partition_repair(partition, header.replica)
.await;
- }
}
#[allow(clippy::future_not_send)]
@@ -3736,19 +3766,17 @@ where
// 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));
+ // Keyed on the OPERATION, not on whether partition 0/0/0 exists: raw
+ // namespace 0 is `IggyNamespace::new(0, 0, 0)` and ids slab-allocate
+ // from 0, so 0/0/0 is the first partition every cluster creates -- on
a
+ // single-shard node a "no partition 0 materialised" conjunct goes
false
+ // the moment one topic exists and disables this migration exactly
where
+ // it is needed. `is_metadata_plane` is the plane's OWN applicability
+ // predicate (the session ops `Register`/`Logout` replicate here
without
+ // being metadata mutations, so `is_metadata` alone is too narrow),
which
+ // is why both sites share it rather than re-deriving the set.
+ let metadata_plane_op = header.operation.is_metadata_plane();
+ let legacy_metadata_claim = header.namespace == 0 && metadata_plane_op;
if let Some(ref consensus) = planes.0.consensus
&& (consensus.namespace() == header.namespace ||
legacy_metadata_claim)
{
@@ -3791,6 +3819,25 @@ where
consensus.set_last_prepare_checksum(header.checksum);
return;
}
+ // A metadata-plane op that did not match above (no metadata consensus
on
+ // this shard, or a namespace neither plane claims) is DROPPED, never
+ // offered to the partition arm. Falling through let a metadata prepare
+ // reach `apply_repaired_prepare`: it journals nothing, but it resets
the
+ // partition repair session's idle ticks (masking a genuine stall) and
+ // carries the metadata prepare's checksum into the partition consensus
+ // via `set_last_prepare_checksum` -- inert only while prepare
checksums
+ // are structurally zero, and a cross-plane parent stamp the moment the
+ // checksum chain is activated (see the note in `consensus::impls`).
+ if metadata_plane_op {
+ tracing::debug!(
+ shard = self.id,
+ op = header.op,
+ operation = ?header.operation,
+ namespace_raw = header.namespace,
+ "dropping a metadata-plane repair prepare this shard cannot
journal"
+ );
+ return;
+ }
let Some(partition) = planes
.1
.0
@@ -4291,7 +4338,7 @@ where
/// Serve one `RequestStateTransfer`: build a fresh offer (or refuse),
/// cache it for the chunk pulls, and answer with the descriptor.
- #[allow(clippy::future_not_send)]
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
async fn on_request_state_transfer(&self, msg:
&Message<RequestStateTransferHeader>)
where
B: MessageBus,
@@ -4354,8 +4401,12 @@ where
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::available(&offer.manifest(),
offer.commit_op)
- .serving(consensus.view(), consensus.commit_max()),
+ TransferDescriptor::available(
+ &offer.manifest(),
+ offer.commit_op,
+ consensus.view(),
+ consensus.commit_max(),
+ ),
)
.await;
return;
@@ -4378,8 +4429,12 @@ where
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::available(&offer.manifest(),
offer.commit_op)
- .serving(consensus.view(), consensus.commit_max()),
+ TransferDescriptor::available(
+ &offer.manifest(),
+ offer.commit_op,
+ consensus.view(),
+ consensus.commit_max(),
+ ),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
@@ -4409,8 +4464,11 @@ where
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::unavailable(false)
- .serving(consensus.view(), consensus.commit_max()),
+ TransferDescriptor::unavailable(
+ false,
+ consensus.view(),
+ consensus.commit_max(),
+ ),
)
.await;
}
@@ -4835,7 +4893,10 @@ where
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::unavailable(false),
+ // Transient for the same reason as the partition arm: an
+ // offer that aged out between chunks is not a peer
failure,
+ // and the restarted session converges.
+ TransferDescriptor::unavailable(true, consensus.view(),
consensus.commit_max()),
)
.await;
}
@@ -5431,9 +5492,6 @@ 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
// the receiver chunks from a different offer than the manifest it
@@ -5458,14 +5516,19 @@ where
requester = header.replica,
"re-answering a partition state transfer request from the
offer already served"
);
+ let (serving_view, serving_commit_max) =
serving_progress(partition);
self.send_state_transfer_target(
cluster,
self_id,
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::available(&offer.manifest(),
offer.commit_op)
- .serving(serving_view, serving_commit_max),
+ TransferDescriptor::available(
+ &offer.manifest(),
+ offer.commit_op,
+ serving_view,
+ serving_commit_max,
+ ),
)
.await;
return;
@@ -5482,14 +5545,19 @@ where
total_len = offer.total_len(),
"serving partition state transfer"
);
+ let (serving_view, serving_commit_max) =
serving_progress(partition);
self.send_state_transfer_target(
cluster,
self_id,
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::available(&offer.manifest(),
offer.commit_op)
- .serving(serving_view, serving_commit_max),
+ TransferDescriptor::available(
+ &offer.manifest(),
+ offer.commit_op,
+ serving_view,
+ serving_commit_max,
+ ),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
@@ -5517,14 +5585,14 @@ where
%reason,
"cannot serve partition state transfer; requester falls
back"
);
+ let (serving_view, serving_commit_max) =
serving_progress(partition);
self.send_state_transfer_target(
cluster,
self_id,
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::unavailable(transient)
- .serving(serving_view, serving_commit_max),
+ TransferDescriptor::unavailable(transient, serving_view,
serving_commit_max),
)
.await;
}
@@ -5573,6 +5641,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();
let chunk_len_max = self.state_chunk_len_max();
let reply = loop {
let attempt = 'attempt: {
@@ -5717,7 +5787,14 @@ where
header.replica,
header.nonce,
header.namespace,
- TransferDescriptor::unavailable(false),
+ // TRANSIENT: routine on a busy primary -- retention GC'd a
+ // served segment, or the offer simply aged out of
+ // `state_transfer_offers` between two chunks. The retry
+ // converges either way (the restarted session reflects the
+ // current segment set), so charging the
consecutive-failure
+ // count would double the backoff up to 1024x for an event
+ // that is not a failure.
+ TransferDescriptor::unavailable(true, serving_view,
serving_commit_max),
)
.await;
}
@@ -5734,10 +5811,16 @@ where
}
}
- /// Alloc cap per PARTITION artifact. Segments are hard-capped at 1 GiB
- /// with a soft-cap overshoot of one batch, so the metadata plane's 1 GiB
- /// cap would deterministically reject legal segments.
- const PARTITION_ARTIFACT_LEN_MAX: u64 = 2 << 30;
+ /// Alloc cap per PARTITION artifact: the configured segment ceiling plus
the
+ /// one maximum-size batch a segment may overshoot it by (rotation checks
the
+ /// cap after appending). The metadata plane's flat 1 GiB cap would
+ /// deterministically reject a legal overshooting segment, and the previous
+ /// 2 GiB left the receiver holding twice the largest legal artifact --
+ /// `mem::take` moves the buffer out of the session, not out of memory, so
it
+ /// stays resident through verify + walk + staging write, times the
in-flight
+ /// cap, times the shard count.
+ const PARTITION_ARTIFACT_LEN_MAX: u64 =
+ SEGMENT_SIZE_CEILING_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES;
/// Sanity cap across a partition manifest. Segment artifacts spill to
/// disk as they complete, so this bounds corruption, not memory.
@@ -6272,7 +6355,7 @@ where
let Some(session) = partition.transfer.as_ref() else {
return;
};
- let all_done =
session.artifacts.iter().all(TransferArtifact::complete);
+ let all_done = session.artifacts.iter().all(ChunkProgress::complete);
if !all_done {
self.request_pending_partition_chunk(namespace).await;
return;
