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 <[email protected]> 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" + ); } } }
