This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch durable-offset-watermark in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c22e79817176a4e779553a450c674f32aa9798c2 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Sep 4 11:21:30 2026 +0200 fix(partitions): refuse a create whose first offset claim fails A failed claim only logged, and the partition went live anyway. The failed write arms a 20 ms superblock retry backoff, so the first send inside that window is refused at the admitted path with a transient the HTTP plane does not replay: an acked produce answers 503 and an ack=none produce answers 202 with the message silently dropped. That is the defect the create-time claim was added to remove, reopened in a narrow window. Failing the build instead leaves the namespace unmaterialised, so the reconciler backs off and retries with a partition whose backoff cell starts clear, and produces park until a later pass succeeds. The claim also moves behind consumer-offset configuration and the initial segment, which restores the step list's idempotence: a build that failed after it used to leave the claim on disk and burn another lease block per retry. It is now skipped once the offset space is live, because a rebuild resumes its append point exactly on the reservation it recovered and would otherwise write and burn a block every time. Such a rebuild pays one inline fence on its first send, the same cost a graceful stop and boot already carries. The unit coverage asserted only on the returned partition, which the inline fence at the mint satisfies on its own; it now reads the durable record back. Two comments claimed the fresh-create path already paid for a superblock write, and it did not: before the claim, that path only read the superblock. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01K8yxwD21a8EzjyxFHLiCLE --- core/partitions/src/iggy_partition.rs | 13 ++- core/server/src/partition_helpers.rs | 177 +++++++++++++++++++++++----------- core/server/src/server_error.rs | 7 ++ 3 files changed, 137 insertions(+), 60 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index a8976bb1e..69d625579 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -1416,8 +1416,8 @@ where /// `1 / 2` would never trigger, leaving every append to pay the inline claim. /// A partition that has never minted is skipped: extending every idle /// partition at boot would write a superblock per partition for nothing, and - /// the first block is already claimed where the partition is created, on a - /// path that pays for a superblock write anyway. + /// the first block is already claimed where the partition is created, off + /// the append path. #[must_use] pub fn needs_offset_reservation_extension(&self) -> bool { if self.consensus.replica_count() > 1 || self.superblock.is_none() { @@ -1549,8 +1549,10 @@ where /// skipped, so `commit_max` can never pass it and nothing later can commit /// either: `on_replicate` fences the partition there and takes the node down. /// At CREATE (`build_partition_fresh`) nothing has been externalised at all, - /// so a refusal only drops the partition back to claiming its first block - /// inline on the append path. + /// so a refusal fails the build and leaves the namespace unmaterialised for + /// the reconciler to retry. Going live without the block instead would let + /// the first send land inside the backoff the failed write just armed, where + /// the admitted path refuses it with a transient. #[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 { @@ -6809,6 +6811,9 @@ mod tests { fn given_a_storeless_partition_when_ticking_should_not_extend() { let mut partition = solo_recording_partition(); partition.set_offset_reservation_lease(test_lease(16)); + // Without this the untouched offset space would satisfy the gate on its + // own and the store check would go untested. + partition.note_append_live(); assert!(partition.superblock.is_none(), "the premise: no store"); assert!(!partition.needs_offset_reservation_extension()); } diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 2c88403dd..b55aacf42 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -1116,9 +1116,9 @@ fn hydrate_reopen_error( /// Steps performed (all idempotent on retry after a partial failure): /// 1. Create directory hierarchy on disk. /// 2. Build per-partition VSR consensus group, resuming any superblock-recorded view. -/// 3. Claim the group's first offset-reservation block (solo groups with a store). -/// 4. Configure empty consumer-offset storage with the on-disk paths set. -/// 5. Provision the initial segment + writers (offset 0). +/// 3. Configure empty consumer-offset storage with the on-disk paths set. +/// 4. Provision the initial segment + writers (offset 0). +/// 5. Claim the group's first offset-reservation block (solo groups with a store). /// /// The namespace arrives packed, so its components are in range by /// construction. Metadata admission is what bounds them. @@ -1135,8 +1135,8 @@ fn hydrate_reopen_error( /// /// # Errors /// -/// Returns [`ServerError`] when directory creation, superblock recovery, or -/// segment provisioning fails. +/// Returns [`ServerError`] when directory creation, superblock recovery, +/// segment provisioning, or the first offset-reservation claim fails. #[allow(clippy::too_many_arguments)] pub async fn build_partition_fresh( config: &ServerConfig, @@ -1298,34 +1298,42 @@ pub async fn build_partition_fresh( // the refused chain's max `end_offset` on its error. partition.restore_offset_frontier(recovered_state.as_ref()); - // Claim the first offset-reservation block HERE, where this path is already - // paying for a superblock write, so no send ever pays the create, write, - // file fsync, rename and directory fsync of a first claim inline in the - // shard's request pump, where the consensus tick is a sibling arm. No-op - // above one replica and with no store attached, where nothing is reserved, - // and no-op on a rebuild whose record already covers the next mint. + let current_offset = partition.offset.load(Ordering::Acquire); + + configure_consumer_offsets(&mut partition, config, namespace, current_offset)?; + ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?; + + // Claim the first offset-reservation block HERE so no send ever pays the + // create, write, file fsync, rename and directory fsync of a first claim + // inline in the shard's request pump, where the consensus tick is a sibling + // arm. It is a NEW write on a path that otherwise only READS the superblock: + // one atomic replace per created partition, serialised with its siblings in + // the reconciler's addition loop, so it lengthens the window a produce + // arriving with the create spends parked. + // + // LAST of the steps, so the rest stay idempotent on retry: a failure between + // the claim and the return would burn a lease block per reconciler pass. + // + // No-op above one replica and with no store attached, where nothing is + // reserved. Skipped once the offset space is live (`mint_frontier` reads 0 + // only while it is not): a rebuild resumes its append point exactly ON the + // reservation it recovered, never above it, so an unconditional claim would + // write and burn a block every time. It pays one inline fence on its first + // send instead, which is what a graceful stop and boot already costs. // // The shard tick takes over from the first mint onward // (`needs_offset_reservation_extension`), which stays gated on a partition // that has minted so boot cannot write a superblock per idle partition. - if !partition - .reserve_offsets_through(partition.mint_frontier()) - .await - { - // Degraded, not fatal: the fence on the append path still claims - // inline, so the first send pays for the block instead of the create. - warn!( - stream_id, - topic_id, - partition_id, - "could not claim the partition's first offset reservation; its first send \ - will claim one inline" - ); + if partition.mint_frontier() == 0 && !partition.reserve_offsets_through(0).await { + // Not degraded-but-live: the failed write armed the group's superblock + // retry backoff, and `reserve_offsets_through_retryable` refuses every + // send arriving inside it with a transient the HTTP plane does not + // replay. The reconciler backs the namespace off and retries with a + // fresh partition, whose backoff cell starts clear. + return Err(ServerError::PartitionOffsetReservationClaim { + namespace_raw: namespace.inner(), + }); } - let current_offset = partition.offset.load(Ordering::Acquire); - - configure_consumer_offsets(&mut partition, config, namespace, current_offset)?; - ensure_initial_segment(&mut partition, config, stream_id, topic_id, partition_id).await?; Ok(partition) } @@ -1435,6 +1443,78 @@ mod tests { } } + /// The offset reservation is solo-only, so every test that touches it builds + /// under this identity. + const fn solo_identity() -> ReplicaIdentity { + ReplicaIdentity { + cluster: CLUSTER, + replica_id: 0, + replica_count: 1, + } + } + + fn solo_config(root: &tempfile::TempDir) -> ServerConfig { + ServerConfig { + system: Arc::new(ServerSystemConfig { + path: root.path().to_string_lossy().into_owned(), + ..ServerSystemConfig::default() + }), + ..ServerConfig::default() + } + } + + async fn build_solo_partition( + config: &ServerConfig, + ) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerError> { + build_partition_fresh( + config, + IggyNamespace::new(1, 1, 0), + Arc::new(PartitionStats::default()), + 0, + TopicRuntimeOptions::default(), + CLUSTER, + 0, + 1, + 0, + Rc::new(IggyMessageBus::new(0)), + ) + .await + } + + /// The reservation the partition left on disk, which is the only copy a + /// restart or a first send can read. + async fn recorded_reservation(dir: &str) -> u64 { + let (_store, recorded) = open_partition_superblock(dir, solo_identity()) + .await + .expect("reopen the partition superblock"); + recorded + .expect("a partition that recorded a reservation") + .offset_reserved + } + + /// The create claims the first lease block, so the DURABLE record covers the + /// first send before it arrives. Asserting on the returned partition alone + /// would pass with no claim at all: the inline fence at the mint writes the + /// same block on the first send, which is exactly what this moves off the + /// append path. + #[compio::test] + async fn given_a_fresh_solo_partition_when_building_should_record_its_first_claim() { + let root = tempfile::tempdir().expect("tempdir"); + let config = solo_config(&root); + let dir = config.system.get_partition_path(1, 1, 0); + + let partition = build_solo_partition(&config) + .await + .expect("build a fresh partition"); + drop(partition); + + assert_eq!( + recorded_reservation(&dir).await, + 1 + u64::from(config.partition.offset_reservation_lease.get()), + "the create must leave a full lease block covering offset 0 on disk" + ); + } + /// A rebuild that reads a reservation back must NAME its planted segment for /// the append point. Named 0, the segment takes the first append's /// `base_offset` of N instead, `rposition(|s| s.start_offset <= offset)` @@ -1444,22 +1524,10 @@ mod tests { async fn given_a_recorded_reservation_when_building_fresh_should_plant_at_the_append_point() { const RESERVED: u64 = 65_537; let root = tempfile::tempdir().expect("tempdir"); - let config = ServerConfig { - system: Arc::new(ServerSystemConfig { - path: root.path().to_string_lossy().into_owned(), - ..ServerSystemConfig::default() - }), - ..ServerConfig::default() - }; - let namespace = IggyNamespace::new(1, 1, 0); + let config = solo_config(&root); let dir = config.system.get_partition_path(1, 1, 0); - let identity = ReplicaIdentity { - cluster: CLUSTER, - replica_id: 0, - replica_count: 1, - }; - let (store, recovered) = open_partition_superblock(&dir, identity) + let (store, recovered) = open_partition_superblock(&dir, solo_identity()) .await .expect("open a fresh partition superblock"); assert!(recovered.is_none()); @@ -1469,20 +1537,9 @@ mod tests { .expect("record the reservation"); drop(store); - let partition = build_partition_fresh( - &config, - namespace, - Arc::new(PartitionStats::default()), - 0, - TopicRuntimeOptions::default(), - CLUSTER, - 0, - 1, - 0, - Rc::new(IggyMessageBus::new(0)), - ) - .await - .expect("rebuild the partition over its recorded reservation"); + let partition = build_solo_partition(&config) + .await + .expect("rebuild the partition over its recorded reservation"); assert_eq!( partition.mint_frontier(), @@ -1510,6 +1567,14 @@ mod tests { vec![format!("{RESERVED:0>20}.log")], "the initial segment must be named for the append point, not offset 0" ); + + drop(partition); + assert_eq!( + recorded_reservation(&dir).await, + RESERVED, + "a rebuild resumes ON its recorded reservation, so re-claiming here would \ + burn a lease block and two fsyncs per rebuild" + ); } #[compio::test] diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 7f0f86c5e..dee1fa0ac 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -201,6 +201,13 @@ pub enum ServerError { partition_id: usize, reason: PartitionRecoveryRefusal, }, + /// Fails the create rather than letting the partition go live without its + /// first reservation: the failed write arms the group's superblock retry + /// backoff, and a send arriving inside that window is refused with a + /// transient the HTTP plane does not replay. `namespace_raw` joins this to + /// the write's own `iggy.partitions.diag` line, which carries the cause. + #[error("partition namespace {namespace_raw} could not claim its first offset reservation")] + PartitionOffsetReservationClaim { namespace_raw: u64 }, #[error( "shard {shard_id} aborted while waiting for shard-0 to broadcast the metadata \ factory bundle; shard 0 dropped its sender (most likely it failed to recover)"
