You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/09/01 14:21:08 UTC
(iggy) branch durable-offset-watermark updated: fix: address review comments.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch durable-offset-watermark
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/durable-offset-watermark by this push:
new 192daa51d fix: address review comments.
192daa51d is described below
commit 192daa51d0169822c3465750fda3c60ebb616ed0
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Sep 1 19:48:31 2026 +0530
fix: address review comments.
---
core/configs/src/server_config/partition.rs | 3 +
core/consensus/src/vsr_state.rs | 125 ++-
.../tests/cluster/crash_offset_reuse.rs | 145 +++-
core/journal/src/superblock.rs | 4 +-
core/partitions/src/iggy_partition.rs | 963 +++++++++++++++------
core/partitions/src/lib.rs | 1 +
core/partitions/src/log.rs | 5 +-
core/partitions/src/segment_anchor.rs | 205 +++++
core/partitions/src/state_transfer.rs | 15 +-
core/server/config.toml | 4 +
core/server/src/bootstrap.rs | 50 +-
core/server/src/partition_helpers.rs | 2 +-
core/server/src/segment_recovery.rs | 342 +++++++-
core/shard/src/lib.rs | 130 ++-
14 files changed, 1556 insertions(+), 438 deletions(-)
diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs
index 38a1d6e5b..b94d40e13 100644
--- a/core/configs/src/server_config/partition.rs
+++ b/core/configs/src/server_config/partition.rs
@@ -132,6 +132,9 @@ pub struct PartitionConfig {
/// One superblock write per block: lowering it raises the fsync rate,
/// raising it wastes at most one block per crash. Must be > 0 and <=
/// [`MAX_OFFSET_RESERVATION_LEASE`].
+ ///
+ /// SINGLE-REPLICA groups only: a replicated group acks once a quorum has
+ /// journaled the batch, claims nothing, and ignores this.
pub offset_reservation_lease: u32,
/// Entries the evicted ring retains per multi-replica partition for
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
index f01978fcb..40631d337 100644
--- a/core/consensus/src/vsr_state.rs
+++ b/core/consensus/src/vsr_state.rs
@@ -32,19 +32,25 @@ use std::fmt;
/// `replica_id`(1) + `replica_count`(1) + `view`(4) + `log_view`(4) +
/// `commit_max`(8) + `checkpoint_op`(8) + `checkpoint_checksum`(16) +
/// `offset_frontier`(8) + `offset_reserved`(8).
+///
+/// Growing this is one-way: a record of this length is [`VsrStateError::WrongLength`]
+/// to every build that predates the field, so a ROLLBACK needs the data directory
+/// wiped even though the upgrade does not.
pub const ENCODED_LEN: usize = 74;
/// The layout before `offset_frontier` was appended.
///
-/// [`VsrState::try_from`] still accepts records of this length and zero-fills
-/// the new field. Without it every superblock already on disk -- the metadata
-/// plane writes one on every view change and checkpoint, single-node included --
-/// would decode as [`VsrStateError::WrongLength`] and refuse boot as a
-/// durability violation. A version bump instead of this would not help on its
-/// own: `classify` compares the version for exact equality, so a v2 build turns
-/// every v1 record into `Unreadable`, which is the same refusal wearing a
-/// different name.
-pub const ENCODED_LEN_WITHOUT_FRONTIER: usize = 58;
+/// [`VsrState::try_from`] accepts records of this length. Without it every
+/// superblock already on disk -- the metadata plane writes one on every view
+/// change and checkpoint, single-node included -- decodes as
+/// [`VsrStateError::WrongLength`], which the metadata plane treats as a
+/// durability violation and refuses the whole node's boot on.
+///
+/// A version bump instead of this would not help on its own: `classify` compares
+/// the version for exact equality, so a v2 build turns every v1 record into
+/// `Unreadable`, which is the same refusal wearing a different name -- and it
+/// would make this tolerance unreachable, refusing even records that decode.
+pub const ENCODED_LEN_WITHOUT_RESERVATION: usize = 66;
/// The durable consensus state of one replica for one consensus group.
///
@@ -136,20 +142,26 @@ impl TryFrom<&[u8]> for VsrState {
type Error = VsrStateError;
fn try_from(bytes: &[u8]) -> Result<Self, Self::Error> {
- // Length-tolerant for ONE legacy layout: a pre-`offset_frontier` record
- // pads out and its trailing fields read as 0, which the read sites
- // filter. The one length check then puts every field slice below in
- // bounds by construction, so the `try_into`s cannot fail.
+ // Length-tolerant for ONE legacy layout, the pre-`offset_reserved` record
+ // every tagged release wrote. The one length check then puts every field
+ // slice below in bounds by construction, so the `try_into`s cannot fail.
+ //
+ // `offset_reserved` is filled from `offset_frontier`, not zeroed. The two
+ // agree on a record written before the reservation existed: the frontier
+ // is what that build's data proved, and the write side clamps the
+ // reservation up to it anyway, so this is the same value a first write
+ // under this build would record. A 0 would instead claim "nothing
+ // reserved" for offsets the frontier says exist.
//
- // `offset_reserved` gets no tolerance of its own: a silent 0 reads as
- // "nothing reserved" and re-mints confirmed offsets, the exact defect
- // the field closes, so a frontier-but-no-reservation record is refused.
- // Pre-production, so an older data directory is wiped, not migrated.
+ // It cannot recover what the old build never wrote down -- offsets acked
+ // out of RAM above the frontier are gone with the process either way --
+ // but refusing the record recovers nothing and costs the node its boot.
let mut padded = [0u8; ENCODED_LEN];
match bytes.len() {
ENCODED_LEN => padded.copy_from_slice(bytes),
- ENCODED_LEN_WITHOUT_FRONTIER => {
- padded[..ENCODED_LEN_WITHOUT_FRONTIER].copy_from_slice(bytes);
+ ENCODED_LEN_WITHOUT_RESERVATION => {
+ padded[..ENCODED_LEN_WITHOUT_RESERVATION].copy_from_slice(bytes);
+ padded[66..74].copy_from_slice(&bytes[58..66]);
}
actual => {
return Err(VsrStateError::WrongLength {
@@ -210,8 +222,8 @@ impl fmt::Display for VsrStateError {
Self::WrongLength { expected, actual } => {
write!(
f,
- "VsrState needs {expected} bytes (or {ENCODED_LEN_WITHOUT_FRONTIER}, \
- the layout before the offset frontier), got {actual}"
+ "VsrState needs {expected} bytes (or {ENCODED_LEN_WITHOUT_RESERVATION}, \
+ the layout before the offset reservation), got {actual}"
)
}
Self::LogViewAheadOfView { view, log_view } => write!(
@@ -264,12 +276,12 @@ mod tests {
assert!(VsrState::try_from(&bytes[..ENCODED_LEN - 1]).is_err());
}
- /// A superblock written before `offset_frontier` existed must still decode:
- /// the metadata plane writes one on every view change, so an exact-length
- /// decode turns an in-place upgrade into a boot refusal on every deployment
- /// that ever ran.
+ /// A superblock written before `offset_reserved` existed must still decode:
+ /// every release from `server-0.9.0-edge.2` to `edge.6` wrote that layout on
+ /// both planes, and the metadata plane refuses the whole node's boot on a
+ /// record it cannot decode.
#[test]
- fn given_pre_frontier_record_when_decoded_should_accept_and_zero_fill() {
+ fn given_pre_reservation_record_when_decoded_should_fill_from_the_frontier() {
let full = VsrState {
cluster: 3,
replica_id: 1,
@@ -284,31 +296,56 @@ mod tests {
}
.to_bytes();
- let legacy = &full[..ENCODED_LEN_WITHOUT_FRONTIER];
- let decoded = VsrState::try_from(legacy).expect("a pre-frontier record must decode");
- assert_eq!(decoded.offset_frontier, 0, "the new field zero-fills");
- assert_eq!(decoded.offset_reserved, 0, "the new field zero-fills");
+ let legacy = &full[..ENCODED_LEN_WITHOUT_RESERVATION];
+ let decoded = VsrState::try_from(legacy).expect("a pre-reservation record must decode");
+ assert_eq!(decoded.offset_frontier, 77);
+ assert_eq!(
+ decoded.offset_reserved, 77,
+ "the reservation fills from the frontier, not from zero: a 0 would claim \
+ nothing was reserved for offsets the frontier says exist"
+ );
assert_eq!(decoded.view, 9);
assert_eq!(decoded.log_view, 8);
assert_eq!(decoded.commit_max, 41);
assert_eq!(decoded.checkpoint_op, 7);
assert_eq!(decoded.checkpoint_checksum, 5);
- // Anything that is neither layout is still refused.
- assert!(matches!(
- VsrState::try_from(&full[..40]),
- Err(VsrStateError::WrongLength { .. })
- ));
+ // Anything that is neither layout is still refused, the pre-frontier
+ // layout included: it left trunk before any release carried it, so a
+ // record that short is corruption, not history.
+ for short in [40, 58] {
+ assert!(
+ matches!(
+ VsrState::try_from(&full[..short]),
+ Err(VsrStateError::WrongLength { .. })
+ ),
+ "a {short}-byte record must not decode"
+ );
+ }
+ }
- // Including the frontier-but-no-reservation layout every earlier build
- // wrote: zero-filling it would let a restart re-mint confirmed offsets.
- assert!(
- matches!(
- VsrState::try_from(&full[..ENCODED_LEN - 8]),
- Err(VsrStateError::WrongLength { .. })
- ),
- "a record without the reservation must not zero-fill it"
- );
+ /// A zero frontier is the shape a record written before either offset field
+ /// existed decodes into, and the fill must not turn that into a claim.
+ #[test]
+ fn given_pre_reservation_record_with_no_frontier_when_decoded_should_reserve_nothing() {
+ let full = VsrState {
+ cluster: 3,
+ replica_id: 1,
+ replica_count: 3,
+ view: 2,
+ log_view: 2,
+ commit_max: 0,
+ checkpoint_op: 0,
+ checkpoint_checksum: 0,
+ offset_frontier: 0,
+ offset_reserved: 0,
+ }
+ .to_bytes();
+
+ let decoded = VsrState::try_from(&full[..ENCODED_LEN_WITHOUT_RESERVATION])
+ .expect("a pre-reservation record must decode");
+ assert_eq!(decoded.offset_frontier, 0);
+ assert_eq!(decoded.offset_reserved, 0);
}
#[test]
diff --git a/core/integration/tests/cluster/crash_offset_reuse.rs b/core/integration/tests/cluster/crash_offset_reuse.rs
index 1ff7e03b2..3d8d175f1 100644
--- a/core/integration/tests/cluster/crash_offset_reuse.rs
+++ b/core/integration/tests/cluster/crash_offset_reuse.rs
@@ -121,10 +121,10 @@ async fn create_topic(client: &IggyClient, messages_required_to_save: Option<u32
/// Base offsets of every segment file under `root`, from the file names, which
/// are the on-disk claim about where each range begins.
///
-/// Reading them is the only way to assert the shape the re-anchor produces:
-/// recovery tolerates a discontiguity inside a segment whenever the index
-/// survives, so a black-box offset assertion passes either way and the wrong
-/// shape sits there until an index is torn.
+/// Reading them is the only way to assert the shape the re-anchor produces. A
+/// black-box offset assertion passes either way on the boot that WRITES the
+/// wrong shape; the cost only lands on the boot that reads it back, where the
+/// walk refuses and the solo arm tombstones the partition.
fn segment_base_offsets(root: &Path) -> Vec<u64> {
let mut offsets = Vec::new();
let mut stack = vec![root.to_path_buf()];
@@ -160,31 +160,13 @@ async fn given_confirmed_sends_below_flush_threshold_when_a_solo_node_is_killed_
harness: &mut TestHarness,
) {
let client = harness.tcp_root_client().await.unwrap();
- client
- .create_stream(STREAM_NAME)
- .await
- .expect("create stream");
- client
- .create_topic(
- &Identifier::named(STREAM_NAME).unwrap(),
- TOPIC_NAME,
- &TopicCreateOptions {
- partitions_count: Some(1),
- message_expiry: Some(IggyExpiry::NeverExpire),
- ..TopicCreateOptions::default()
- },
- )
- .await
- .expect("create topic");
+ create_topic(&client, None).await;
let acked = produce_acked(&client, "pre-crash", PRE_CRASH_SENDS).await;
let highest_confirmed = *acked.last().expect("confirmed sends");
drop(client);
- harness.kill_node(0).expect("SIGKILL the only node");
- harness.restart_node(0).expect("restart it");
-
- let client = wait_until_serving(harness, SERVE_TIMEOUT).await;
+ let client = crash_and_recover(harness).await;
let post_crash_offset = produce_acked(&client, "post-crash", 1).await[0];
assert!(
@@ -196,11 +178,120 @@ async fn given_confirmed_sends_below_flush_threshold_when_a_solo_node_is_killed_
so two different messages now share an offset and consumers reading by offset get \
silently different data"
);
+
+ // The shape this boot WROTE is only paid for by the boot that reads it back:
+ // nothing reached a segment before the crash, so the re-anchor emptied the
+ // chain and had to plant at the append point rather than leave the segment
+ // named 0 with the first mint a lease block inside it.
+ let bases = segment_base_offsets(harness.test_dir());
+ assert!(
+ bases.iter().any(|&base| base > highest_confirmed),
+ "the re-anchor left no segment above the pre-crash offsets, so the post-crash \
+ mint landed inside a segment named below it: segment bases {bases:?}, last \
+ offset confirmed before the crash {highest_confirmed}"
+ );
+ drop(client);
+
+ // And the boot that reads it: a hole inside a segment refuses the walk and
+ // tombstones the partition, which shows up here as a node that never serves
+ // the stream again.
+ let client = crash_and_recover(harness).await;
+ let third_life = produce_acked(&client, "third-life", 1).await[0];
+ assert!(
+ third_life > post_crash_offset,
+ "the second restart re-minted: {post_crash_offset} was confirmed after the \
+ first crash, yet the node came back and handed out {third_life}"
+ );
+}
+
+/// The graceful stop is the runbook answer to an incident, so it must not be the
+/// action that undoes the fix. Between the two boots here the node takes no
+/// traffic at all: nothing reaches a segment, the committed frontier stays at
+/// what the crash left, and the reservation is the only record that offsets were
+/// handed out.
+///
+/// End to end, not a probe of one mechanism: three separate things carry the
+/// append point across this sequence (the boot seed, the segment the re-anchor
+/// plants, and the collapse writing the append point rather than the committed
+/// frontier), so any one of them alone keeps this green. The collapse is pinned
+/// on its own by `given_an_unspent_reservation_when_collapsing_should_leave_it_standing`
+/// in `core/partitions`.
+#[iggy_harness(cluster_nodes = 1)]
+async fn given_a_crash_restarted_node_when_stopped_cleanly_should_still_not_remint_offsets(
+ harness: &mut TestHarness,
+) {
+ let client = harness.tcp_root_client().await.unwrap();
+ create_topic(&client, None).await;
+
+ let acked = produce_acked(&client, "pre-crash", PRE_CRASH_SENDS).await;
+ let highest_confirmed = *acked.last().expect("confirmed sends");
+ drop(client);
+
+ // Boot one: reads the reservation back and never spends it.
+ let client = crash_and_recover(harness).await;
+ drop(client);
+
+ // The clean stop, then boot two. `restart_node` stops the running node with
+ // SIGTERM and waits for it, so the shutdown flush and its collapse both run.
+ harness
+ .restart_node(0)
+ .expect("cleanly restart the only node");
+ let client = wait_until_serving(harness, SERVE_TIMEOUT).await;
+ let post_stop_offset = produce_acked(&client, "post-clean-stop", 1).await[0];
+
+ assert!(
+ post_stop_offset > highest_confirmed,
+ "a clean stop between the crash and the next send re-minted: offset \
+ {highest_confirmed} was handed to a client before the SIGKILL, and after a \
+ graceful restart the node confirmed {post_stop_offset}. The shutdown collapse \
+ dropped a reservation the boot had not spent yet"
+ );
+}
+
+/// The replicated fence path, which no other case here reaches. A three-node
+/// group acks a send once a quorum has journaled it, so it claims no reservation
+/// and re-anchors nothing -- and one node crashing must still not disturb the
+/// offsets the group hands out.
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_replicated_group_when_a_node_is_killed_should_not_remint_offsets(
+ harness: &mut TestHarness,
+) {
+ let client = harness.tcp_root_client().await.unwrap();
+ create_topic(&client, None).await;
+
+ let acked = produce_acked(&client, "pre-crash", PRE_CRASH_SENDS).await;
+ let highest_confirmed = *acked.last().expect("confirmed sends");
+ assert_eq!(
+ acked,
+ (0..u64::from(PRE_CRASH_SENDS)).collect::<Vec<_>>(),
+ "the pre-crash run mints a contiguous range from zero"
+ );
+ drop(client);
+
+ let client = crash_and_recover(harness).await;
+ let post_crash_offset = produce_acked(&client, "post-crash", 1).await[0];
+
+ assert!(
+ post_crash_offset > highest_confirmed,
+ "a replicated group re-minted after one node restarted: {highest_confirmed} was \
+ confirmed before the SIGKILL, then {post_crash_offset} after it"
+ );
+
+ // No reservation was claimed, so no gap was planted: a replicated group's
+ // segment boundaries have to stay a function of its batches alone, or the
+ // reconciler's offset-keyed segment GC never converges.
+ let bases = segment_base_offsets(harness.test_dir());
+ assert!(
+ bases.iter().all(|&base| base <= post_crash_offset),
+ "a replicated group planted a segment above every offset it minted, so it \
+ re-anchored around a reservation it should never have claimed: bases {bases:?}"
+ );
}
/// The fix has to survive its own side effect: the hole the reservation leaves
-/// between the recovered segments and the new append point truncates everything
-/// past it on the next boot if it lands INSIDE a segment.
+/// between the recovered segments and the new append point makes the next boot
+/// REFUSE the chain if it lands INSIDE a segment, which tombstones the partition
+/// on a solo node.
///
/// So the SECOND crash is the one that matters, and only if the run between the
/// two reaches disk, which is what the flush threshold is for.
@@ -249,7 +340,7 @@ async fn given_a_crash_restarted_node_when_it_flushes_and_crashes_again_should_s
third_life > second_life_max,
"the SECOND restart re-minted: {second_life_max} was confirmed between the \
two crashes, yet the node came back and handed out {third_life}. A hole \
- left INSIDE a segment truncates the tail that proved the frontier"
+ left INSIDE a segment costs the tail that proved the frontier"
);
}
diff --git a/core/journal/src/superblock.rs b/core/journal/src/superblock.rs
index fd8b76124..77f49ffe0 100644
--- a/core/journal/src/superblock.rs
+++ b/core/journal/src/superblock.rs
@@ -70,8 +70,8 @@ const MIN_RECORD_LEN: usize = HEADER_LEN + CHECKSUM_LEN;
/// Ceiling on a record's payload, bounding every allocation this module makes from
/// a length it read off disk (`PrepareJournal::MAX_ENTRY_SIZE` bounds the WAL for the
-/// same reason). The only payload today is a [`consensus::VsrState`], 66 bytes now
-/// that it carries the offset frontier (58 before it, a length its decode still
+/// same reason). The only payload today is a [`consensus::VsrState`], 74 bytes now
+/// that it carries the offset reservation (66 before it, a length its decode still
/// accepts); the headroom is for a payload that grows fields, not for bulk data.
/// `read_slot` treats a longer file as corrupt WITHOUT reading it, and
/// `build_record` refuses to write one, so a length this store could have
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 960f639ba..a3322114a 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -70,8 +70,8 @@ use server_common::{
MESSAGE_ALIGN, Message, SegmentStorage,
iobuf::{Frozen, Owned},
send_messages::{
- BatchHeader, ChecksumMode, convert_request_message, decode_prepare_slice,
- decode_prepare_slice_trusted, stamp_prepare_for_persistence,
+ BatchHeader, ChecksumMode, convert_request_message, decode_batch_slice,
+ decode_prepare_slice, decode_prepare_slice_trusted, stamp_prepare_for_persistence,
},
sharding::IggyNamespace,
};
@@ -242,11 +242,6 @@ where
/// Offsets the append fence claims per superblock write; installed by boot
/// from `PartitionsConfig`.
offset_reservation_lease: u64,
- /// Whether the recorded reservation still has to floor a mint. Armed when a
- /// record is attached at boot, spent by the first mint after it: it is the
- /// PREVIOUS incarnation's claim, and applying it again would jump the
- /// counter to the top of the block on every mint.
- mint_floor_pending: Cell<bool>,
/// In-flight state transfer for this group (rejoin whose repair floor was
/// refused); tail repair takes over at install. See
/// [`PartitionTransferSession`].
@@ -510,7 +505,6 @@ where
durable_offset_frontier: Cell::new(0),
durable_offset_reserved: Cell::new(0),
offset_reservation_lease: u64::from(iggy_common::DEFAULT_OFFSET_RESERVATION_LEASE),
- mint_floor_pending: Cell::new(false),
transfer: None,
transfer_attempts: 0,
transfer_failures: 0,
@@ -618,11 +612,8 @@ where
self.superblock = Some(superblock);
self.durable_offset_frontier
.set(recovered.map_or(0, |state| state.offset_frontier));
- let reserved = recovered.map_or(0, |state| state.offset_reserved);
- self.durable_offset_reserved.set(reserved);
- // Attaching a record is the moment the previous incarnation's claim
- // becomes this one's constraint.
- self.mint_floor_pending.set(reserved > 0);
+ self.durable_offset_reserved
+ .set(recovered.map_or(0, |state| state.offset_reserved));
}
/// Persist this group's VSR state to its superblock when the view changed
@@ -730,9 +721,10 @@ where
}
/// The advance rule for the frontier, shared by every writer that claims
- /// one: never below what bytes prove, never below what the record holds.
- /// Bytes, NOT the append counter, which stands a lease block above them
- /// after a reservation-seeded boot.
+ /// one: never below what this replica holds, never below what the record
+ /// already says. Held messages, NOT [`Self::mint_frontier`], which stands a
+ /// lease block above them after a reservation-seeded boot and names none of
+ /// them.
fn advanced_frontier(&self, claim: u64) -> u64 {
claim
.max(self.held_offset_frontier())
@@ -846,34 +838,69 @@ where
/// recompute `batch_checksum` over the result, so the next replicated
/// prepare would persist different bytes here than on every peer, silently.
///
- /// The RESERVATION is deliberately not folded in here: a backup mints
- /// nothing, it re-stamps what the primary sends and rejects anything that
- /// does not continue its own counter, so seeding from a ceiling would put it
- /// a lease block above its group and fork its chain. Applied at the point of
- /// MINTING instead: see `Self::mint_floor`.
+ /// The two counters are seeded separately, because the record carries two
+ /// bounds. `offset_frontier` is what messages reached, so it seeds the
+ /// COMMITTED counter; `offset_reserved` is what may have been handed to a
+ /// client, so it seeds the APPEND counter and nothing else. Folding the
+ /// reservation into the committed one would publish a `current_offset` over
+ /// a lease-block hole, and `store_consumer_offset` would then admit offsets
+ /// inside it.
+ ///
+ /// The reservation is SOLO ONLY. A backup mints nothing: it re-stamps what
+ /// the primary sends and rejects anything that does not continue its own
+ /// counter, so an append point a lease block above its group would have every
+ /// peer refuse the batch. A replicated group is also less exposed, since an
+ /// ack there means a quorum journaled the batch and the hole needs a
+ /// FULL-cluster crash.
///
/// Lives HERE rather than in the server crate so the boot paths and the
/// simulator share one implementation. A copy in the harness was a copy of
/// the max rule that had lost the max, in the one place built to catch
/// violations of it.
pub fn restore_offset_frontier(&mut self, recovered: Option<&consensus::VsrState>) {
- let Some(frontier) = recovered
- .map(|state| state.offset_frontier)
- .filter(|&f| f > 0)
- else {
+ let Some(state) = recovered else {
return;
};
- let recovered_end = frontier - 1;
- if self.should_increment_offset && self.offset.load(Ordering::Acquire) >= recovered_end {
+ let frontier = state.offset_frontier;
+ let reserved = if self.consensus.replica_count() == 1 {
+ state.offset_reserved
+ } else {
+ 0
+ };
+ // NOT `frontier > 0`: the shape a crash before the first flush leaves is
+ // a zero frontier and a nonzero reservation, because the append fence
+ // runs before the journal append, so the first record a partition ever
+ // writes names no data at all.
+ let append_point = frontier.max(reserved);
+ if append_point == 0 {
+ return;
+ }
+ let seeded = self.should_increment_offset;
+ // Each counter takes its own max, since the record's two bounds move
+ // independently: a graceful stop collapses the reservation onto the
+ // append point while the frontier stays where the data ended.
+ let committed_restored = frontier.checked_sub(1).is_some_and(|committed_end| {
+ let raise = !seeded || self.offset.load(Ordering::Acquire) < committed_end;
+ if raise {
+ self.offset.store(committed_end, Ordering::Release);
+ }
+ raise
+ });
+ let append_end = append_point - 1;
+ let append_restored = !seeded || self.dirty_offset.load(Ordering::Relaxed) < append_end;
+ if append_restored {
+ self.dirty_offset.store(append_end, Ordering::Relaxed);
+ }
+ if !committed_restored && !append_restored {
return;
}
tracing::info!(
namespace_raw = self.consensus().group(),
offset_frontier = frontier,
- "restored partition offset frontier from its superblock"
+ offset_reserved = reserved,
+ append_point,
+ "restored partition offset counters from its superblock"
);
- self.offset.store(recovered_end, Ordering::Release);
- self.dirty_offset.store(recovered_end, Ordering::Relaxed);
self.should_increment_offset = true;
}
@@ -954,8 +981,13 @@ where
.set_current_offset(self.offset.load(Ordering::Acquire));
}
- /// The next message offset this replica will mint, `0` while the offset
- /// space is still empty. The value stamped into the durable record.
+ /// One past the highest COMMITTED offset, `0` while the offset space is
+ /// still empty. The value stamped into the durable record, and what a
+ /// transfer offer advertises.
+ ///
+ /// Not [`Self::mint_frontier`]: after a reservation-seeded boot the append
+ /// point stands a lease block above this, and neither the record nor an
+ /// offer may claim offsets no message reached.
#[must_use]
pub fn offset_frontier(&self) -> u64 {
if self.should_increment_offset {
@@ -965,15 +997,37 @@ where
}
}
- /// One past the highest offset this replica actually HOLDS: on disk in a
- /// sized segment, or resident in the journal. `0` when it holds nothing.
+ /// The offset the next mint will take, which is where the segment chain has
+ /// to be anchored for the appends that follow to land contiguously.
+ ///
+ /// Above [`Self::offset_frontier`] by exactly the offsets this replica has
+ /// journaled but not committed, plus -- on the first boot after a crash that
+ /// took acked-but-unflushed messages with it -- the lease block the durable
+ /// reservation claimed. That gap is the whole point: the reservation is the
+ /// only surviving witness that those offsets were handed to a client, so the
+ /// counter resumes above them instead of re-minting them.
+ #[must_use]
+ pub fn mint_frontier(&self) -> u64 {
+ if self.should_increment_offset {
+ self.dirty_offset.load(Ordering::Relaxed).saturating_add(1)
+ } else {
+ 0
+ }
+ }
+
+ /// One past the highest offset this replica holds and may not lose: named by
+ /// a sized segment, or committed and still resident in the journal. `0` when
+ /// it holds nothing.
///
- /// Not [`Self::offset_frontier`], which reads the append counter: after a
- /// reservation-seeded boot that stands a lease block above the last byte
- /// anywhere. Anything asking "would this destroy something I have?" has to
- /// ask about messages. The journal arm is not redundant either, since the
+ /// The committed arm is not redundant with the disk arm, since the
/// threshold-gated flush routinely leaves committed messages unnamed by any
- /// segment.
+ /// segment. It reads the COMMITTED counter and not `journal.info`, whose
+ /// `current_offset` is the DIRTY tail: a view change truncates that tail
+ /// (`truncate_uncommitted_from`) while the durable frontier only advances, so
+ /// recording it would leave every later boot seeding the counter above the
+ /// group, where each replicated prepare fails the
+ /// `base_offset == dirty_offset + 1` check until a state transfer -- again on
+ /// the boot after that one.
#[must_use]
pub fn held_offset_frontier(&self) -> u64 {
let on_disk = self
@@ -984,13 +1038,7 @@ where
.map(|segment| segment.end_offset.saturating_add(1))
.max()
.unwrap_or(0);
- let journal = self.log.journal().info;
- let resident = if journal.messages_count > 0 {
- journal.current_offset.saturating_add(1)
- } else {
- 0
- };
- on_disk.max(resident)
+ on_disk.max(self.offset_frontier())
}
/// Force the durable record to catch up with the current offset frontier,
@@ -1104,16 +1152,37 @@ where
/// standing it would make every ordinary restart resume a lease block
/// higher and hole the offset space for nothing.
///
+ /// Collapses onto [`Self::mint_frontier`], not [`Self::offset_frontier`]: the
+ /// append point is what the next boot has to resume at, and on a boot that
+ /// consumed a reservation without appending it is the reservation itself, so
+ /// reading the committed frontier here would write a record BELOW what an
+ /// earlier life already confirmed to a client. A clean stop is the runbook
+ /// answer to an incident, which would make it the one action that undoes the
+ /// protection.
+ ///
+ /// The frontier field still records only what is held: a graceful stop
+ /// flushes the committed prefix, but the journal can hold an uncommitted tail
+ /// that the next view legitimately truncates.
+ ///
/// Callers must have flushed FIRST, and must not call this when the flush
/// failed: the claim it makes is precisely that the flush succeeded.
#[allow(clippy::future_not_send)]
#[must_use = "the bool is the durability verdict; a failed collapse leaves a gap"]
pub async fn collapse_offset_reservation(&self) -> bool {
- let frontier = self.offset_frontier();
- if self.durable_offset_reserved.get() <= frontier {
+ let append_point = self.mint_frontier();
+ if self.durable_offset_reserved.get() <= append_point {
return true;
}
- self.reset_offset_frontier_at(frontier).await
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ if self.superblock_write_is_backed_off() {
+ return false;
+ }
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ let held = self.advanced_frontier(0);
+ self.write_superblock_inner(superblock.as_ref(), held, append_point)
+ .await
}
/// [`Self::persist_offset_frontier`] for a frontier this replica has not
@@ -1139,33 +1208,75 @@ where
self.write_superblock(superblock.as_ref(), frontier).await
}
- /// The lowest offset this replica may mint, given what its durable record
- /// already permitted it to hand out. A floor, not a seed -- see
- /// [`Self::restore_offset_frontier`]. It bites once, on the first mint after
- /// a crash that took acked-but-unflushed messages with it: segments and
- /// counter come back short, and the reservation is the only surviving
- /// witness that those offsets were confirmed.
+ /// Whether the offset reservation is close enough to being consumed that it
+ /// should be extended NOW, off the append path.
///
- /// SOLO GROUPS ONLY. A backup validates an incoming prepare with
- /// `base_offset == dirty_offset + 1`
- /// ([`Self::append_received_send_messages_to_journal`]), so a primary
- /// minting from a floor its peers do not share would have every one of them
- /// refuse the batch; relaxing that check deserves its own evidence. A
- /// replicated group is also less exposed -- an ack means a quorum journaled
- /// the batch, so the hole there is a FULL-cluster crash.
- fn mint_floor(&self) -> u64 {
- let floor = self.armed_mint_floor();
- self.mint_floor_pending.set(false);
- floor
+ /// The append fence is correct but badly placed: it writes the superblock
+ /// inline in the shard's frame pump, where the consensus tick is a sibling
+ /// arm, so its two fsyncs delay heartbeats for every group on the core. The
+ /// fix is to make the fence's fast path
+ /// (`durable_offset_reserved > end_offset`) the only path it ever takes under
+ /// load, by extending from the tick instead.
+ ///
+ /// HALF a block of headroom, which is a wide margin on purpose: over-claiming
+ /// costs nothing but offset space, while arriving late puts the write back on
+ /// the append path. A partition that has never minted is skipped -- its first
+ /// append legitimately pays for the first claim, and extending every idle
+ /// partition at boot would write a superblock per partition for nothing.
+ #[must_use]
+ pub fn needs_offset_reservation_extension(&self) -> bool {
+ if self.consensus.replica_count() > 1
+ || self.superblock.is_none()
+ || !self.should_increment_offset
+ {
+ return false;
+ }
+ let headroom = self
+ .durable_offset_reserved
+ .get()
+ .saturating_sub(self.mint_frontier());
+ headroom < self.offset_reservation_lease / 2
}
- /// [`Self::mint_floor`] without spending it, for the boot re-anchor, which
- /// has to know where the first mint will land before there is one.
- const fn armed_mint_floor(&self) -> u64 {
- if self.consensus.replica_count() > 1 || !self.mint_floor_pending.get() {
- return 0;
+ /// Extend the reservation a full block past the current append point.
+ ///
+ /// Pairs with [`Self::needs_offset_reservation_extension`]; the caller is the
+ /// shard tick, so this write is off the append path. A failure needs no
+ /// handling beyond the writer's own logging and backoff: the fence at the
+ /// mint is still there, and it is what refuses the append if the ceiling
+ /// never caught up.
+ #[allow(clippy::future_not_send)]
+ pub async fn extend_offset_reservation(&self) -> bool {
+ let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else {
+ return true;
+ };
+ if self.superblock_write_is_backed_off() {
+ return false;
}
- self.durable_offset_reserved.get()
+ let _superblock_guard = self.superblock_lock.acquire().await;
+ self.write_offset_claim(superblock.as_ref(), self.mint_frontier())
+ .await
+ }
+
+ /// Upper bound on the offsets a pending `SendMessages` request will mint, for
+ /// fencing it BEFORE it enters the pipeline.
+ ///
+ /// `project` assigns an op, not a base offset, so the exact range is unknown
+ /// until the mint runs under `write_lock`. This is deliberately loose: a
+ /// request pipelined behind others can land above it, and the fence at the
+ /// mint stays as the exact check. It does not need to be tight -- the claim
+ /// runs a whole lease block past whatever it is handed, so one of these
+ /// covers every batch in flight unless a run of them crosses a block
+ /// boundary.
+ ///
+ /// `None` when the body is not one canonical batch, which
+ /// `convert_request_message` has already rejected by the time this runs.
+ fn request_mint_ceiling(&self, message: &Message<RoutedRequestHeader>) -> Option<u64> {
+ let body = message
+ .as_slice()
+ .get(std::mem::size_of::<RoutedRequestHeader>()..message.header().size as usize)?;
+ let count = decode_batch_slice(body).ok()?.message_count();
+ Some(self.mint_frontier().saturating_add(u64::from(count)))
}
/// The append fence: make sure the durable record already permits every
@@ -1177,18 +1288,42 @@ where
/// is client-visible long before the threshold-gated flush names it in a
/// segment, and a crash in between hands a second message an offset a client
/// already holds. Fencing here rather than at commit puts it upstream of
- /// every way an offset escapes -- the reply, the poll tier, the peers a
- /// prepare reaches -- on the one path both a primary's mint and a backup's
- /// re-stamp take. Claiming through `end_offset + 1 + lease` rather than from
+ /// every way a NEWLY minted offset escapes -- the reply, the poll tier, the
+ /// peers a prepare reaches -- on the one path both a primary's mint and a
+ /// backup's re-stamp take. Journal repair
+ /// (`append_repaired_send_messages`) is the exception and needs none: it
+ /// re-journals offsets a peer already minted and fenced, so there is nothing
+ /// new to claim. Claiming through `end_offset + 1 + lease` rather than from
/// the live counter needs no special case for an oversized batch.
///
- /// `false` only when a write was attempted and failed, and the caller must
- /// refuse the append. Fail-closed: the send is rejected with nothing
- /// externalised, or a backup withholds its `PrepareOk` and the group elects
- /// around it, exactly as a failed view persist does.
+ /// SOLO ONLY, like everything the reservation feeds: `restore_offset_frontier`
+ /// seeds no counter from it above one replica, and the boot re-anchor that
+ /// shapes the chain around it never runs there either. A replicated group
+ /// paying a superblock write per block would buy nothing -- and an ack there
+ /// already means a quorum journaled the batch, so re-minting needs a
+ /// FULL-cluster crash.
+ ///
+ /// `false` when a write was attempted and failed, and also when an earlier
+ /// failure's backoff window is still open, in which case no write is
+ /// attempted at all -- that cell is shared with every other superblock writer
+ /// on the partition, so this can refuse with no fault on the append path
+ /// itself. Either way the caller must refuse the append. Fail-closed: the
+ /// send is rejected with nothing externalised, exactly as a failed view
+ /// persist withholds its sends.
+ ///
+ /// WHERE it is called decides how much a refusal costs. Ahead of the pipeline
+ /// (`on_request`) the client gets a retryable transient and the group keeps
+ /// serving. At the mint the op already has its number and its ack is already
+ /// skipped, so `commit_max` can never pass it and nothing later can commit
+ /// either: `on_replicate` fences the partition there and takes the node down,
+ /// because a one-second backoff must not cost a partition the rest of the
+ /// process's life in the dark.
#[allow(clippy::future_not_send)]
#[must_use = "the bool is the fence verdict; dropping it lets the append escape unreserved"]
pub async fn reserve_offsets_through(&self, end_offset: u64) -> bool {
+ if self.consensus.replica_count() > 1 {
+ return true;
+ }
// A frontier: offsets strictly below it are permitted, so covering
// `end_offset` needs a record strictly above it.
if self.durable_offset_reserved.get() > end_offset {
@@ -1206,11 +1341,23 @@ where
if self.durable_offset_reserved.get() > end_offset {
return true;
}
+ self.write_offset_claim(superblock.as_ref(), end_offset)
+ .await
+ }
+
+ /// Claim a block past `end_offset` unconditionally.
+ ///
+ /// Split from [`Self::reserve_offsets_through`] because the tick's extension
+ /// has to write while the ceiling still covers the append point -- that is the
+ /// point of extending early -- and the fence's coverage fast path would
+ /// short-circuit exactly that call. The advancing write is what keeps this
+ /// monotone: a claim below the record cannot lower it.
+ #[allow(clippy::future_not_send)]
+ async fn write_offset_claim(&self, superblock: &SB, end_offset: u64) -> bool {
let claim = end_offset
.saturating_add(1)
.saturating_add(self.offset_reservation_lease);
- self.write_superblock_advancing(superblock.as_ref(), 0, claim)
- .await
+ self.write_superblock_advancing(superblock, 0, claim).await
}
/// Burn one transfer stall round; `true` once the budget is exhausted.
@@ -1324,8 +1471,8 @@ where
/// Carried on the partition because the fence runs inside `on_request` /
/// `on_replicate`, which take no config. Floored at 1: a zero block reserves
/// nothing and would write the superblock before every append.
- pub const fn set_offset_reservation_lease(&mut self, lease: u32) {
- self.offset_reservation_lease = if lease == 0 { 1 } else { lease as u64 };
+ pub fn set_offset_reservation_lease(&mut self, lease: u32) {
+ self.offset_reservation_lease = if lease == 0 { 1 } else { u64::from(lease) };
}
/// Whether this partition's segments reserve their bytes on open.
@@ -2048,7 +2195,7 @@ where
// Only here: this is the only path that mints. A backup re-stamps what
// the primary sends (`append_received_send_messages_to_journal`) and
// must follow it exactly, so raising ITS counter would fork the group.
- let dirty_offset = next.max(self.mint_floor());
+ let dirty_offset = next;
// Reuse the prepare's monotonic timestamp, assigned once by the primary
// in `project()` (`next_monotonic_timestamp`) and replicated verbatim to
@@ -2074,6 +2221,15 @@ where
self.fatal.as_ref()
}
+ /// Consecutive superblock write failures for this group, for the shard's
+ /// wedge fail-stop. A partition that cannot record its state withholds every
+ /// view-scoped send and refuses every append, so past some window it is
+ /// serving nothing and a supervisor should be handling it instead.
+ #[must_use]
+ pub const fn superblock_write_failures(&self) -> u64 {
+ self.superblock_write_failures.get()
+ }
+
/// Fence this partition after the shutdown flush failed to persist its
/// committed journal prefix: that data is cluster-committed and now lives
/// only in this process's memory, so the shard must not report a clean
@@ -2361,6 +2517,40 @@ where
offset,
}
} else {
+ // Fence AHEAD of the pipeline for a mint, not only at the mint.
+ // A refusal at the mint arrives after the sequencer took the op,
+ // where the only honest answer left is to fence the partition and
+ // take the node down (`on_replicate`). Here the request has
+ // entered nothing, so a transient disk fault costs the client one
+ // retry instead of costing the process its life.
+ //
+ // `TransientNotAccepted`, per its contract: nothing was admitted,
+ // so the client may re-issue anywhere without double-apply risk.
+ // It does make the SDK recheck the leader and walk the roster,
+ // which finds no better node when the fault is this one's disk --
+ // wasteful, but the weaker code would claim an unknown outcome
+ // for a request that provably has none.
+ if message.header().operation == Operation::SendMessages
+ && let Some(ceiling) = self.request_mint_ceiling(&message)
+ && !self.reserve_offsets_through(ceiling).await
+ {
+ emit_partition_diag(
+ tracing::Level::WARN,
+ &PartitionDiagEvent::new(
+ ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
+ "refusing a send: the offset reservation could not be extended",
+ )
+ .with_operation(Operation::SendMessages),
+ );
+ Self::send_partition_deny_or_log(
+ consensus,
+ message.header(),
+ IggyError::TransientNotAccepted.as_code(),
+ "unreserved send transient reply send failed",
+ )
+ .await;
+ return;
+ }
// Two-queue: prepare slot -> project+replicate; prepare full +
// request room -> buffer; both full -> drop+warn (client retries
// via read-timeout).
@@ -2674,16 +2864,57 @@ where
let frozen_for_forward = match replicated_result {
Ok(frozen) => frozen,
Err(error) => {
+ // A BACKUP refusing here is the design, not a fault: it rejects
+ // any prepare whose `base_offset` does not continue its own
+ // counter, which is exactly what a dropped or out-of-order
+ // prepare leaves, and withholding `PrepareOk` is the fail-closed
+ // answer. The primary retransmits, journal repair fills the gap,
+ // and the group elects around the replica if it cannot catch up.
+ // Nothing here is owed an ack this replica already skipped.
+ //
+ // On the PRIMARY the same return is unrecoverable. The op is
+ // already in the pipeline with the sequencer advanced past it
+ // (`push_prepare_entry`), and this sits ahead of
+ // `send_prepare_ok`, so it never gets its ack and `commit_max`
+ // can never pass it: every later op journals fine and queues
+ // behind it forever. Nothing lifts that -- the prepare timeout
+ // only backs off, a solo group's retransmit target is itself,
+ // this plane has no `repair_primary_self_acks`, and a solo group
+ // never starts a view change. Clients get no reply at all, since
+ // replies are generated on commit, so they wait out their read
+ // timeout, and once the queues fill so does every send after.
+ //
+ // So fence there, the way a failed local commit of a
+ // cluster-committed op does: the shard picks `fatal` up on its
+ // next tick and takes the node down. A one-second superblock
+ // backoff must not cost a partition the rest of the process's
+ // life in the dark.
emit_partition_diag(
- tracing::Level::WARN,
+ if is_backup {
+ tracing::Level::WARN
+ } else {
+ tracing::Level::ERROR
+ },
&PartitionDiagEvent::new(
self.diag_ctx(),
- "failed to apply replicated partition operation",
+ if is_backup {
+ "failed to apply replicated partition operation"
+ } else {
+ "failed to apply an operation this replica sequenced; \
+ fencing the partition and shutting down"
+ },
)
.with_operation(header.operation)
.with_op(header.op)
.with_error(error.to_string()),
);
+ if !is_backup && self.fatal.is_none() {
+ self.fatal = Some(FatalCommit {
+ namespace_raw: self.namespace().inner(),
+ op: header.op,
+ operation: header.operation,
+ });
+ }
return;
}
};
@@ -4088,98 +4319,49 @@ where
}
async fn rotate_segment(&mut self, config: &PartitionsConfig) -> Result<(), IggyError> {
- let namespace = self.namespace();
- let old_segment_index = self.log.segments().len() - 1;
- let active_segment = self.log.active_segment_mut();
- active_segment.sealed = true;
- let start_offset = active_segment.end_offset + 1;
-
- let segment_size = self.effective_segment_size(config);
- let enforce_fsync = self.effective_enforce_fsync(config);
- let preallocate_segments = self.effective_preallocate_segments(config);
- let segment = Segment::new(start_offset, segment_size);
- // Prefer the active writer's location: a per-topic path override or a
- // config change after the initial segment was created must not scatter
- // one partition's segments across two directories. The config layout
- // only decides for a partition with no writer yet.
- let (messages_path, index_path) = self.partition_dir().map_or_else(
- || {
- (
- config.get_messages_path(
- namespace.stream_id(),
- namespace.topic_id(),
- namespace.partition_id(),
- start_offset,
- ),
- config.get_index_path(
- namespace.stream_id(),
- namespace.topic_id(),
- namespace.partition_id(),
- start_offset,
- ),
- )
- },
- |dir| {
- (
- format!("{dir}/{start_offset:0>20}.log"),
- format!("{dir}/{start_offset:0>20}.index"),
- )
- },
- );
+ let start_offset = self.log.active_segment().end_offset + 1;
+ self.rotate_segment_at(config, start_offset).await
+ }
- let storage = SegmentStorage::new(&messages_path, &index_path, 0, 0, false)
- .await
- .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?;
- let messages_size_bytes = storage
- .messages_writer
- .as_ref()
- .ok_or_else(|| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?
- .size_counter();
- let messages_writer = Rc::new(
- MessagesWriter::new(
- &messages_path,
- messages_size_bytes,
- enforce_fsync,
- false,
- preallocate_segments.then_some(segment_size),
- )
- .await
- .map_err(|_| IggyError::CannotCreateSegmentLogFile(messages_path.clone()))?,
- );
- let index_size_bytes = storage
- .index_writer
- .as_ref()
- .ok_or_else(|| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?
- .size_counter();
- let index_writer = Rc::new(
- IggyIndexWriter::new(&index_path, index_size_bytes, enforce_fsync, false)
- .await
- .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?,
- );
+ /// Seal the active segment and plant a fresh empty one at `start_offset`.
+ ///
+ /// Shared by the size-driven roll, which plants at `end_offset + 1`, and the
+ /// boot re-anchor, which plants at the append point the reservation moved the
+ /// counter to. One seal path, and one order: the new segment's files are
+ /// created BEFORE the sealed segment's writers are torn down, so a failed
+ /// create leaves the chain serviceable.
+ async fn rotate_segment_at(
+ &mut self,
+ config: &PartitionsConfig,
+ start_offset: u64,
+ ) -> Result<(), IggyError> {
+ let namespace = self.namespace();
+ let sealed_index = self.log.segments().len() - 1;
+ let sealed_end = self.log.active_segment().end_offset;
+ self.log.active_segment_mut().sealed = true;
+ self.install_empty_segment(config, start_offset).await?;
+ self.stats.increment_segments_count(1);
- let old_storage = &mut self.log.storages_mut()[old_segment_index];
- let _ = old_storage.shutdown();
- self.log.messages_writers_mut()[old_segment_index] = None;
- self.log.index_writers_mut()[old_segment_index] = None;
+ let sealed_storage = &mut self.log.storages_mut()[sealed_index];
+ let _ = sealed_storage.shutdown();
+ self.log.messages_writers_mut()[sealed_index] = None;
+ self.log.index_writers_mut()[sealed_index] = None;
// Drop the sealed segment's in-memory index cache: only the ACTIVE
// segment's cache is ever read (the `commit_messages` flush staging),
// so a sealed cache is dead weight.
- self.log.indexes_mut()[old_segment_index] = None;
+ self.log.indexes_mut()[sealed_index] = None;
// The read fd cached while this segment was active is not counted by
// the sealed LRU budget, so it must not survive the seal; the next
// sealed poll re-fills the fresh slot under the LRU's rules.
- self.log.reset_read_state(old_segment_index);
-
- self.log
- .add_persisted_segment(segment, storage, Some(messages_writer), Some(index_writer));
- self.stats.increment_segments_count(1);
+ self.log.reset_read_state(sealed_index);
debug!(
target: "iggy.partitions.diag",
plane = "partitions",
namespace_raw = namespace.inner(),
+ sealed_end,
start_offset,
- "rotated to new segment"
+ "sealed the active segment and planted a fresh one"
);
Ok(())
}
@@ -4435,17 +4617,20 @@ where
/// what the recovered segment chain holds.
///
/// A hole INSIDE a segment is not survivable: `recover_segment_bounds` walks
- /// a segment from its FILENAME with a running `expected_offset`, so the next
- /// boot truncates the post-hole suffix and the counter falls back below what
- /// was already confirmed, undoing the fix on the second crash. On a segment
- /// BOUNDARY every reader copes -- absolute offsets in the index,
- /// `disk_poll_start` walking on into later segments, a contiguity guard that
- /// only looks within one file.
+ /// a segment from its FILENAME with a running `expected_offset` and REFUSES
+ /// at the first offset that does not continue it (`OffsetDiscontinuity`),
+ /// which on a solo group tombstones the partition. A surviving index does not
+ /// help: a first entry that is not the file-name offset makes recovery
+ /// discard the index and walk from byte 0, reaching the same refusal. On a
+ /// segment BOUNDARY every reader copes -- absolute offsets in the index,
+ /// `disk_poll_start` walking on into later segments, and a chain guard that
+ /// admits a forward gap the reservation covers.
///
- /// So an empty tail is unlinked (its name claims a range it does not hold)
- /// and a sized tail, the only copy of its messages, is sealed with a fresh
- /// segment planted at the frontier. An empty chain is left to the caller's
- /// `ensure_initial_segment`.
+ /// So an empty tail is unlinked (its name claims a range it does not hold), a
+ /// sized tail (the only copy of its messages) is sealed with a fresh segment
+ /// planted at the append point, and a chain the unlinks emptied is planted
+ /// directly -- `ensure_initial_segment` names its segment for the COMMITTED
+ /// frontier and would put the first mint inside it.
///
/// # Errors
/// [`IggyError`] when the fresh segment cannot be created, leaving the
@@ -4458,7 +4643,7 @@ where
// Where the next append will land: the counter, or an armed mint floor
// above it. The floor is the whole reason a hole can appear, so
// anchoring to the counter alone would leave the chain as unprepared.
- let frontier = self.offset_frontier().max(self.armed_mint_floor());
+ let frontier = self.mint_frontier();
if frontier == 0 {
return Ok(());
}
@@ -4479,14 +4664,22 @@ where
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
- warn!(
+ // Refused, not logged. The segment is already out of the
+ // in-memory chain, so a file left behind becomes a
+ // non-tail empty segment as soon as the plant lands --
+ // `[sized][stale empty][planted]` -- which the next boot
+ // refuses outright as `EmptyNonTailSegment`. Failing boot
+ // here says so while the directory is still readable.
+ error!(
target: "iggy.partitions.diag",
plane = "partitions",
namespace_raw = namespace.inner(),
path = %path,
%error,
- "failed to unlink a stale empty segment during the boot re-anchor"
+ "failed to unlink a stale empty segment during the boot \
+ re-anchor; refusing to plant beside it"
);
+ return Err(IggyError::CannotDeleteFile);
}
}
}
@@ -4498,12 +4691,16 @@ where
offset_frontier = frontier,
"unlinked an empty segment named below the restored offset frontier"
);
+ // Boot DOES count the recovered chain -- `load_persisted_segments`
+ // increments per segment before it looks at the size, so empty tails
+ // are in the total -- and retention pairs its own retire with a
+ // decrement. Without this the count stays one high on the wire for
+ // the life of the process.
+ self.stats.decrement_segments_count(1);
retired += 1;
}
// Durable before anything is planted beside them: a crash in between
- // would boot the stale name back into the chain. No stats decrement to
- // pair with the retire -- boot never counted the recovered chain, and
- // `fetch_sub` does not saturate.
+ // would boot the stale name back into the chain.
if retired > 0
&& let Some(partition_dir) = self.partition_dir.clone()
&& let Err(error) = crate::state_transfer::fsync_dir(&partition_dir).await
@@ -4517,32 +4714,106 @@ where
"boot re-anchor could not fsync the partition dir after unlinking"
);
}
- // Only a SIZED tail: an empty one either just went, or is already named
- // at the frontier and can take the appends as it is.
- let needs_plant = self.log.segments().last().is_some_and(|segment| {
- segment.size.as_bytes_u64() > 0 && segment.end_offset + 1 < frontier
- });
- if !needs_plant {
- return Ok(());
+ // Bounds copied out: the plant below takes `&mut self`, so the borrow on
+ // the chain cannot still be live.
+ let tail = self
+ .log
+ .segments()
+ .last()
+ .map(|segment| (segment.start_offset, segment.end_offset, segment.size));
+ match tail {
+ // An EMPTIED chain still needs the plant, and it cannot be left to
+ // the caller's `ensure_initial_segment`, which names the segment for
+ // the COMMITTED frontier and knows nothing of the append point. On
+ // the shape a crash before the first flush leaves -- committed
+ // frontier 0, append point a lease block up -- that plants
+ // `0.log` and then mints inside it, the hole this function exists to
+ // prevent. The index does not save it either: a first entry that is
+ // not the file-name offset makes recovery discard the index and walk
+ // from byte 0, where the discontinuity tombstones the partition.
+ //
+ // No anchor: with nothing before it the plant leaves no gap, so the
+ // chain guard has no pair to judge.
+ None => {
+ self.install_empty_segment(config, frontier).await?;
+ self.stats.increment_segments_count(1);
+ tracing::info!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ offset_frontier = frontier,
+ "planted a fresh segment at the restored offset frontier over an \
+ empty recovered chain"
+ );
+ }
+ // Only a SIZED tail: an empty one either just went, or is already
+ // named at the frontier and can take the appends as it is.
+ Some((sealed_start, sealed_end, size))
+ if size.as_bytes_u64() > 0 && sealed_end.saturating_add(1) < frontier =>
+ {
+ self.record_reanchor_gap(frontier, sealed_start, sealed_end)
+ .await?;
+ self.rotate_segment_at(config, frontier).await?;
+ tracing::info!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ sealed_end,
+ offset_frontier = frontier,
+ "sealed the recovered tail and planted a fresh segment at the \
+ restored offset frontier"
+ );
+ }
+ Some(_) => {}
}
- let sealed_index = self.log.segments().len() - 1;
- let sealed_end = self.log.active_segment().end_offset;
- self.log.active_segment_mut().sealed = true;
- let sealed_storage = &mut self.log.storages_mut()[sealed_index];
- let _ = sealed_storage.shutdown();
- self.log.messages_writers_mut()[sealed_index] = None;
- self.log.index_writers_mut()[sealed_index] = None;
- self.log.indexes_mut()[sealed_index] = None;
- self.install_empty_segment(config, frontier).await?;
- self.stats.increment_segments_count(1);
- tracing::info!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- namespace_raw = namespace.inner(),
+ Ok(())
+ }
+
+ /// Write the anchor that makes the gap a plant at `frontier` leaves
+ /// legitimate, and make it durable before the segment exists.
+ ///
+ /// The chain guard admits a forward gap only when the far side carries an
+ /// anchor naming exactly the near side, so this record is what separates the
+ /// re-anchor's own gap from a lost segment. Ordering is load-bearing in one
+ /// direction only: an anchor with no segment is swept at the next boot,
+ /// while a segment with no anchor reads as damage.
+ ///
+ /// # Errors
+ /// [`IggyError::CannotCreateSegmentLogFile`] naming the anchor path. The
+ /// caller must not plant.
+ #[allow(clippy::future_not_send)]
+ async fn record_reanchor_gap(
+ &self,
+ frontier: u64,
+ sealed_start: u64,
+ sealed_end: u64,
+ ) -> Result<(), IggyError> {
+ // No directory means an in-memory partition, whose chain no boot reads.
+ let Some(partition_dir) = self.partition_dir() else {
+ return Ok(());
+ };
+ let anchor = crate::segment_anchor::SegmentAnchor {
+ sealed_start,
sealed_end,
- offset_frontier = frontier,
- "sealed the recovered tail and planted a fresh segment at the restored frontier"
- );
+ };
+ if let Err(error) =
+ crate::segment_anchor::write_anchor(&partition_dir, frontier, anchor).await
+ {
+ error!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ offset_frontier = frontier,
+ sealed_start,
+ sealed_end,
+ %error,
+ "could not record the boot re-anchor's gap; refusing to plant a segment \
+ the next boot would read as a lost one"
+ );
+ return Err(IggyError::CannotCreateSegmentLogFile(
+ crate::segment_anchor::anchor_path(&partition_dir, frontier),
+ ));
+ }
Ok(())
}
@@ -5725,7 +5996,7 @@ mod tests {
#[compio::test]
async fn given_appends_inside_the_block_when_fencing_should_write_the_superblock_once() {
let store = Rc::new(RecordingSuperblock::default());
- let mut partition = partition_at_view(1, 1);
+ let mut partition = solo_recording_partition();
partition.set_superblock(store.clone(), None);
partition.set_offset_reservation_lease(16);
@@ -5760,7 +6031,7 @@ mod tests {
#[compio::test]
async fn given_failing_superblock_when_fencing_should_refuse() {
let store = Rc::new(RecordingSuperblock::default());
- let mut partition = partition_at_view(1, 1);
+ let mut partition = solo_recording_partition();
partition.set_superblock(store.clone(), None);
partition.set_offset_reservation_lease(4);
store.fail_writes.set(true);
@@ -5771,74 +6042,192 @@ mod tests {
);
}
- /// The reservation floors the next MINT, and only on a solo group. The whole
- /// fix: after a crash below the flush thresholds it is the only witness that
- /// those offsets were confirmed.
+ /// The point of extending from the tick: after it runs, the append path finds
+ /// the ceiling already covering it and writes nothing. Without this the two
+ /// fsyncs land in front of a produce, inside the frame pump the consensus
+ /// tick shares.
+ #[compio::test]
+ async fn given_a_consumed_block_when_the_tick_extends_should_leave_the_append_path_writeless() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = solo_recording_partition();
+ partition.set_superblock(store.clone(), None);
+ partition.set_offset_reservation_lease(16);
+
+ // First append pays for the first claim, as it must: nothing durable yet.
+ assert!(partition.reserve_offsets_through(0).await);
+ assert_eq!(store.attempts.get(), 1);
+ partition.should_increment_offset = true;
+
+ // Inside the block there is nothing to do, from either caller.
+ partition.dirty_offset.store(4, Ordering::Relaxed);
+ assert!(
+ !partition.needs_offset_reservation_extension(),
+ "a full block of headroom needs no extension"
+ );
+
+ // Past half the block the tick takes the write.
+ partition.dirty_offset.store(11, Ordering::Relaxed);
+ assert!(
+ partition.needs_offset_reservation_extension(),
+ "under half a block of headroom the tick must extend"
+ );
+ assert!(partition.extend_offset_reservation().await);
+ assert_eq!(
+ store.attempts.get(),
+ 2,
+ "the tick wrote, not the append path"
+ );
+
+ // And now the append path is writeless across the rest of the old block.
+ let before = store.attempts.get();
+ for offset in 12..=16 {
+ assert!(partition.reserve_offsets_through(offset).await);
+ }
+ assert_eq!(
+ store.attempts.get(),
+ before,
+ "every append after the extension must take the fence's fast path"
+ );
+ }
+
+ /// The extension must not fire for a partition that has never minted, or boot
+ /// would write a superblock per idle partition for nothing.
+ #[test]
+ fn given_an_untouched_partition_when_ticking_should_not_extend_the_reservation() {
+ let mut partition = solo_recording_partition();
+ partition.set_superblock(Rc::new(RecordingSuperblock::default()), None);
+ assert!(!partition.should_increment_offset);
+ assert!(!partition.needs_offset_reservation_extension());
+ }
+
+ /// A boot that consumed a reservation has ZERO headroom -- the append point
+ /// sits exactly at the ceiling -- so the extension has to fire before the
+ /// first produce rather than after it.
+ #[test]
+ fn given_a_reservation_seeded_boot_when_ticking_should_extend_before_the_first_produce() {
+ let mut partition = solo_recording_partition();
+ let recovered = recorded_state(0, 65_537);
+ partition.set_superblock(Rc::new(RecordingSuperblock::default()), Some(&recovered));
+ partition.restore_offset_frontier(Some(&recovered));
+ assert_eq!(partition.mint_frontier(), 65_537);
+ assert!(
+ partition.needs_offset_reservation_extension(),
+ "the seeded append point is at the ceiling, so the next append would \
+ otherwise pay for the write"
+ );
+ }
+
+ /// A replicated group pays nothing for a protection it cannot use: nothing
+ /// seeds its counter from the reservation, its chain never gets re-anchored,
+ /// and an ack there means a quorum journaled the batch.
+ #[compio::test]
+ async fn given_a_replicated_group_when_fencing_should_not_write_at_all() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = partition_at_view(1, 1);
+ assert!(partition.consensus().replica_count() > 1);
+ partition.set_superblock(store.clone(), None);
+ partition.set_offset_reservation_lease(16);
+ store.fail_writes.set(true);
+
+ assert!(
+ partition.reserve_offsets_through(1_000).await,
+ "the fence is inert above one replica, so not even a failing store can \
+ refuse the append"
+ );
+ assert_eq!(store.attempts.get(), 0, "and it attempts no write");
+ }
+
+ /// The reservation seeds the APPEND counter and only on a solo group. The
+ /// whole fix: after a crash below the flush thresholds it is the only witness
+ /// that those offsets were confirmed.
#[test]
- fn given_a_recorded_reservation_when_solo_should_floor_the_next_mint() {
+ fn given_a_recorded_reservation_when_solo_should_seed_the_append_counter() {
let mut partition = solo_recording_partition();
assert_eq!(
partition.consensus().replica_count(),
1,
- "the floor is scoped to solo groups"
+ "the reservation seed is scoped to solo groups"
);
let store = Rc::new(RecordingSuperblock::default());
partition.set_superblock(store.clone(), None);
+ partition.restore_offset_frontier(None);
assert_eq!(
- partition.mint_floor(),
+ partition.mint_frontier(),
0,
- "nothing recorded, nothing floored"
+ "nothing recorded, nothing seeded"
);
- partition.set_superblock(store, Some(&recorded_state(0, 65_537)));
+ // The shape a crash before the first flush leaves: the fence recorded a
+ // block, and no message ever reached a segment, so the frontier is 0.
+ let recovered = recorded_state(0, 65_537);
+ partition.set_superblock(store, Some(&recovered));
+ partition.restore_offset_frontier(Some(&recovered));
assert_eq!(
- partition.mint_floor(),
+ partition.mint_frontier(),
65_537,
- "the first mint of the incarnation must land above every offset the \
- reservation covered"
+ "the first mint must land above every offset the reservation covered"
);
assert_eq!(
- partition.mint_floor(),
- 0,
- "and only the first: after that the counter is the better number, and a \
- floor applied again would jump to the top of the block every time"
+ partition.offset_frontier(),
+ 1,
+ "the committed frontier must not inherit the reservation: nothing was \
+ flushed, so publishing it would claim a lease block of data"
);
}
- /// A replicated group must not take the jump on its own: a backup rejects any
- /// prepare whose `base_offset` does not continue its own counter.
+ /// A replicated group must not take the jump: a backup rejects any prepare
+ /// whose `base_offset` does not continue its own counter, so an append point a
+ /// lease block above the group has every peer refuse the batch.
#[test]
- fn given_a_recorded_reservation_when_replicated_should_not_floor_the_mint() {
+ fn given_a_recorded_reservation_when_replicated_should_not_seed_the_append_counter() {
let mut partition = partition_at_view(1, 1);
assert!(partition.consensus().replica_count() > 1);
- partition.set_superblock(
- Rc::new(RecordingSuperblock::default()),
- Some(&recorded_state(0, 70_000)),
- );
+ let recovered = recorded_state(0, 70_000);
+ partition.set_superblock(Rc::new(RecordingSuperblock::default()), Some(&recovered));
+ partition.restore_offset_frontier(Some(&recovered));
assert_eq!(
- partition.mint_floor(),
+ partition.mint_frontier(),
0,
"a replicated group's offsets are the group's to decide"
);
}
- /// The rewind fence asks whether an offer would destroy something this
- /// replica has, and a counter above every message it holds -- what a mint
- /// floor leaves behind until the first append -- is not something it has.
+ /// The committed frontier still seeds both counters: it is a claim about data
+ /// every replica shares, so it is not gated on the replica count.
+ #[test]
+ fn given_a_recorded_frontier_when_replicated_should_seed_both_counters() {
+ let mut partition = partition_at_view(1, 1);
+ assert!(partition.consensus().replica_count() > 1);
+ let recovered = recorded_state(40, 70_000);
+ partition.set_superblock(Rc::new(RecordingSuperblock::default()), Some(&recovered));
+ partition.restore_offset_frontier(Some(&recovered));
+ assert_eq!(partition.offset_frontier(), 40);
+ assert_eq!(
+ partition.mint_frontier(),
+ 40,
+ "the reservation is ignored here, so the append point is the frontier"
+ );
+ }
+
+ /// The rewind fence asks whether an offer would destroy something this replica
+ /// COMMITTED. An append point standing a lease block above that -- what a
+ /// reservation-seeded boot leaves until the first append -- claims no
+ /// messages, so it must not turn a legitimate offer inside the block into a
+ /// refusal the replica would then cycle on forever.
#[compio::test]
- async fn given_a_counter_above_every_message_when_an_offer_arrives_should_not_call_it_a_rewind()
- {
- let partition_dir = transfer_fence_dir("counter-is-not-data").await;
+ async fn given_an_append_point_above_every_message_when_an_offer_arrives_should_not_call_it_a_rewind()
+ {
+ let partition_dir = transfer_fence_dir("append-point-is-not-data").await;
let mut partition = test_partition();
partition.set_partition_dir(partition_dir.clone());
partition.should_increment_offset = true;
- partition.offset.store(69_999, Ordering::Release);
- assert_eq!(partition.offset_frontier(), 70_000);
+ partition.dirty_offset.store(69_999, Ordering::Relaxed);
+ assert_eq!(partition.mint_frontier(), 70_000);
assert_eq!(
- partition.held_offset_frontier(),
- 0,
- "no segment and no journal entry: the counter speaks for no messages"
+ partition.offset_frontier(),
+ 1,
+ "nothing committed: the append point speaks for no messages"
);
let offer = crate::state_transfer::ConsumerOffsetsWire {
@@ -5871,8 +6260,10 @@ mod tests {
let store = Rc::new(RecordingSuperblock::default());
let mut partition = solo_recording_partition();
partition.set_superblock(store.clone(), Some(&recorded_state(0, 65_537)));
+ // A flushed partition: the append point and the committed head agree.
partition.should_increment_offset = true;
partition.offset.store(24, Ordering::Release);
+ partition.dirty_offset.store(24, Ordering::Relaxed);
assert!(partition.collapse_offset_reservation().await);
assert_eq!(last_recorded_frontier(&store), 25);
@@ -5883,13 +6274,45 @@ mod tests {
);
// And the restart that follows mints where it left off, not a block up.
+ let recovered = recorded_state(25, 25);
let mut restarted = solo_recording_partition();
- restarted.set_superblock(store.clone(), Some(&recorded_state(25, 25)));
+ restarted.set_superblock(store.clone(), Some(&recovered));
+ restarted.restore_offset_frontier(Some(&recovered));
assert_eq!(
- restarted.mint_floor(),
+ restarted.mint_frontier(),
25,
- "the floor is the frontier itself, so it moves nothing"
+ "the collapsed record puts the append point back at the frontier"
+ );
+ }
+
+ /// The graceful stop must not undo the crash protection. A boot that read a
+ /// reservation back and then took no traffic has a committed frontier of 0
+ /// while the reservation is the only record that offsets were confirmed, so a
+ /// collapse reading the committed frontier would write it away -- and a clean
+ /// stop is the runbook answer to an incident, which would make it the one
+ /// action that re-opens the defect.
+ #[compio::test]
+ async fn given_an_unspent_reservation_when_collapsing_should_leave_it_standing() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let recovered = recorded_state(0, 65_537);
+ let mut partition = solo_recording_partition();
+ partition.set_superblock(store.clone(), Some(&recovered));
+ partition.restore_offset_frontier(Some(&recovered));
+
+ assert!(partition.collapse_offset_reservation().await);
+ assert_eq!(
+ store.attempts.get(),
+ 0,
+ "the reservation already covers the append point, so there is nothing to \
+ collapse and nothing to write"
);
+
+ // The restart after the clean stop still resumes above every offset the
+ // crashed incarnation confirmed.
+ let mut restarted = solo_recording_partition();
+ restarted.set_superblock(store, Some(&recovered));
+ restarted.restore_offset_frontier(Some(&recovered));
+ assert_eq!(restarted.mint_frontier(), 65_537);
}
/// An install is the one place the reservation may come down: left high, it
@@ -7852,8 +8275,8 @@ mod tests {
partition.set_partition_dir(partition_dir.clone());
partition.should_increment_offset = true;
partition.offset.store(99, Ordering::Release);
- // The fence reads held messages, not the counter, which is also what a
- // reservation-seeded boot leaves behind.
+ // Committed and resident: the threshold-gated flush leaves exactly this
+ // shape, and the fence has to count it as data.
{
let info = &mut partition.log.journal_mut().info;
info.messages_count = 100;
@@ -7907,6 +8330,52 @@ mod tests {
let _ = std::fs::remove_dir_all(&partition_dir);
}
+ /// A chain installed EMPTY at frontier N holds no sized segment and no
+ /// journal entry, so a fence reading only held bytes reads 0 and skips
+ /// itself, letting a stale offer rewind the counter under offsets this
+ /// replica already claimed. The committed frontier is what carries N.
+ #[compio::test]
+ async fn given_an_empty_chain_installed_at_a_frontier_when_a_stale_offer_arrives_should_refuse()
+ {
+ let partition_dir = transfer_fence_dir("empty-install-rewind").await;
+ let mut partition = test_partition();
+ partition.set_partition_dir(partition_dir.clone());
+ // What an install of an all-GC'd origin leaves: the counter at the group
+ // frontier, nothing on disk, nothing resident.
+ partition.should_increment_offset = true;
+ partition.offset.store(4_095, Ordering::Release);
+ partition.dirty_offset.store(4_095, Ordering::Relaxed);
+ assert_eq!(
+ partition.held_offset_frontier(),
+ 4_096,
+ "the committed arm has to carry a frontier no byte on disk names"
+ );
+
+ let stale = crate::state_transfer::ConsumerOffsetsWire {
+ purge_generation: 0,
+ next_offset: 1_000,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ };
+ let refused = partition
+ .install_state_transfer(&repair_config(), 12, Vec::new(), &stale.encode(), 0)
+ .await;
+ assert!(
+ matches!(
+ refused,
+ Err(
+ crate::state_transfer::PartitionInstallError::OfferRewindsDurableData {
+ offer_next_offset: 1_000,
+ local_next_offset: 4_096,
+ }
+ )
+ ),
+ "expected a rewind refusal, got {refused:?}"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
+
/// The canonical post-restart rejoin: this replica applied a purge before
/// the restart but was killed before the purge's `purge.gen` record step,
/// so the metadata plane's COMMITTED generation is 1 while its own
@@ -7920,8 +8389,8 @@ mod tests {
partition.set_partition_dir(partition_dir.clone());
partition.should_increment_offset = true;
partition.offset.store(99, Ordering::Release);
- // The fence reads held messages, not the counter, which is also what a
- // reservation-seeded boot leaves behind.
+ // Committed and resident: the threshold-gated flush leaves exactly this
+ // shape, and the fence has to count it as data.
{
let info = &mut partition.log.journal_mut().info;
info.messages_count = 100;
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 818872029..a0246b246 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -28,6 +28,7 @@ mod messages_writer;
pub mod offset_storage;
mod poll_plan;
mod segment;
+pub mod segment_anchor;
pub mod state_transfer;
mod types;
diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs
index b089d6551..64ae2f520 100644
--- a/core/partitions/src/log.rs
+++ b/core/partitions/src/log.rs
@@ -192,8 +192,9 @@ where
}
/// Mutable segment views. Length mutation lives in
- /// [`Self::add_persisted_segment`] / [`Self::retire_front`] only, so the
- /// parallel vecs cannot desync from the outside.
+ /// [`Self::add_persisted_segment`] / [`Self::retire_front`] /
+ /// [`Self::retire_back`] only, so the parallel vecs cannot desync from the
+ /// outside.
pub fn segments_mut(&mut self) -> &mut [Segment] {
&mut self.segments
}
diff --git a/core/partitions/src/segment_anchor.rs b/core/partitions/src/segment_anchor.rs
new file mode 100644
index 000000000..70875f4fc
--- /dev/null
+++ b/core/partitions/src/segment_anchor.rs
@@ -0,0 +1,205 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The record that makes a gap in the segment chain legitimate.
+//!
+//! Recovery derives each segment's bounds from its own bytes, so a gap has no
+//! author: the boot re-anchor's planted gap and a lost segment look identical.
+//! The re-anchor writes its intent down instead, and the chain guard admits a
+//! forward gap only when the far side carries an anchor naming exactly the near
+//! side.
+//!
+//! Written and directory-fsynced BEFORE that segment is created. A crash in the
+//! window then leaves an anchor with no segment, which the boot sweep collects;
+//! the other order leaves a planted segment with no anchor, which the guard
+//! reads as damage on an intact chain.
+
+use compio::io::AsyncWriteAtExt;
+use consensus::state_artifact_checksum;
+use std::io;
+
+/// File extension for an anchor record, `{start_offset:020}.anchor` beside the
+/// `{start_offset:020}.log` it belongs to.
+pub const ANCHOR_EXTENSION: &str = "anchor";
+
+/// Leading bytes of an anchor record, so a file that is not one (a truncated
+/// write, an operator's copy) is refused rather than decoded.
+const ANCHOR_MAGIC: u64 = u64::from_le_bytes(*b"IGGYANCH");
+
+/// `magic`(8) + `sealed_start`(8) + `sealed_end`(8) + `checksum`(8).
+pub const ANCHOR_ENCODED_LEN: usize = 32;
+
+/// The gap one planted segment is allowed to leave behind it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SegmentAnchor {
+ /// Start offset of the segment that was sealed, i.e. the near side of the
+ /// gap. Names WHICH segment, so an anchor cannot be satisfied by a
+ /// different file that happens to end where this one expects.
+ pub sealed_start: u64,
+ /// End offset the sealed segment held when it was sealed. The gap runs from
+ /// here to the planted segment's own start offset.
+ pub sealed_end: u64,
+}
+
+impl SegmentAnchor {
+ /// Encode to the fixed little-endian on-disk layout.
+ #[must_use]
+ pub fn to_bytes(&self) -> [u8; ANCHOR_ENCODED_LEN] {
+ let mut out = [0u8; ANCHOR_ENCODED_LEN];
+ out[0..8].copy_from_slice(&ANCHOR_MAGIC.to_le_bytes());
+ out[8..16].copy_from_slice(&self.sealed_start.to_le_bytes());
+ out[16..24].copy_from_slice(&self.sealed_end.to_le_bytes());
+ let checksum = state_artifact_checksum(&out[0..24]);
+ out[24..32].copy_from_slice(&checksum.to_le_bytes());
+ out
+ }
+
+ /// Decode a record, returning `None` for anything this build did not write:
+ /// a wrong length, a wrong magic, or a checksum that does not match.
+ ///
+ /// A `None` is never treated as "no gap was intended". It means the record
+ /// proves nothing, so the gap it would have covered stays damage.
+ #[must_use]
+ pub fn from_bytes(bytes: &[u8]) -> Option<Self> {
+ if bytes.len() != ANCHOR_ENCODED_LEN {
+ return None;
+ }
+ let field = |at: usize| -> u64 {
+ let mut raw = [0u8; 8];
+ raw.copy_from_slice(&bytes[at..at + 8]);
+ u64::from_le_bytes(raw)
+ };
+ if field(0) != ANCHOR_MAGIC || field(24) != state_artifact_checksum(&bytes[0..24]) {
+ return None;
+ }
+ Some(Self {
+ sealed_start: field(8),
+ sealed_end: field(16),
+ })
+ }
+
+ /// Whether this anchor legitimises the gap between the segment starting at
+ /// `sealed_start` / ending at `sealed_end` and the segment it sits beside.
+ ///
+ /// BOTH bounds must match. Matching only the end offset would let an anchor
+ /// left by an earlier incarnation of the chain cover a gap it never saw.
+ #[must_use]
+ pub const fn covers(&self, sealed_start: u64, sealed_end: u64) -> bool {
+ self.sealed_start == sealed_start && self.sealed_end == sealed_end
+ }
+}
+
+/// Path of the anchor record beside the segment starting at `start_offset`.
+#[must_use]
+pub fn anchor_path(partition_dir: &str, start_offset: u64) -> String {
+ format!("{partition_dir}/{start_offset:0>20}.{ANCHOR_EXTENSION}")
+}
+
+/// Write the anchor for a segment about to be planted at `start_offset`, then
+/// fsync the directory so the record cannot arrive after the segment it
+/// describes.
+///
+/// # Errors
+///
+/// Any I/O failure. The caller must NOT plant the segment: a planted segment
+/// whose anchor is missing reads as damage on the next boot.
+pub async fn write_anchor(
+ partition_dir: &str,
+ start_offset: u64,
+ anchor: SegmentAnchor,
+) -> io::Result<()> {
+ let path = anchor_path(partition_dir, start_offset);
+ let mut file = compio::fs::File::create(&path).await?;
+ let (result, _buf) = file
+ .write_all_at(anchor.to_bytes().to_vec(), 0)
+ .await
+ .into();
+ result?;
+ file.sync_all().await?;
+ crate::state_transfer::fsync_dir(partition_dir).await
+}
+
+/// Read the anchor beside the segment starting at `start_offset`.
+///
+/// `None` when the file is absent or does not decode; both mean the same thing
+/// to the guard, so the caller needs no distinction.
+pub async fn read_anchor(partition_dir: &str, start_offset: u64) -> Option<SegmentAnchor> {
+ let path = anchor_path(partition_dir, start_offset);
+ let bytes = compio::fs::read(&path).await.ok()?;
+ SegmentAnchor::from_bytes(&bytes)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_an_anchor_when_round_tripped_should_decode_identically() {
+ let anchor = SegmentAnchor {
+ sealed_start: 7,
+ sealed_end: 4_095,
+ };
+ let bytes = anchor.to_bytes();
+ assert_eq!(bytes.len(), ANCHOR_ENCODED_LEN);
+ assert_eq!(SegmentAnchor::from_bytes(&bytes), Some(anchor));
+ }
+
+ #[test]
+ fn given_a_flipped_bit_when_decoded_should_refuse() {
+ let anchor = SegmentAnchor {
+ sealed_start: 7,
+ sealed_end: 4_095,
+ };
+ // Every byte the checksum covers, so a corrupted record can never read as
+ // a legitimate gap.
+ for index in 0..24 {
+ let mut bytes = anchor.to_bytes();
+ bytes[index] ^= 1;
+ assert_eq!(
+ SegmentAnchor::from_bytes(&bytes),
+ None,
+ "a record corrupted at byte {index} must not decode"
+ );
+ }
+ }
+
+ #[test]
+ fn given_a_wrong_length_when_decoded_should_refuse() {
+ let bytes = SegmentAnchor {
+ sealed_start: 1,
+ sealed_end: 2,
+ }
+ .to_bytes();
+ assert_eq!(SegmentAnchor::from_bytes(&bytes[..31]), None);
+ assert_eq!(SegmentAnchor::from_bytes(&[]), None);
+ }
+
+ #[test]
+ fn given_an_anchor_when_matching_a_different_predecessor_should_not_cover_it() {
+ let anchor = SegmentAnchor {
+ sealed_start: 10,
+ sealed_end: 20,
+ };
+ assert!(anchor.covers(10, 20));
+ assert!(!anchor.covers(10, 21), "a different end must not match");
+ assert!(
+ !anchor.covers(0, 20),
+ "the same end under a different segment must not match: an anchor left \
+ by an earlier chain would otherwise cover a gap it never saw"
+ );
+ }
+}
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 177dfced5..bfea0939a 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -1994,13 +1994,14 @@ where
let purge_advances = offsets_wire.purge_generation > committed_purge_generation
|| (self.applied_purge_generation < committed_purge_generation
&& offsets_wire.next_offset == 0);
- // What this replica HOLDS, not the append counter: the two diverge once
- // the counter is seeded from an offset reservation, which stands a lease
- // block above the last message anywhere and names none of them. Against
- // that, every legitimate offer inside the block reads as a rewind, and
- // the replica that restarted with a reservation is exactly the one
- // needing the transfer, so it would cycle refusal -> backoff forever.
- let local_next_offset = self.held_offset_frontier();
+ // The COMMITTED frontier, which is what an offer is comparable against:
+ // `held_offset_frontier` reads 0 for a chain installed empty at frontier
+ // N (its disk arm filters empty segments and the install clears the
+ // journal), and a 0 skips the guard below entirely, letting a stale offer
+ // rewind the counter under data this replica already claimed. The append
+ // point is not usable either -- it can stand a lease block high -- but
+ // only on a solo group, which never receives an offer.
+ let local_next_offset = self.offset_frontier();
if !purge_advances && local_next_offset > 0 && offsets_wire.next_offset < local_next_offset
{
return Err(PartitionInstallError::OfferRewindsDurableData {
diff --git a/core/server/config.toml b/core/server/config.toml
index 52689fadb..e239efce2 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -979,6 +979,10 @@ prepare_queue_depth = 32
# message. One superblock write (two fsyncs) per block: lowering it raises the
# fsync rate on the write path, raising it wastes at most one block of the u64
# offset space per crash. Must be > 0 and <= 16777216.
+#
+# SINGLE-REPLICA groups only. A replicated group acks a send once a quorum has
+# journaled it, so re-minting there needs a full-cluster crash; it claims
+# nothing, pays no write, and ignores this value.
offset_reservation_lease = 65536
# Entries the evicted ring retains per multi-replica partition for journal
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 1f4b54c90..48af4de48 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -2593,7 +2593,6 @@ async fn recover_partition_segments(
namespace: IggyNamespace,
runtime_options: TopicRuntimeOptions,
stats: &PartitionStats,
- offset_reserved: u64,
) -> Result<Vec<RecoveredSegment>, ServerError> {
let stream_id = namespace.stream_id();
let topic_id = namespace.topic_id();
@@ -2604,25 +2603,18 @@ async fn recover_partition_segments(
let enforce_fsync = runtime_options
.enforce_fsync
.unwrap_or(iggy_common::DEFAULT_ENFORCE_FSYNC);
- load_persisted_segments(
- config,
- namespace,
- segment_size,
- enforce_fsync,
- stats,
- offset_reserved,
- )
- .await
- .map_err(|source| {
- error!(
- stream_id,
- topic_id,
- partition_id,
- error = %source,
- "failed to load partition log during server bootstrap"
- );
- source
- })
+ load_persisted_segments(config, namespace, segment_size, enforce_fsync, stats)
+ .await
+ .map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ error = %source,
+ "failed to load partition log during server bootstrap"
+ );
+ source
+ })
}
#[allow(clippy::too_many_arguments)]
@@ -2705,13 +2697,8 @@ async fn load_partition(
// recovered message timestamp here, or an NTP rewind across a restart could
// regress persisted `base_timestamp`.
- // The chain guard needs the claimed ceiling to tell the boot re-anchor's
- // own gap from damage, and the superblock is already open by here.
- let reserved = recovered_state
- .as_ref()
- .map_or(0, |state| state.offset_reserved);
let recovered_segments =
- recover_partition_segments(config, namespace, runtime_options, &stats, reserved).await?;
+ recover_partition_segments(config, namespace, runtime_options, &stats).await?;
let mut partition = IggyPartition::new(stats.clone(), consensus);
partition.set_runtime_options(runtime_options);
@@ -2805,12 +2792,13 @@ async fn restore_partition_offsets(
// taking the max means real recovered data always wins.
partition.restore_offset_frontier(recovered_state);
// Minting from the reservation leaves a hole between the recovered chain
- // and the new append point, and the recovery walk truncates at a hole
- // INSIDE a segment, so put it on a segment boundary instead.
+ // and the new append point, and the recovery walk REFUSES a hole inside a
+ // segment (tombstoning the partition on the solo arm), so put it on a
+ // segment boundary instead.
//
- // Solo only, in step with the floor itself: a replicated group's segment
- // boundaries must be a function of the batches alone or the reconciler's
- // offset-keyed segment GC never converges.
+ // Solo only, in step with the reservation itself: a replicated group's
+ // segment boundaries must be a function of the batches alone or the
+ // reconciler's offset-keyed segment GC never converges.
if partition.consensus().replica_count() == 1 {
partition
.reanchor_to_offset_frontier(partitions_config)
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index b150205b1..ca6d65812 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -320,7 +320,7 @@ pub async fn ensure_initial_segment(
// `rposition(|s| s.start_offset <= offset)` routes every poll for `0..N-1`
// into it, the next boot makes that shape durable, and this replica starts
// offering peers a segment that claims `[0..N]`.
- let start_offset = partition.offset_frontier();
+ let start_offset = partition.mint_frontier();
let messages_path =
config
.system
diff --git a/core/server/src/segment_recovery.rs b/core/server/src/segment_recovery.rs
index afeb520ed..8a1c81ed8 100644
--- a/core/server/src/segment_recovery.rs
+++ b/core/server/src/segment_recovery.rs
@@ -30,16 +30,18 @@
use crate::server_error::{PartitionRecoveryRefusal, ServerError};
use configs::server::ServerConfig;
use iggy_common::{IggyByteSize, IggyError, MAX_MESSAGE_SIZE_UPPER_BYTES, PartitionStats};
+use partitions::segment_anchor::{ANCHOR_EXTENSION, SegmentAnchor};
use partitions::state_transfer::STAGING_SUFFIX;
use partitions::{IggyIndex, IggyIndexReader, Segment};
use server_common::send_messages::{BatchHeader, COMMAND_HEADER_SIZE, decode_batch_slice};
use server_common::sharding::IggyNamespace;
use server_common::{SegmentStorage, yield_to_reactor};
+use std::collections::HashMap;
use std::fs;
use std::io;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
-use tracing::{error, warn};
+use tracing::{error, info, warn};
const LOG_EXTENSION: &str = "log";
const INDEX_EXTENSION: &str = "index";
@@ -174,6 +176,10 @@ pub struct RecoveredSegment {
/// makes a durable index entry evidence about the log (see
/// [`PartitionRecoveryRefusal::FsyncedLogLoss`]), so passing it wrong either
/// refuses healthy chains or hides previously durable data loss.
+///
+/// Takes no offset ceiling. A legitimate gap is proved by the anchor the boot
+/// re-anchor writes beside the segment it plants, not inferred from how far the
+/// superblock's reservation happens to reach.
#[allow(clippy::too_many_lines)]
pub async fn load_persisted_segments(
config: &ServerConfig,
@@ -181,7 +187,6 @@ pub async fn load_persisted_segments(
segment_size: IggyByteSize,
enforce_fsync: bool,
stats: &PartitionStats,
- offset_reserved: u64,
) -> Result<Vec<RecoveredSegment>, ServerError> {
let stream_id = namespace.stream_id();
let topic_id = namespace.topic_id();
@@ -301,9 +306,12 @@ pub async fn load_persisted_segments(
last.segment.sealed = false;
}
- // Pass B: the chain guard reads only the planned bounds, so it can refuse
- // BEFORE anything is truncated.
- ensure_contiguous_chain(identity, &planned, offset_reserved)?;
+ // Pass B: the chain guard reads the planned bounds and the anchors beside
+ // them, so it can refuse BEFORE anything is truncated. Only gaps need an
+ // anchor, so the reads are skipped entirely for a contiguous chain, which is
+ // every chain that never crashed mid-block.
+ let anchors = read_segment_anchors(&partition_path, &planned).await;
+ ensure_contiguous_chain(identity, &planned, &anchors)?;
// Pass C: the chain is accepted; make disk match the bounds and open
// storage over them.
@@ -526,15 +534,57 @@ struct ScanScratch {
probe_budget: ProbeBudget,
}
+/// Read the anchor beside the far side of every gap in `planned`.
+///
+/// Keyed by the planted segment's start offset. Only gaps are looked up: a
+/// contiguous chain needs no anchors, and reading one per segment would put a
+/// file open on the boot path of every partition for nothing.
+async fn read_segment_anchors(
+ partition_path: &str,
+ planned: &[PlannedSegment],
+) -> HashMap<u64, SegmentAnchor> {
+ let mut anchors = HashMap::new();
+ for pair in planned.windows(2) {
+ let (previous, next) = (&pair[0].segment, &pair[1].segment);
+ if previous.end_offset.checked_add(1) == Some(next.start_offset) {
+ continue;
+ }
+ if let Some(anchor) =
+ partitions::segment_anchor::read_anchor(partition_path, next.start_offset).await
+ {
+ anchors.insert(next.start_offset, anchor);
+ }
+ }
+ anchors
+}
+
/// Contiguity guard: recovery takes every `.log` stem in the directory, so a
/// stray file (an unlink a failed state-transfer install could not finish,
/// 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.
///
-/// `offset_reserved` is the superblock's ceiling on offsets this replica may
-/// already have minted, and it is what separates the boot re-anchor's own
-/// gap from damage. See the hole arm below.
+/// A FORWARD gap is admitted only when the far side carries a
+/// [`SegmentAnchor`] naming exactly the near side, which is the record the boot
+/// re-anchor writes before it plants. Nothing else legitimises a gap: an
+/// overlap, a backwards pair, or a gap with no anchor is damage.
+///
+/// The anchors are what a monotone offset ceiling could not be. A ceiling says
+/// only "some boot claimed up to N", and every plant base sits below the current
+/// N -- but so does the successor of a segment that was deleted, so a lost middle
+/// segment read as a plant. The anchor is written by the one component that
+/// creates legitimate gaps, names which segment it sealed, and is swept as soon
+/// as its segment is gone.
+///
+/// Retention needs no allowance: it removes a contiguous FRONT prefix, so the
+/// remaining chain stays contiguous and no interior gap appears.
+///
+/// # Known residual
+///
+/// An anchor whose planted segment survived while the sealed segment it names was
+/// itself lost still reads as legitimate, because the pair the anchor describes
+/// is then simply absent from the chain and the guard never examines it. Catching
+/// that needs a durable count of the chain, which this record does not carry.
///
/// Runs on the planned bounds alone, BEFORE any truncation, so the segment
/// files a refusal quarantines are exactly the bytes boot found. The refusal
@@ -545,7 +595,7 @@ struct ScanScratch {
fn ensure_contiguous_chain(
identity: PartitionIdentity<'_>,
planned: &[PlannedSegment],
- offset_reserved: u64,
+ anchors: &HashMap<u64, SegmentAnchor>,
) -> Result<(), ServerError> {
// Walked, decodable bytes across the whole chain: the refusals carry it
// so the single-replica boot arm can tell a shape with nothing servable
@@ -575,18 +625,20 @@ fn ensure_contiguous_chain(
}
// `checked_add`, not `+`: an end offset at u64::MAX must read as a
// hole (no start offset can follow it), not overflow.
- //
- // A gap whose far side sits inside `offset_reserved` is the boot
- // re-anchor's, not damage: `reanchor_to_offset_frontier` seals the
- // recovered tail and plants the next segment at a frontier the
- // superblock already claimed, so the skipped offsets were minted (or
- // may have been) and no file was ever meant to hold them. Refusing it
- // would tombstone the partition on the boot AFTER the one that fixed
- // the re-mint. The ceiling is what bounds the leniency: a stray file
- // splicing a gap ABOVE anything this replica claimed still refuses.
- if previous.end_offset.checked_add(1) != Some(next.start_offset)
- && next.start_offset > offset_reserved
- {
+ if previous.end_offset.checked_add(1) == Some(next.start_offset) {
+ continue;
+ }
+ // FORWARD only, and only with the plant's own record beside it. Start
+ // offsets come off the file names so they ascend, but each end offset is
+ // walked from that file's own bytes with nothing clamping it against the
+ // next start, so a half-installed transfer or an operator copy can leave
+ // a pair that overlaps -- and no re-anchor ever plants a segment whose
+ // range a predecessor already covers.
+ let anchored = previous.end_offset < next.start_offset
+ && anchors
+ .get(&next.start_offset)
+ .is_some_and(|anchor| anchor.covers(previous.start_offset, previous.end_offset));
+ if !anchored {
return Err(identity.refusal(PartitionRecoveryRefusal::Hole {
previous_start: previous.start_offset,
previous_end: previous.end_offset,
@@ -594,6 +646,14 @@ fn ensure_contiguous_chain(
recoverable_bytes,
}));
}
+ info!(
+ partition_path = identity.partition_path,
+ previous_start = previous.start_offset,
+ previous_end = previous.end_offset,
+ next_start = next.start_offset,
+ "admitted a gap in the recovered segment chain: the planted segment \
+ carries the boot re-anchor's own record of it"
+ );
}
Ok(())
}
@@ -655,7 +715,11 @@ fn sweep_scratch_files_and_collect_offsets(partition_path: &str) -> Result<Vec<u
}
}
}
- Some(INDEX_EXTENSION) => orphan_candidates.push(path),
+ // An anchor outlives nothing: it describes the gap in front of ONE
+ // segment, so once that segment is gone (retention, a failed plant
+ // that never landed) the record can only mislead a later guard into
+ // admitting a gap it never saw.
+ Some(INDEX_EXTENSION | ANCHOR_EXTENSION) => orphan_candidates.push(path),
_ => {}
}
}
@@ -2584,6 +2648,43 @@ mod tests {
(messages_path, index_path)
}
+ /// Path of the anchor beside the segment planted at `start_offset`.
+ fn anchor_fixture_path(config: &ServerConfig, start_offset: u64) -> String {
+ partitions::segment_anchor::anchor_path(
+ &config
+ .system
+ .get_partition_path(STREAM_ID, TOPIC_ID, PARTITION_ID),
+ start_offset,
+ )
+ }
+
+ /// Write the anchor a plant at `planted_start` leaves behind, naming the tail
+ /// it sealed.
+ fn write_anchor_fixture(
+ config: &ServerConfig,
+ planted_start: u64,
+ sealed_start: u64,
+ sealed_end: u64,
+ ) {
+ let anchor = SegmentAnchor {
+ sealed_start,
+ sealed_end,
+ };
+ fs::write(
+ anchor_fixture_path(config, planted_start),
+ anchor.to_bytes(),
+ )
+ .expect("write anchor fixture");
+ }
+
+ /// Recover expecting a refusal, with `context` naming what should have failed.
+ async fn refusal(config: &ServerConfig, context: &str) -> ServerError {
+ match recover(config).await {
+ Ok(recovered) => panic!("{context}, got {} segments", recovered.len()),
+ Err(error) => error,
+ }
+ }
+
fn len_of(path: &str) -> u64 {
fs::metadata(path).expect("stat fixture file").len()
}
@@ -2603,27 +2704,19 @@ mod tests {
}
async fn recover(config: &ServerConfig) -> Result<Vec<RecoveredSegment>, ServerError> {
- recover_with(config, false, 0).await
+ recover_with(config, false).await
}
async fn recover_under_fsync(
config: &ServerConfig,
enforce_fsync: bool,
) -> Result<Vec<RecoveredSegment>, ServerError> {
- recover_with(config, enforce_fsync, 0).await
- }
-
- async fn recover_reserved(
- config: &ServerConfig,
- offset_reserved: u64,
- ) -> Result<Vec<RecoveredSegment>, ServerError> {
- recover_with(config, false, offset_reserved).await
+ recover_with(config, enforce_fsync).await
}
async fn recover_with(
config: &ServerConfig,
enforce_fsync: bool,
- offset_reserved: u64,
) -> Result<Vec<RecoveredSegment>, ServerError> {
load_persisted_segments(
config,
@@ -2631,7 +2724,6 @@ mod tests {
IggyByteSize::from(SEGMENT_MAX_SIZE),
enforce_fsync,
&PartitionStats::default(),
- offset_reserved,
)
.await
}
@@ -3051,40 +3143,181 @@ mod tests {
assert_eq!(bytes_of(&next_index_path), next_index);
}
+ /// The shape the boot re-anchor leaves: a sealed tail, then the next segment
+ /// planted above it, with the anchor beside the plant naming the tail.
#[compio::test]
- async fn given_a_hole_inside_the_reservation_when_recovering_should_accept_the_chain() {
+ async fn given_an_anchored_gap_when_recovering_should_accept_the_chain() {
let tmp = tempdir().expect("tempdir");
let config = test_config(&tmp);
prepare_partition_dir(&config);
- // The shape the boot re-anchor leaves: a sealed tail, then the next
- // segment planted at a frontier the superblock already claimed.
write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
write_segment(&config, 10, &encoded_batch(10, 1), &index_entry(10, 0));
+ write_anchor_fixture(&config, 10, 0, 2);
- let recovered = recover_reserved(&config, 10)
+ let recovered = recover(&config)
.await
- .expect("a gap the reservation covers is the re-anchor's, not damage");
+ .expect("a gap the plant recorded is the re-anchor's, not damage");
assert_eq!(recovered.len(), 2);
assert_eq!(recovered[0].segment.start_offset, 0);
assert_eq!(recovered[1].segment.start_offset, 10);
}
+ /// The regression this record exists to close: with no anchor the gap is a
+ /// segment that went missing, and admitting it serves a holed log silently.
#[compio::test]
- async fn given_a_hole_above_the_reservation_when_recovering_should_still_refuse() {
+ async fn given_an_unanchored_gap_when_recovering_should_refuse() {
let tmp = tempdir().expect("tempdir");
let config = test_config(&tmp);
prepare_partition_dir(&config);
write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
write_segment(&config, 10, &encoded_batch(10, 1), &index_entry(10, 0));
- // One below the gap's far side: a reservation that stops short of it
- // never claimed those offsets, so the segment is a stray file.
- let error = recover_reserved(&config, 9)
+ let error = refusal(&config, "a gap no plant recorded must refuse recovery").await;
+
+ assert!(
+ matches!(
+ &error,
+ ServerError::PartitionRecoveryRefused {
+ reason: PartitionRecoveryRefusal::Hole { .. },
+ ..
+ }
+ ),
+ "expected a hole refusal, got {error:?}"
+ );
+ }
+
+ /// An anchor names WHICH segment it sealed, so one left behind by an earlier
+ /// chain cannot legitimise a gap it never saw.
+ #[compio::test]
+ async fn given_an_anchor_naming_another_segment_when_recovering_should_refuse() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
+ write_segment(&config, 10, &encoded_batch(10, 1), &index_entry(10, 0));
+ // Right shape, wrong predecessor: this anchor describes a tail ending at
+ // 5, and the chain's tail ends at 2.
+ write_anchor_fixture(&config, 10, 0, 5);
+
+ let error = refusal(&config, "an anchor for a different tail must not cover it").await;
+ assert!(
+ matches!(
+ &error,
+ ServerError::PartitionRecoveryRefused {
+ reason: PartitionRecoveryRefusal::Hole { .. },
+ ..
+ }
+ ),
+ "expected a hole refusal, got {error:?}"
+ );
+ }
+
+ /// A corrupt anchor proves nothing, so the gap it would have covered stays
+ /// damage rather than becoming legitimate by default.
+ #[compio::test]
+ async fn given_a_corrupt_anchor_when_recovering_should_refuse() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
+ write_segment(&config, 10, &encoded_batch(10, 1), &index_entry(10, 0));
+ let path = anchor_fixture_path(&config, 10);
+ let mut bytes = SegmentAnchor {
+ sealed_start: 0,
+ sealed_end: 2,
+ }
+ .to_bytes();
+ bytes[8] ^= 1;
+ fs::write(&path, bytes).expect("write corrupt anchor");
+
+ let error = refusal(&config, "a corrupt anchor must not cover a gap").await;
+ assert!(
+ matches!(
+ &error,
+ ServerError::PartitionRecoveryRefused {
+ reason: PartitionRecoveryRefusal::Hole { .. },
+ ..
+ }
+ ),
+ "expected a hole refusal, got {error:?}"
+ );
+ }
+
+ /// Every crash cycle leaves one more re-anchor gap, and the guard runs on the
+ /// NEXT boot -- before the re-anchor -- so by the second cycle the earlier gap
+ /// is no longer the last pair. A rule keyed on the last pair alone would
+ /// refuse this, and the solo arm tombstones a chain with bytes in it, taking a
+ /// healthy partition dark from the second crash onward.
+ #[compio::test]
+ async fn given_gaps_from_several_crash_cycles_when_recovering_should_accept_the_chain() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
+ write_segment(&config, 10, &encoded_batch(10, 2), &index_entry(10, 0));
+ write_segment(&config, 20, &encoded_batch(20, 1), &index_entry(20, 0));
+ write_anchor_fixture(&config, 10, 0, 2);
+ write_anchor_fixture(&config, 20, 10, 11);
+
+ let recovered = recover(&config)
.await
- .err()
- .expect("a gap past the claimed ceiling must still refuse recovery");
+ .expect("anchored gaps accumulate one per crash, not one total");
+
+ assert_eq!(recovered.len(), 3);
+ assert_eq!(recovered[0].segment.start_offset, 0);
+ assert_eq!(recovered[1].segment.start_offset, 10);
+ assert_eq!(recovered[2].segment.start_offset, 20);
+ }
+ /// One lost segment in the middle of a chain whose OTHER gaps are all
+ /// anchored. The anchors say nothing about this pair, so it must still refuse
+ /// -- the shape a single monotone ceiling could not separate from a plant.
+ #[compio::test]
+ async fn given_a_lost_segment_among_anchored_gaps_when_recovering_should_refuse() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
+ write_segment(&config, 10, &encoded_batch(10, 2), &index_entry(10, 0));
+ // 20 was an ordinary rotation off 12, then went missing; 30 is a plant.
+ write_segment(&config, 30, &encoded_batch(30, 1), &index_entry(30, 0));
+ write_anchor_fixture(&config, 10, 0, 2);
+
+ let error = refusal(&config, "a lost middle segment must refuse recovery").await;
+ assert!(
+ matches!(
+ &error,
+ ServerError::PartitionRecoveryRefused {
+ reason: PartitionRecoveryRefusal::Hole {
+ previous_start: 10,
+ previous_end: 11,
+ next_start: 30,
+ ..
+ },
+ ..
+ }
+ ),
+ "expected a hole refusal naming the lost pair, got {error:?}"
+ );
+ }
+
+ /// No re-anchor plants a segment whose range a predecessor already covers, so
+ /// an anchor must not launder an overlap either. Reachable because each end
+ /// offset is walked from its own file with nothing clamping it against the
+ /// next start.
+ #[compio::test]
+ async fn given_overlapping_segments_when_recovering_should_refuse_even_when_anchored() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ // `0.log` walks to 0..=15 while `10.log` claims 10 onward: the pair runs
+ // backwards, which no legitimate chain does.
+ write_segment(&config, 0, &encoded_batch(0, 16), &index_entry(0, 0));
+ write_segment(&config, 10, &encoded_batch(10, 3), &index_entry(10, 0));
+ write_anchor_fixture(&config, 10, 0, 15);
+
+ let error = refusal(&config, "an overlap must refuse however it is recorded").await;
assert!(
matches!(
&error,
@@ -3097,6 +3330,27 @@ mod tests {
);
}
+ /// An anchor whose segment is gone can only mislead a later guard, so the boot
+ /// sweep collects it the way it collects an orphaned index.
+ #[compio::test]
+ async fn given_an_anchor_with_no_segment_when_recovering_should_sweep_it() {
+ let tmp = tempdir().expect("tempdir");
+ let config = test_config(&tmp);
+ prepare_partition_dir(&config);
+ write_segment(&config, 0, &encoded_batch(0, 3), &index_entry(0, 0));
+ // The window a crash between the anchor write and the plant leaves.
+ write_anchor_fixture(&config, 10, 0, 2);
+ let orphan = anchor_fixture_path(&config, 10);
+
+ let recovered = recover(&config).await.expect("recover the intact chain");
+
+ assert_eq!(recovered.len(), 1);
+ assert!(
+ !Path::new(&orphan).exists(),
+ "an anchor naming a segment that does not exist must be swept"
+ );
+ }
+
#[compio::test]
async fn given_non_monotone_index_entries_when_recovering_should_rebuild_the_index_from_the_log()
{
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 438b87f23..c4f841b4b 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -887,6 +887,16 @@ impl<M> RestorableMetadataStm for M where
/// so the bounded per-peer bus queue can never drop a burst tail. Clamped
/// against the live bus ceiling by
/// [`IggyShard::state_chunk_len_max`] rather than assumed to fit.
+/// Superblock writes issued at once when a whole shard's groups need one in the
+/// same pass: a node-wide view change, or a graceful stop collapsing every
+/// partition's offset reservation.
+///
+/// Each write is a create + write + 2 fsyncs. Serial, a few hundred groups on
+/// ordinary storage overrun the view-change escalation window (and, on the stop
+/// path, a supervisor's kill timeout); unbounded, they dump the whole burst of
+/// fds and fsyncs onto the reactor in one pass.
+const SUPERBLOCK_FAN_OUT: usize = 16;
+
const STATE_CHUNK_LEN: u32 = 256 * 1024;
/// Bus frame ceiling assumed before bootstrap overrides it. Matches the
@@ -6458,40 +6468,49 @@ where
// partitions-plane borrow is held across the tick `.await`.
namespace_scratch.extend(partitions.namespaces().copied());
- // Pre-pass: issue every group's pending superblock persist
- // CONCURRENTLY. A cluster-wide view change makes every group on
- // this shard need one in the same tick, and each `atomic_replace`
- // is a create + write + 2 fsyncs; run serially, a few hundred
- // groups on ordinary storage exceed the 5s view-change escalation
- // and loop elections. The persists are independent (each group owns
- // its store, lock, and failure bookkeeping, all behind `&self`),
- // and the per-group loop below re-checks the gate on its lock-free
- // fast path, so gating semantics are unchanged.
+ // Pre-pass: issue every group's pending superblock write CONCURRENTLY.
+ // A cluster-wide view change makes every group on this shard need one in
+ // the same tick, and each `atomic_replace` is a create + write + 2
+ // fsyncs; run serially, a few hundred groups on ordinary storage exceed
+ // the 5s view-change escalation and loop elections. The writes are
+ // independent (each group owns its store, lock, and failure bookkeeping,
+ // all behind `&self`), and the per-group loop below re-checks the persist
+ // gate on its lock-free fast path, so gating semantics are unchanged.
+ //
+ // The offset-reservation extension rides the same pre-pass, which is the
+ // whole point of it being here: the append fence writes the superblock
+ // INLINE in this pump, where those two fsyncs delay the tick above for
+ // every group on the core. Extending at half a block of headroom keeps
+ // the fence on its lock-free fast path under load, so the write happens
+ // here instead of in front of a produce. Ordered BEFORE the persist
+ // because any write marks the view durable, so one write can satisfy
+ // both and the persist gate below then finds nothing to do.
let pending_persists: Vec<_> = namespace_scratch
.iter()
.copied()
.filter(|namespace| {
- partitions
- .get_by_ns(namespace)
- .is_some_and(|partition| partition.consensus().needs_superblock_persist())
+ partitions.get_by_ns(namespace).is_some_and(|partition| {
+ partition.consensus().needs_superblock_persist()
+ || partition.needs_offset_reservation_extension()
+ })
})
.map(|namespace| async move {
if let Some(partition) = partitions.get_by_ns(&namespace) {
- // The only dropped durability verdict in the tree: this pre-pass
- // exists to coalesce the writes, and the per-group loop below re-runs
- // the same gate on its lock-free fast path and withholds every
- // view-scoped send when it fails, so the verdict here is redundant
- // rather than ignored.
+ // Verdicts dropped on purpose. The reservation is backstopped
+ // by the fence at the mint, which refuses the append if the
+ // ceiling never caught up; and the persist gate is re-run by
+ // the per-group loop below on its lock-free fast path, which
+ // withholds every view-scoped send when it fails.
+ if partition.needs_offset_reservation_extension() {
+ let _ = partition.extend_offset_reservation().await;
+ }
let _ = partition.persist_superblock_if_needed().await;
}
})
.collect();
- // Capped fan-out: each persist is a create + write + 2 fsyncs, and a
- // node-wide view change over many partitions must not dump an
- // unbounded fd/fsync burst onto the reactor in one tick.
let mut pending_persists = pending_persists.into_iter();
loop {
- let chunk: Vec<_> = pending_persists.by_ref().take(16).collect();
+ let chunk: Vec<_> = pending_persists.by_ref().take(SUPERBLOCK_FAN_OUT).collect();
if chunk.is_empty() {
break;
}
@@ -6521,6 +6540,26 @@ where
}
continue;
}
+ // Same bound the metadata plane fail-stops on, applied per group. A
+ // partition whose superblock keeps refusing withholds every
+ // view-scoped send AND refuses every append, so it serves nothing
+ // while the process still reports healthy.
+ let superblock_failures = partition.superblock_write_failures();
+ if superblock_wedged(
+ superblock_failures,
+ self.superblock_wedged_fatal_failures.get(),
+ ) {
+ consensus::fatal(
+ FatalReason::SuperblockWedged,
+ &format!(
+ "partition superblock persist failed {superblock_failures} consecutive \
+ times for namespace {}, past the [cluster] \
+ superblock_wedged_fatal_timeout window; exiting so a supervisor handles \
+ the wedge instead of the replica limping fenced",
+ namespace.inner()
+ ),
+ );
+ }
let consensus = partition.consensus();
// Only while a view change is live. A `Normal` tick has no consumer:
@@ -6744,6 +6783,7 @@ where
partitions = namespaces.len(),
"shutdown flush: draining committed journals to segment storage"
);
+ let mut collapse_pending = Vec::new();
for namespace in namespaces {
let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
continue;
@@ -6762,21 +6802,45 @@ where
// after this flush). A partition already fenced by the commit
// path keeps its original fault.
partition.fence_flush_failure();
- // The collapse below claims the segments account for every
- // confirmed offset, which a failed flush is exactly the case
- // against, so leave the reservation standing.
+ // The collapse claims the segments account for every confirmed
+ // offset, which a failed flush is exactly the case against, so
+ // leave the reservation standing.
continue;
}
- // The segments now prove where the offset space ends, so the
- // reservation has nothing left to witness. Without the collapse
- // every clean stop would leave a lease-block-wide hole.
- if !partition.collapse_offset_reservation().await {
- tracing::warn!(
- namespace_raw = namespace.inner(),
- "could not collapse the offset reservation on shutdown; the restart \
- will resume above it and leave a gap in the offset space"
- );
+ collapse_pending.push(namespace);
+ }
+
+ // Collapsed CONCURRENTLY, for the same reason the tick coalesces its view
+ // persists: see [`SUPERBLOCK_FAN_OUT`]. Each group owns its store, lock
+ // and failure bookkeeping, all behind `&self`.
+ //
+ // The flushes above stay serial: they take `&mut`, and the writers they
+ // drive are the shard's, not the partition's.
+ let mut pending = collapse_pending
+ .into_iter()
+ .map(|namespace| async move {
+ // The segments now prove where the offset space ends, so the
+ // reservation has nothing left to witness. Without the collapse
+ // every clean stop would leave a lease-block-wide hole.
+ let Some(partition) = partitions.get_by_ns(&namespace) else {
+ return;
+ };
+ if !partition.collapse_offset_reservation().await {
+ tracing::warn!(
+ namespace_raw = namespace.inner(),
+ "could not collapse the offset reservation on shutdown; the restart \
+ will resume above it and leave a gap in the offset space"
+ );
+ }
+ })
+ .collect::<Vec<_>>()
+ .into_iter();
+ loop {
+ let chunk: Vec<_> = pending.by_ref().take(SUPERBLOCK_FAN_OUT).collect();
+ if chunk.is_empty() {
+ break;
}
+ futures::future::join_all(chunk).await;
}
}