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/08/27 09:51:47 UTC
(iggy) 01/01: fix(partitions): reserve offsets before confirming them
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
commit e82b27693bbe67a30c3396f8e2ee5537370911ad
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Thu Aug 27 15:19:16 2026 +0530
fix(partitions): reserve offsets before confirming them
---
core/common/src/types/options/mod.rs | 9 +
core/configs/src/server_config/defaults.rs | 1 +
core/configs/src/server_config/displays.rs | 7 +-
core/configs/src/server_config/partition.rs | 21 +
core/consensus/src/impls.rs | 4 +-
core/consensus/src/vsr_state.rs | 53 +-
.../tests/cluster/crash_offset_reuse.rs | 164 ++++-
core/metadata/src/impls/recovery.rs | 1 +
core/partitions/src/iggy_partition.rs | 679 ++++++++++++++++++++-
core/partitions/src/log.rs | 22 +
core/partitions/src/state_transfer.rs | 10 +-
core/server/config.toml | 8 +
core/server/src/bootstrap.rs | 31 +
core/server/src/partition_helpers.rs | 2 +
core/shard/src/lib.rs | 14 +
15 files changed, 983 insertions(+), 43 deletions(-)
diff --git a/core/common/src/types/options/mod.rs b/core/common/src/types/options/mod.rs
index 44a2eaf9c..edb5be3a8 100644
--- a/core/common/src/types/options/mod.rs
+++ b/core/common/src/types/options/mod.rs
@@ -259,6 +259,15 @@ pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
/// 1 MiB (was `[system.partition] size_of_messages_required_to_save`).
pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
+/// Offsets a partition claims in its superblock ahead of the mint counter
+/// before it will append, so a crash-restarted replica resumes above every
+/// offset it confirmed instead of re-minting it.
+///
+/// One superblock write (two fsyncs) per block: at 100k messages/s a 1Ki block
+/// costs ~200 fsyncs/s, 64Ki costs ~3/s. The waste is at most one block of a
+/// `u64` space per crash, visible only as a segment boundary at boot.
+pub const DEFAULT_OFFSET_RESERVATION_LEASE: u32 = 64 * 1024;
+
/// Every runtime knob at its default, for a partition built with no resolved
/// topic options (simulator, unit tests).
impl Default for TopicRuntimeDefaults {
diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs
index f752d3265..863d60957 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -174,6 +174,7 @@ impl Default for PartitionConfig {
let partition = &SERVER_CONFIG.partition;
PartitionConfig {
prepare_queue_depth: partition.prepare_queue_depth as usize,
+ offset_reservation_lease: partition.offset_reservation_lease as u32,
evicted_ring_capacity: partition.evicted_ring_capacity as usize,
evicted_ring_bytes_max: partition.evicted_ring_bytes_max.parse().unwrap(),
transfer_served_cache_bytes_max: partition
diff --git a/core/configs/src/server_config/displays.rs b/core/configs/src/server_config/displays.rs
index aa4f79f98..5c46c7b82 100644
--- a/core/configs/src/server_config/displays.rs
+++ b/core/configs/src/server_config/displays.rs
@@ -56,10 +56,11 @@ impl Display for PartitionConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ prepare_queue_depth: {}, evicted_ring_capacity: {}, \
- evicted_ring_bytes_max: {}, transfer_served_cache_bytes_max: {}, \
- transfer_artifact_bytes_max: {} }}",
+ "{{ prepare_queue_depth: {}, offset_reservation_lease: {}, \
+ evicted_ring_capacity: {}, evicted_ring_bytes_max: {}, \
+ transfer_served_cache_bytes_max: {}, transfer_artifact_bytes_max: {} }}",
self.prepare_queue_depth,
+ self.offset_reservation_lease,
self.evicted_ring_capacity,
self.evicted_ring_bytes_max,
self.transfer_served_cache_bytes_max,
diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs
index d6e327361..38a1d6e5b 100644
--- a/core/configs/src/server_config/partition.rs
+++ b/core/configs/src/server_config/partition.rs
@@ -97,6 +97,10 @@ pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 =
/// count.
pub const MAX_TRANSFER_BYTES: u64 = 64 * 1024 * 1024 * 1024;
+/// Upper bound on `offset_reservation_lease`: a typo guard, so a slipped digit
+/// cannot reach the `saturating_add` in the append fence.
+pub const MAX_OFFSET_RESERVATION_LEASE: u32 = 16 * 1024 * 1024;
+
/// Mirrors `partitions::EVICTED_RING_CAPACITY`.
pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 4096;
@@ -123,6 +127,13 @@ pub struct PartitionConfig {
/// pinned request-buffer memory by the partition count.
pub prepare_queue_depth: usize,
+ /// Offsets claimed in the superblock ahead of the mint counter before an
+ /// append, so a crash-restarted replica resumes above what it confirmed.
+ /// 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`].
+ pub offset_reservation_lease: u32,
+
/// Entries the evicted ring retains per multi-replica partition for
/// journal repair after a peer rejoins. Larger widens the window a
/// restarting peer can be served from the ring before falling back to
@@ -177,6 +188,16 @@ impl Validatable<ConfigurationError> for PartitionConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.offset_reservation_lease == 0
+ || self.offset_reservation_lease > MAX_OFFSET_RESERVATION_LEASE
+ {
+ eprintln!(
+ "{COMPONENT} partition.offset_reservation_lease ({}) must be > 0 and <= \
+ {MAX_OFFSET_RESERVATION_LEASE}",
+ self.offset_reservation_lease
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
if self.evicted_ring_capacity == 0 {
eprintln!("{COMPONENT} partition.evicted_ring_capacity must be > 0");
return Err(ConfigurationError::InvalidConfigurationValue);
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 2e4741dab..2c7c72f43 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -2014,8 +2014,10 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
checkpoint_op,
checkpoint_checksum,
// Consensus mints no message offsets: the PARTITION plane stamps
- // this in before it writes (`IggyPartition::write_superblock`).
+ // both of these in before it writes
+ // (`IggyPartition::write_superblock`).
offset_frontier: 0,
+ offset_reserved: 0,
}
}
diff --git a/core/consensus/src/vsr_state.rs b/core/consensus/src/vsr_state.rs
index 9040906ad..f01978fcb 100644
--- a/core/consensus/src/vsr_state.rs
+++ b/core/consensus/src/vsr_state.rs
@@ -31,8 +31,8 @@ use std::fmt;
/// Number of bytes [`VsrState::to_bytes`] produces: `cluster`(16) +
/// `replica_id`(1) + `replica_count`(1) + `view`(4) + `log_view`(4) +
/// `commit_max`(8) + `checkpoint_op`(8) + `checkpoint_checksum`(16) +
-/// `offset_frontier`(8).
-pub const ENCODED_LEN: usize = 66;
+/// `offset_frontier`(8) + `offset_reserved`(8).
+pub const ENCODED_LEN: usize = 74;
/// The layout before `offset_frontier` was appended.
///
@@ -97,6 +97,20 @@ pub struct VsrState {
///
/// Always `0` on the metadata plane, which mints no message offsets.
pub offset_frontier: u64,
+ /// PARTITION plane: a monotone CEILING on the offsets this replica may
+ /// already have minted, claimed ahead of the counter in blocks so an append
+ /// pays one superblock write per block instead of one per batch.
+ ///
+ /// Never folded into [`Self::offset_frontier`]. The frontier is a claim
+ /// about DATA, which state transfer's rewind guard refuses to destroy; a
+ /// reservation names no bytes, only "an offset up to here may have reached
+ /// a client". Comparing an offer against it refuses every legitimate offer
+ /// below the lease headroom, and the replica cycles transfer -> refusal ->
+ /// backoff forever. Boot seeds the mint counter from both; the rewind guard
+ /// reads the frontier alone.
+ ///
+ /// Always `0` on the metadata plane, which mints no message offsets.
+ pub offset_reserved: u64,
}
impl VsrState {
@@ -113,6 +127,7 @@ impl VsrState {
out[34..42].copy_from_slice(&self.checkpoint_op.to_le_bytes());
out[42..58].copy_from_slice(&self.checkpoint_checksum.to_le_bytes());
out[58..66].copy_from_slice(&self.offset_frontier.to_le_bytes());
+ out[66..74].copy_from_slice(&self.offset_reserved.to_le_bytes());
out
}
}
@@ -121,11 +136,15 @@ impl TryFrom<&[u8]> for VsrState {
type Error = VsrStateError;
fn try_from(bytes: &[u8]) -> Result<Self, Self::Error> {
- // Length-tolerant: a pre-`offset_frontier` record is padded out and the
- // new field reads as 0, which is exactly "no recorded frontier" (the
- // read sites filter it). One length check up front then puts every
- // field slice below in bounds by construction, so the `try_into`s
- // cannot fail.
+ // 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.
+ //
+ // `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.
let mut padded = [0u8; ENCODED_LEN];
match bytes.len() {
ENCODED_LEN => padded.copy_from_slice(bytes),
@@ -150,6 +169,7 @@ impl TryFrom<&[u8]> for VsrState {
checkpoint_op: u64::from_le_bytes(field(bytes, 34)),
checkpoint_checksum: u128::from_le_bytes(field(bytes, 42)),
offset_frontier: u64::from_le_bytes(field(bytes, 58)),
+ offset_reserved: u64::from_le_bytes(field(bytes, 66)),
};
// A record violating `log_view <= view` decodes into a replica that looks
// healthy locally while `DoViewChangeHeader::validate` makes every peer drop
@@ -224,7 +244,8 @@ mod tests {
commit_max: 6,
checkpoint_op: 7,
checkpoint_checksum: 8,
- offset_frontier: 0,
+ offset_frontier: 9,
+ offset_reserved: 10,
};
let bytes = state.to_bytes();
assert_eq!(bytes.len(), ENCODED_LEN);
@@ -236,6 +257,8 @@ mod tests {
assert_eq!(bytes[26], 6, "commit_max low byte");
assert_eq!(bytes[34], 7, "checkpoint_op low byte");
assert_eq!(bytes[42], 8, "checkpoint_checksum low byte");
+ assert_eq!(bytes[58], 9, "offset_frontier low byte");
+ assert_eq!(bytes[66], 10, "offset_reserved low byte");
assert_eq!(VsrState::try_from(&bytes[..]).unwrap(), state);
assert!(VsrState::try_from(&bytes[..ENCODED_LEN - 1]).is_err());
@@ -257,12 +280,14 @@ mod tests {
checkpoint_op: 7,
checkpoint_checksum: 5,
offset_frontier: 77,
+ offset_reserved: 88,
}
.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");
assert_eq!(decoded.view, 9);
assert_eq!(decoded.log_view, 8);
assert_eq!(decoded.commit_max, 41);
@@ -274,6 +299,16 @@ mod tests {
VsrState::try_from(&full[..40]),
Err(VsrStateError::WrongLength { .. })
));
+
+ // 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"
+ );
}
#[test]
@@ -294,9 +329,11 @@ mod tests {
// Distinct and nonzero: with 0 here a transposed write over the
// trailing field would still satisfy every assertion below.
offset_frontier: 9,
+ offset_reserved: 11,
}
.to_bytes();
assert_eq!(bytes[58], 9, "offset_frontier must occupy bytes 58..66");
+ assert_eq!(bytes[66], 11, "offset_reserved must occupy bytes 66..74");
bytes[22] = 5; // log_view = 5, view stays 4
assert_eq!(
diff --git a/core/integration/tests/cluster/crash_offset_reuse.rs b/core/integration/tests/cluster/crash_offset_reuse.rs
index 447a6cc89..1ff7e03b2 100644
--- a/core/integration/tests/cluster/crash_offset_reuse.rs
+++ b/core/integration/tests/cluster/crash_offset_reuse.rs
@@ -15,17 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-//! RED SPEC, expected to FAIL: offset identity across a crash.
+//! Offset identity across a crash.
//!
//! `SendMessagesResponse::confirmations` hands clients concrete base offsets,
-//! which makes offset reuse client-visible: a client that recorded offset N
-//! for its message must never see the server confirm a DIFFERENT message at
-//! N later. A solo node acks below the flush thresholds from RAM only and
-//! persists no offset watermark, so after a SIGKILL it restarts the partition
-//! log at the last flushed position and re-mints offsets it already
-//! confirmed. Passes only once a durable watermark (or durable journal tail)
-//! keeps post-restart offsets above everything ever acked.
+//! which makes offset reuse client-visible: a client that recorded offset N for
+//! its message must never see the server confirm a DIFFERENT message at N
+//! later. A solo node acks below the flush thresholds from RAM only, so nothing
+//! in the segments says those offsets were ever handed out. What keeps them
+//! from being re-minted is the offset RESERVATION in the partition superblock,
+//! claimed by the append fence before any of them exist and read back by boot.
+use std::path::Path;
use std::time::Duration;
use iggy::prelude::*;
@@ -97,8 +97,64 @@ async fn wait_until_serving(harness: &TestHarness, budget: Duration) -> IggyClie
}
}
-// TODO(hubcio): fix this test
-#[ignore = "confirmed offsets re-minted after a crash; no durable offset watermark"]
+/// Create the stream and its single-partition topic.
+async fn create_topic(client: &IggyClient, messages_required_to_save: Option<u32>) {
+ 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),
+ messages_required_to_save,
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("create topic");
+}
+
+/// 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.
+fn segment_base_offsets(root: &Path) -> Vec<u64> {
+ let mut offsets = Vec::new();
+ let mut stack = vec![root.to_path_buf()];
+ while let Some(dir) = stack.pop() {
+ let Ok(entries) = std::fs::read_dir(&dir) else {
+ continue;
+ };
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.is_dir() {
+ stack.push(path);
+ } else if path.extension().is_some_and(|extension| extension == "log")
+ && let Some(stem) = path.file_stem().and_then(|stem| stem.to_str())
+ && let Ok(offset) = stem.parse::<u64>()
+ {
+ offsets.push(offset);
+ }
+ }
+ }
+ offsets.sort_unstable();
+ offsets
+}
+
+/// Kill the node, bring it back, and return a client onto the restarted one.
+async fn crash_and_recover(harness: &mut TestHarness) -> IggyClient {
+ harness.kill_node(0).expect("SIGKILL the only node");
+ harness.restart_node(0).expect("restart it");
+ wait_until_serving(harness, SERVE_TIMEOUT).await
+}
+
#[iggy_harness(cluster_nodes = 1)]
async fn given_confirmed_sends_below_flush_threshold_when_a_solo_node_is_killed_should_not_remint_offsets(
harness: &mut TestHarness,
@@ -141,3 +197,91 @@ async fn given_confirmed_sends_below_flush_threshold_when_a_solo_node_is_killed_
silently different data"
);
}
+
+/// 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.
+///
+/// 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.
+#[iggy_harness(cluster_nodes = 1)]
+async fn given_a_crash_restarted_node_when_it_flushes_and_crashes_again_should_still_not_remint_offsets(
+ harness: &mut TestHarness,
+) {
+ const FLUSH_THRESHOLD: u32 = 4;
+ /// Past the threshold, so every life leaves a chain for the next boot.
+ const SENDS_PER_LIFE: u32 = 6;
+
+ let client = harness.tcp_root_client().await.unwrap();
+ create_topic(&client, Some(FLUSH_THRESHOLD)).await;
+
+ let acked = produce_acked(&client, "first-life", SENDS_PER_LIFE).await;
+ let first_life_max = *acked.last().expect("confirmed sends");
+ drop(client);
+
+ // The boot that consumes a reservation and re-anchors, then flushes the
+ // hole's far side to disk.
+ let client = crash_and_recover(harness).await;
+ let second_life = produce_acked(&client, "second-life", SENDS_PER_LIFE).await;
+ let second_life_min = second_life[0];
+ let second_life_max = *second_life.last().expect("confirmed sends");
+ assert!(
+ second_life_min > first_life_max,
+ "the first restart re-minted: confirmed {first_life_max} before the crash, \
+ then {second_life_min} after it"
+ );
+ drop(client);
+
+ // ON a segment boundary, not inside one: a segment still named 0 while
+ // holding the second life's offsets claims a range it does not have.
+ let bases = segment_base_offsets(harness.test_dir());
+ assert!(
+ bases.iter().any(|&base| base > first_life_max),
+ "no segment is anchored above the pre-crash offsets, so the second life \
+ appended into a segment named for the first: segment bases {bases:?}, last \
+ offset confirmed before the crash {first_life_max}"
+ );
+
+ // The first boot that has to read a chain the re-anchor wrote.
+ let client = crash_and_recover(harness).await;
+ let third_life = produce_acked(&client, "third-life", 1).await[0];
+ assert!(
+ 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"
+ );
+}
+
+/// The partially-flushed shape a real workload crashes in: the run below the
+/// threshold is confirmed out of the journal while everything before it is on
+/// disk. The recovered chain then ends BELOW the reservation with bytes in it,
+/// so the re-anchor has to seal it rather than append into the gap.
+#[iggy_harness(cluster_nodes = 1)]
+async fn given_sends_straddling_the_flush_threshold_when_the_node_is_killed_should_not_remint_offsets(
+ harness: &mut TestHarness,
+) {
+ const FLUSH_THRESHOLD: u32 = 4;
+ const STRADDLING_SENDS: u32 = 6;
+
+ let client = harness.tcp_root_client().await.unwrap();
+ create_topic(&client, Some(FLUSH_THRESHOLD)).await;
+
+ let acked = produce_acked(&client, "straddle", STRADDLING_SENDS).await;
+ let highest_confirmed = *acked.last().expect("confirmed sends");
+ assert_eq!(
+ acked,
+ (0..u64::from(STRADDLING_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 = produce_acked(&client, "post-straddle", 1).await[0];
+ assert!(
+ post_crash > highest_confirmed,
+ "offsets confirmed out of the journal above the last flushed one were \
+ re-minted: {highest_confirmed} went to a client before the SIGKILL, and the \
+ first send after it was confirmed at {post_crash}"
+ );
+}
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 39d5b9847..60ffcf58f 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -843,6 +843,7 @@ mod tests {
checkpoint_op,
checkpoint_checksum,
offset_frontier: 0,
+ offset_reserved: 0,
}
}
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index fb1ba35c2..19cfd95c5 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -226,6 +226,22 @@ where
/// segments that were the only other witness, after which the rebuild
/// re-mints offsets the group already handed out.
durable_offset_frontier: Cell<u64>,
+ /// The `offset_reserved` ceiling the last successful superblock write
+ /// recorded, seeded at boot from the record that write left behind.
+ ///
+ /// A `Cell` rather than a re-read of the record because every append reads
+ /// it, and the steady state ("the block still covers this batch") has to
+ /// cost nothing. Kept apart from [`Self::durable_offset_frontier`]: see
+ /// `consensus::VsrState::offset_reserved`.
+ durable_offset_reserved: Cell<u64>,
+ /// 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`].
@@ -486,6 +502,9 @@ where
superblock_retry_after_micros: Cell::new(0),
purge_deferred: false,
durable_offset_frontier: Cell::new(0),
+ durable_offset_reserved: Cell::new(0),
+ offset_reservation_lease: iggy_common::DEFAULT_OFFSET_RESERVATION_LEASE as u64,
+ mint_floor_pending: Cell::new(false),
transfer: None,
transfer_attempts: 0,
transfer_failures: 0,
@@ -593,6 +612,11 @@ 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);
}
/// Persist this group's VSR state to its superblock when the view changed
@@ -671,22 +695,77 @@ where
/// THIS partition's group is fenced; the rest of the node keeps serving.
#[allow(clippy::future_not_send)]
async fn write_superblock(&self, superblock: &SB, offset_frontier: u64) -> bool {
- // ADVANCE direction: never below what this replica has already minted,
- // and never below what the record ALREADY holds. Both bounds are
- // needed and neither implies the other -- a failed install leaves the
- // counter behind the record it wrote before the swap, so maxing against
- // the counter alone lets the fence that follows lower the durable
- // frontier. The reset direction goes through `write_superblock_inner`.
- let advanced = offset_frontier
- .max(self.offset_frontier())
- .max(self.durable_offset_frontier.get());
- self.write_superblock_inner(superblock, advanced).await
+ self.write_superblock_advancing(superblock, offset_frontier, 0)
+ .await
+ }
+
+ /// [`Self::write_superblock`] for a caller that also has a reservation to
+ /// claim. Both fields advance, neither can regress.
+ #[allow(clippy::future_not_send)]
+ async fn write_superblock_advancing(
+ &self,
+ superblock: &SB,
+ offset_frontier: u64,
+ offset_reserved: u64,
+ ) -> bool {
+ // ADVANCE direction; the reset direction goes through
+ // `write_superblock_inner`. Both bounds inside `advanced_frontier` are
+ // needed: a failed install leaves the chain behind the record it wrote
+ // before the swap, so maxing against the data alone would let the fence
+ // that follows lower the durable frontier.
+ let advanced = self.advanced_frontier(offset_frontier);
+ // Nothing but the record witnesses a reservation, so a caller with no
+ // claim of its own (every view-change write) passes 0 and carries the
+ // recorded one forward; dropping it would let the next boot seed the
+ // counter below what an earlier append already fenced.
+ let reserved = offset_reserved.max(self.durable_offset_reserved.get());
+ self.write_superblock_inner(superblock, advanced, reserved)
+ .await
+ }
+
+ /// 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.
+ fn advanced_frontier(&self, claim: u64) -> u64 {
+ claim
+ .max(self.held_offset_frontier())
+ .max(self.durable_offset_frontier.get())
+ }
+
+ /// Record an incoming state-transfer frontier, advancing the frontier but
+ /// SETTING the reservation to it.
+ ///
+ /// The one place the otherwise-monotone reservation may come down, and the
+ /// one place it must: the offer describes the group's committed log, so a
+ /// local reservation above it covers offsets this replica never confirmed.
+ /// Carried forward, it re-seeds the counter a lease block above the group
+ /// after the next restart, where every replicated prepare fails the
+ /// `base_offset == dirty_offset + 1` check.
+ #[allow(clippy::future_not_send)]
+ #[must_use = "the bool is the durability verdict; dropping it silently ignores a failed write"]
+ pub async fn install_offset_frontier_at(&self, frontier: u64) -> bool {
+ 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 advanced = self.advanced_frontier(frontier);
+ self.write_superblock_inner(superblock.as_ref(), advanced, frontier)
+ .await
}
/// The write itself; the advance and reset directions differ only in the
- /// frontier they hand in.
+ /// values they hand in.
#[allow(clippy::future_not_send)]
- async fn write_superblock_inner(&self, superblock: &SB, offset_frontier: u64) -> bool {
+ async fn write_superblock_inner(
+ &self,
+ superblock: &SB,
+ offset_frontier: u64,
+ offset_reserved: u64,
+ ) -> bool {
// The pairing fields stay `(0, 0)` and `commit_max` is a dead write
// on this plane: nothing reads either back (`restore_partition_view`
// restores view/log_view only), because recovery re-derives the
@@ -701,11 +780,17 @@ where
// lower bound boot can re-seed from.
let mut state = self.consensus.vsr_state(0, 0);
state.offset_frontier = offset_frontier;
+ // A frontier of N says offsets below N exist, so a reservation under it
+ // is not a reservation. Clamped here rather than per caller so the
+ // reset direction gets it too, where a stale higher reservation would
+ // seed the next boot into the offset space the reset just erased.
+ state.offset_reserved = offset_reserved.max(offset_frontier);
match superblock.write(&state.to_bytes()).await {
Ok(()) => {
self.consensus
.mark_superblock_durable(state.view, state.log_view);
self.durable_offset_frontier.set(state.offset_frontier);
+ self.durable_offset_reserved.set(state.offset_reserved);
self.superblock_write_failures.set(0);
self.superblock_retry_after_micros.set(0);
true
@@ -746,14 +831,21 @@ where
/// proved.
///
/// The record is a lower bound, never a completeness claim: it exists
- /// because three paths leave a replica whose counter would otherwise
- /// restart at 0 while the group is at N (a transfer install of an all-GC'd
- /// origin, a crash inside the install's swap window, and the
- /// fence-and-rebuild path, which needs no crash at all). Restarting the
+ /// because four paths leave a replica whose counter would otherwise
+ /// restart below where the group already is (a transfer install of an
+ /// all-GC'd origin, a crash inside the install's swap window, the
+ /// fence-and-rebuild path, which needs no crash at all, and a crash while
+ /// acked messages were still resident in the journal). Restarting the
/// counter is not a lag -- replicas re-stamp `base_offset` from it and
/// 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`].
+ ///
/// 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
@@ -808,6 +900,34 @@ 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.
+ ///
+ /// 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
+ /// threshold-gated flush routinely leaves committed messages unnamed by any
+ /// segment.
+ #[must_use]
+ pub fn held_offset_frontier(&self) -> u64 {
+ let on_disk = self
+ .log
+ .segments()
+ .iter()
+ .filter(|segment| segment.size.as_bytes_u64() > 0)
+ .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)
+ }
+
/// Force the durable record to catch up with the current offset frontier,
/// outside the view-change gate.
///
@@ -862,7 +982,9 @@ where
return false;
}
let _superblock_guard = self.superblock_lock.acquire().await;
- self.write_superblock_inner(superblock.as_ref(), frontier)
+ // Reservation reset with it: left above, it would seed the next boot
+ // back into the offset space this reset just left behind.
+ self.write_superblock_inner(superblock.as_ref(), frontier, frontier)
.await
}
@@ -889,7 +1011,7 @@ where
let _superblock_guard = self.superblock_lock.acquire().await;
match intended {
Some(frontier) => {
- self.write_superblock_inner(superblock.as_ref(), frontier)
+ self.write_superblock_inner(superblock.as_ref(), frontier, frontier)
.await
}
None => {
@@ -910,6 +1032,25 @@ where
self.consensus.clock_realtime_micros() < self.superblock_retry_after_micros.get()
}
+ /// Drop the reservation back onto the frontier, once a graceful flush has
+ /// made the segments account for every offset this replica confirmed.
+ ///
+ /// The reservation is there for the crash case, where they do not. Left
+ /// standing it would make every ordinary restart resume a lease block
+ /// higher and hole the offset space for nothing.
+ ///
+ /// 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 {
+ return true;
+ }
+ self.reset_offset_frontier_at(frontier).await
+ }
+
/// [`Self::persist_offset_frontier`] for a frontier this replica has not
/// reached yet.
///
@@ -933,6 +1074,80 @@ 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.
+ ///
+ /// 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
+ }
+
+ /// [`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.
+ fn armed_mint_floor(&self) -> u64 {
+ if self.consensus.replica_count() > 1 || !self.mint_floor_pending.get() {
+ return 0;
+ }
+ self.durable_offset_reserved.get()
+ }
+
+ /// The append fence: make sure the durable record already permits every
+ /// offset up to and including `end_offset` before the caller lets them
+ /// exist.
+ ///
+ /// `SendMessagesResponse` hands clients concrete base offsets and the poll
+ /// path serves committed messages out of the resident journal, so an offset
+ /// 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
+ /// 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.
+ #[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 {
+ // 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 {
+ return true;
+ }
+ 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;
+ // A batch queued behind another append's write finds the block already
+ // extended.
+ if self.durable_offset_reserved.get() > end_offset {
+ return true;
+ }
+ let claim = end_offset
+ .saturating_add(1)
+ .saturating_add(self.offset_reservation_lease);
+ self.write_superblock_advancing(superblock.as_ref(), 0, claim)
+ .await
+ }
+
/// Burn one transfer stall round; `true` once the budget is exhausted.
/// Lives on the partition, not the session, so a re-minted session
/// cannot reset it (see [`Self::transfer_attempts`]).
@@ -1038,6 +1253,16 @@ where
.unwrap_or(config.messages_required_to_save)
}
+ /// Install the offset-reservation block size resolved from this node's
+ /// `PartitionsConfig`.
+ ///
+ /// 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 };
+ }
+
/// Whether this partition's segments reserve their bytes on open.
#[must_use]
pub fn effective_preallocate_segments(&self, config: &PartitionsConfig) -> bool {
@@ -1745,7 +1970,7 @@ where
return Err(IggyError::CannotAppendMessage);
}
- let dirty_offset = if self.should_increment_offset {
+ let next = if self.should_increment_offset {
self.dirty_offset
.load(Ordering::Relaxed)
.checked_add(1)
@@ -1753,6 +1978,10 @@ where
} else {
0
};
+ // 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());
// Reuse the prepare's monotonic timestamp, assigned once by the primary
// in `project()` (`next_monotonic_timestamp`) and replicated verbatim to
@@ -2666,6 +2895,19 @@ where
.checked_add(u64::from(batch_messages_count) - 1)
.ok_or(IggyError::CannotAppendMessage)?;
+ // Past this line the offsets are in the journal, hence committable,
+ // pollable, confirmable and forwardable, and no later gate can take
+ // them back. See [`Self::reserve_offsets_through`].
+ //
+ // LOCK ORDER: the caller holds `write_lock` and this takes
+ // `superblock_lock` under it. The install path takes them in the
+ // reverse order, safely only because `reset_offset_frontier_at` drops
+ // `superblock_lock` before `try_install` takes `write_lock`. Never hold
+ // `superblock_lock` across a `write_lock` acquire.
+ if !self.reserve_offsets_through(last_dirty_offset).await {
+ return Err(IggyError::CannotAppendMessage);
+ }
+
let segment_index = self.log.segments().len() - 1;
let current_position = self.log.segments()[segment_index].current_position;
let next_position = current_position
@@ -4035,6 +4277,121 @@ where
Ok(())
}
+ /// Re-anchor the append point after boot re-seeded the offset counter above
+ /// 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.
+ ///
+ /// 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`.
+ ///
+ /// # Errors
+ /// [`IggyError`] when the fresh segment cannot be created, leaving the
+ /// partition without a serviceable chain.
+ #[allow(clippy::future_not_send)]
+ pub async fn reanchor_to_offset_frontier(
+ &mut self,
+ config: &PartitionsConfig,
+ ) -> Result<(), IggyError> {
+ // 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());
+ if frontier == 0 {
+ return Ok(());
+ }
+ let namespace = self.namespace();
+ let mut retired = 0usize;
+ while let Some(segment) = self.log.segments().last() {
+ if segment.size.as_bytes_u64() > 0 || segment.start_offset >= frontier {
+ break;
+ }
+ let Some((segment, mut storage)) = self.log.retire_back() else {
+ break;
+ };
+ let (messages_path, index_path) = storage.segment_and_index_paths();
+ let _ = storage.shutdown();
+ drop(storage);
+ for path in messages_path.into_iter().chain(index_path) {
+ match compio::fs::remove_file(&path).await {
+ Ok(()) => {}
+ Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+ Err(error) => {
+ warn!(
+ 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"
+ );
+ }
+ }
+ }
+ tracing::info!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ start_offset = segment.start_offset,
+ offset_frontier = frontier,
+ "unlinked an empty segment named below the restored offset frontier"
+ );
+ 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.
+ if retired > 0
+ && let Some(partition_dir) = self.partition_dir.clone()
+ && let Err(error) = crate::state_transfer::fsync_dir(&partition_dir).await
+ {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = namespace.inner(),
+ partition_dir,
+ %error,
+ "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(());
+ }
+ 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(),
+ sealed_end,
+ offset_frontier = frontier,
+ "sealed the recovered tail and planted a fresh segment at the restored frontier"
+ );
+ Ok(())
+ }
+
/// Record the purge's frontier reset BEFORE the purge touches anything.
///
/// The unlinks are made durable by their own directory fsync, so a crash
@@ -5032,6 +5389,26 @@ mod tests {
)
}
+ /// A SOLO partition, the shape the offset reservation is scoped to.
+ fn solo_recording_partition() -> IggyPartition<IggyMessageBus, RecordingSuperblock> {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let consensus = VsrConsensus::new(
+ TEST_CLUSTER,
+ 0,
+ 1,
+ namespace.inner(),
+ IggyMessageBus::new(0),
+ LocalPipeline::new(),
+ );
+ consensus.init();
+ IggyPartition::with_in_memory_storage(
+ Arc::new(PartitionStats::default()),
+ consensus,
+ IggyByteSize::from(1024 * 1024),
+ false,
+ )
+ }
+
/// Partition whose consensus already advanced to `(view, log_view)` with
/// nothing marked durable, as after a view change and before the persist
/// gate runs.
@@ -5140,6 +5517,255 @@ mod tests {
.offset_frontier
}
+ fn last_recorded_reservation(store: &RecordingSuperblock) -> u64 {
+ let writes = store.writes.borrow();
+ let bytes = writes.last().expect("a superblock write landed");
+ consensus::VsrState::try_from(bytes.as_slice())
+ .expect("recorded payload decodes as a VsrState")
+ .offset_reserved
+ }
+
+ fn recorded_state(offset_frontier: u64, offset_reserved: u64) -> consensus::VsrState {
+ consensus::VsrState {
+ cluster: TEST_CLUSTER,
+ replica_id: 0,
+ replica_count: 1,
+ view: 1,
+ log_view: 1,
+ commit_max: 0,
+ checkpoint_op: 0,
+ checkpoint_checksum: 0,
+ offset_frontier,
+ offset_reserved,
+ }
+ }
+
+ /// One superblock write per block, not per batch: a fence writing per append
+ /// would put two fsyncs in front of every produce.
+ #[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);
+ partition.set_superblock(store.clone(), None);
+ partition.set_offset_reservation_lease(16);
+
+ assert!(partition.reserve_offsets_through(0).await);
+ assert_eq!(store.attempts.get(), 1, "the first offset claims a block");
+ assert_eq!(
+ last_recorded_reservation(&store),
+ 17,
+ "the claim runs one past the offset plus the lease"
+ );
+
+ for offset in 1..=16 {
+ assert!(partition.reserve_offsets_through(offset).await);
+ }
+ assert_eq!(
+ store.attempts.get(),
+ 1,
+ "every offset inside the block is covered by the claim already on disk"
+ );
+
+ assert!(partition.reserve_offsets_through(17).await);
+ assert_eq!(
+ store.attempts.get(),
+ 2,
+ "the first offset past the block extends it"
+ );
+ assert_eq!(last_recorded_reservation(&store), 34);
+ }
+
+ /// Fail-closed: offsets the record does not cover would be confirmed to a
+ /// client with nothing durable saying they were handed out.
+ #[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);
+ partition.set_superblock(store.clone(), None);
+ partition.set_offset_reservation_lease(4);
+ store.fail_writes.set(true);
+
+ assert!(
+ !partition.reserve_offsets_through(0).await,
+ "an unrecordable claim must refuse the append"
+ );
+ }
+
+ /// 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.
+ #[test]
+ fn given_a_recorded_reservation_when_solo_should_floor_the_next_mint() {
+ let mut partition = solo_recording_partition();
+ assert_eq!(
+ partition.consensus().replica_count(),
+ 1,
+ "the floor is scoped to solo groups"
+ );
+ let store = Rc::new(RecordingSuperblock::default());
+
+ partition.set_superblock(store.clone(), None);
+ assert_eq!(
+ partition.mint_floor(),
+ 0,
+ "nothing recorded, nothing floored"
+ );
+
+ partition.set_superblock(store, Some(&recorded_state(0, 65_537)));
+ assert_eq!(
+ partition.mint_floor(),
+ 65_537,
+ "the first mint of the incarnation 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"
+ );
+ }
+
+ /// 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.
+ #[test]
+ fn given_a_recorded_reservation_when_replicated_should_not_floor_the_mint() {
+ 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)),
+ );
+ assert_eq!(
+ partition.mint_floor(),
+ 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.
+ #[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;
+ 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);
+ assert_eq!(
+ partition.held_offset_frontier(),
+ 0,
+ "no segment and no journal entry: the counter speaks for no messages"
+ );
+
+ let offer = crate::state_transfer::ConsumerOffsetsWire {
+ purge_generation: 0,
+ next_offset: 1_030,
+ consumers: Vec::new(),
+ groups: Vec::new(),
+ };
+ let outcome = partition
+ .install_state_transfer(&repair_config(), 12, Vec::new(), &offer.encode(), 0)
+ .await;
+ assert!(
+ !matches!(
+ outcome,
+ Err(crate::state_transfer::PartitionInstallError::OfferRewindsDurableData { .. })
+ ),
+ "the offer destroys nothing this replica holds, so the fence must let it \
+ through: got {outcome:?}"
+ );
+
+ let _ = std::fs::remove_dir_all(&partition_dir);
+ }
+
+ /// After a clean shutdown the segments account for every confirmed offset,
+ /// so collapsing the reservation keeps the offset space dense across an
+ /// ordinary restart instead of jumping a lease block every time.
+ #[compio::test]
+ async fn given_a_flushed_partition_when_collapsing_should_drop_the_reservation_to_the_frontier()
+ {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = solo_recording_partition();
+ partition.set_superblock(store.clone(), Some(&recorded_state(0, 65_537)));
+ partition.should_increment_offset = true;
+ partition.offset.store(24, Ordering::Release);
+
+ assert!(partition.collapse_offset_reservation().await);
+ assert_eq!(last_recorded_frontier(&store), 25);
+ assert_eq!(
+ last_recorded_reservation(&store),
+ 25,
+ "a clean stop leaves no claim above what the segments prove"
+ );
+
+ // And the restart that follows mints where it left off, not a block up.
+ let mut restarted = solo_recording_partition();
+ restarted.set_superblock(store.clone(), Some(&recorded_state(25, 25)));
+ assert_eq!(
+ restarted.mint_floor(),
+ 25,
+ "the floor is the frontier itself, so it moves nothing"
+ );
+ }
+
+ /// An install is the one place the reservation may come down: left high, it
+ /// re-seeds the counter above the group and every replicated prepare fails
+ /// the `base_offset == dirty_offset + 1` check.
+ #[compio::test]
+ async fn given_install_frontier_when_recorded_should_set_the_reservation_down_to_it() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = partition_at_view(1, 1);
+ partition.set_superblock(store.clone(), Some(&recorded_state(0, 70_000)));
+
+ assert!(partition.install_offset_frontier_at(1_030).await);
+ assert_eq!(
+ last_recorded_reservation(&store),
+ 1_030,
+ "the install's frontier replaces the stale reservation"
+ );
+ assert_eq!(last_recorded_frontier(&store), 1_030);
+ }
+
+ /// A purge resets the offset space to zero and the reservation goes with it:
+ /// a survivor would re-seed the counter into the space just erased.
+ #[compio::test]
+ async fn given_purge_reset_when_recorded_should_clear_the_reservation() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = partition_at_view(1, 1);
+ partition.set_superblock(store.clone(), Some(&recorded_state(500, 70_000)));
+
+ assert!(partition.reset_offset_frontier_at(0).await);
+ assert_eq!(last_recorded_frontier(&store), 0);
+ assert_eq!(
+ last_recorded_reservation(&store),
+ 0,
+ "a reset that left the reservation behind would resurrect the old space"
+ );
+ }
+
+ /// The reservation can never sit below the frontier: "offsets under N exist"
+ /// is stronger than "offsets under N may have been handed out".
+ #[compio::test]
+ async fn given_reservation_below_the_frontier_when_written_should_clamp_it_up() {
+ let store = Rc::new(RecordingSuperblock::default());
+ let mut partition = partition_at_view(1, 1);
+ partition.set_superblock(store.clone(), None);
+ partition.should_increment_offset = true;
+ partition.offset.store(99, Ordering::Release);
+
+ assert!(partition.persist_offset_frontier_at(100).await);
+ assert_eq!(last_recorded_frontier(&store), 100);
+ assert_eq!(
+ last_recorded_reservation(&store),
+ 100,
+ "the reservation is clamped up to the frontier it accompanies"
+ );
+ }
+
/// The fence path persists the frontier while the live counter still sits
/// at its pre-install value, so an advance that maxes against the counter
/// alone erases the record and then quarantines the segments that were its
@@ -5186,6 +5812,7 @@ mod tests {
checkpoint_op: 0,
checkpoint_checksum: 0,
offset_frontier: 4_200,
+ offset_reserved: 0,
};
partition.set_superblock(store.clone(), Some(&recovered));
assert_eq!(partition.offset_frontier(), 0, "nothing minted locally");
@@ -6788,6 +7415,13 @@ 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.
+ {
+ let info = &mut partition.log.journal_mut().info;
+ info.messages_count = 100;
+ info.current_offset = 99;
+ }
let behind = crate::state_transfer::ConsumerOffsetsWire {
purge_generation: 0,
@@ -6849,6 +7483,13 @@ 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.
+ {
+ let info = &mut partition.log.journal_mut().info;
+ info.messages_count = 100;
+ info.current_offset = 99;
+ }
assert_eq!(
partition.applied_purge_generation(),
0,
diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs
index 6c43176f3..a370deac4 100644
--- a/core/partitions/src/log.rs
+++ b/core/partitions/src/log.rs
@@ -294,6 +294,28 @@ where
Some((segment, storage))
}
+ /// Retire the NEWEST segment, the mirror of [`Self::retire_front`].
+ ///
+ /// Boot re-anchor only, and only for an EMPTY tail named below the re-seeded
+ /// counter, which would otherwise claim a range it does not hold. Nothing
+ /// else may take from the back: a sized tail is the only copy of its
+ /// messages.
+ pub fn retire_back(&mut self) -> Option<(Segment, SegmentStorage)> {
+ if self.segments.is_empty() {
+ return None;
+ }
+ self.debug_assert_lockstep();
+ let segment = self.segments.pop()?;
+ let storage = self.storage.pop()?;
+ self.indexes.pop();
+ self.messages_writers.pop();
+ self.index_writers.pop();
+ self.sealed_read_state.pop();
+ self.sealed_lru
+ .retain(|&offset| offset != segment.start_offset);
+ Some((segment, storage))
+ }
+
fn debug_assert_lockstep(&self) {
debug_assert!(
self.segments.len() == self.storage.len()
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 2e0f00c91..88a63d552 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -1994,7 +1994,13 @@ where
let purge_advances = offsets_wire.purge_generation > committed_purge_generation
|| (self.applied_purge_generation < committed_purge_generation
&& offsets_wire.next_offset == 0);
- let local_next_offset = self.offset_frontier();
+ // 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();
if !purge_advances && local_next_offset > 0 && offsets_wire.next_offset < local_next_offset
{
return Err(PartitionInstallError::OfferRewindsDurableData {
@@ -2044,7 +2050,7 @@ where
self.reset_offset_frontier_at(offsets_wire.next_offset)
.await
} else {
- self.persist_offset_frontier_at(offsets_wire.next_offset)
+ self.install_offset_frontier_at(offsets_wire.next_offset)
.await
};
if !frontier_durable {
diff --git a/core/server/config.toml b/core/server/config.toml
index 2bac202b8..a4b94aef8 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -942,6 +942,14 @@ clients_table_max = 8192
# u128 bitset, and this depth bounds that suffix.
prepare_queue_depth = 32
+# How many offsets a partition claims in its superblock ahead of the mint
+# counter before it will append, so a crash-restarted replica resumes above
+# every offset it confirmed to a client instead of re-minting it for a different
+# 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.
+offset_reservation_lease = 65536
+
# Entries the evicted ring retains per multi-replica partition for journal
# repair after a peer rejoins. Larger widens the window a restarting peer can be
# served from the ring before falling back to bulk sync, at the cost of pinned
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 587744f5d..2dd073c82 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1901,6 +1901,7 @@ async fn build_shard_for_thread(
let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
let partition = match load_partition(
config,
+ partitions.config(),
namespace,
Arc::clone(&partition_stats),
&partition_metadata,
@@ -2577,6 +2578,7 @@ async fn recover_partition_segments(
#[allow(clippy::too_many_arguments)]
async fn load_partition(
config: &ServerConfig,
+ partitions_config: &PartitionsConfig,
namespace: IggyNamespace,
stats: Arc<PartitionStats>,
partition_metadata: &Partition,
@@ -2665,6 +2667,7 @@ async fn load_partition(
config.partition.evicted_ring_capacity,
config.partition.evicted_ring_bytes_max.as_bytes_u64(),
);
+ partition.set_offset_reservation_lease(config.partition.offset_reservation_lease);
partition.set_partition_dir(partition_dir.clone());
// Before the hydrate: the durable record is keyed by incarnation, so a
// `purge.gen` left behind by a previous life of this namespace reads 0.
@@ -2724,6 +2727,19 @@ async fn load_partition(
// gone (an all-GC'd origin's install, a crash inside the swap window), and
// taking the max means real recovered data always wins.
partition.restore_offset_frontier(recovered_state.as_ref());
+ // 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.
+ //
+ // 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.
+ if partition.consensus().replica_count() == 1 {
+ partition
+ .reanchor_to_offset_frontier(partitions_config)
+ .await
+ .map_err(|error| ServerError::Iggy(Box::new(error)))?;
+ }
let current_offset = partition.offset.load(Ordering::Acquire);
configure_consumer_offsets(&mut partition, config, namespace, current_offset)?;
@@ -4491,6 +4507,21 @@ mod tests {
);
}
+ #[test]
+ fn default_offset_reservation_lease_matches_common_constant() {
+ // `IggyPartition::new` falls back to the common constant (simulator,
+ // unit tests) while boot installs this one, so drift would have the
+ // fence write at a different rate in the simulator than in production.
+ let config_default =
+ configs::partition::PartitionConfig::default().offset_reservation_lease;
+ assert_eq!(
+ config_default,
+ iggy_common::DEFAULT_OFFSET_RESERVATION_LEASE,
+ "[partition] offset_reservation_lease default drifted from \
+ iggy_common::DEFAULT_OFFSET_RESERVATION_LEASE"
+ );
+ }
+
#[test]
fn default_evicted_ring_capacity_matches_partitions_constant() {
// Belt and suspenders with the static assert above; this pins the
diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs
index 5a2f49fd2..2f2e34252 100644
--- a/core/server/src/partition_helpers.rs
+++ b/core/server/src/partition_helpers.rs
@@ -630,6 +630,7 @@ pub async fn build_partition_fresh(
config.partition.evicted_ring_capacity,
config.partition.evicted_ring_bytes_max.as_bytes_u64(),
);
+ partition.set_offset_reservation_lease(config.partition.offset_reservation_lease);
partition.set_partition_dir(partition_dir);
// Fresh dirs read generation 0; a dir surviving from a crashed process
// (this "fresh" build races repair re-materialization) reads the last
@@ -747,6 +748,7 @@ mod tests {
checkpoint_op: 0,
checkpoint_checksum: 0,
offset_frontier: 0,
+ offset_reserved: 0,
}
}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 1efaa598a..f54895616 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -6602,6 +6602,20 @@ where
%error,
"failed to flush partition journal on shutdown"
);
+ // 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.
+ 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"
+ );
}
}
}