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 4be4ac52e04f6ea2270085cf873234bacaad6109 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Sep 4 10:49:54 2026 +0200 fix(partitions): claim the first offset block at partition create The first send to a solo partition with no reservation on disk was answered TransientNotAccepted so the shard tick would claim the block off the request pump. That assumed every producer replays the transient. The binary SDKs do; the HTTP plane does not. The acked route has no replay loop, so it surfaced the bounce as HTTP 503, and `?ack=none` never reads a reply at all, so it answered 202 and dropped the message. Deterministic, once per partition: a topic with N partitions silently lost the first message to each. Claim the block where the partition is created instead. That path already pays for a superblock write, so the claim costs a new partition one extra atomic replace and takes the transient off every plane at once. The objection the old design raised against eager claiming is about boot, where many idle partitions would each pay a write for nothing; claiming at create does not touch boot, and the tick's trigger stays gated on a partition that has minted. The cost is offset space after an unclean stop: a partition created and never produced to now resumes a lease block above zero, the same hole the reservation already leaves after a crash mid-produce. A graceful stop collapses it. The simulator's `claim_partition_offset_block` stood in for the tick that the bounce required, and goes with it. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01K8yxwD21a8EzjyxFHLiCLE --- core/integration/tests/server/http_vsr.rs | 89 +++++++++++++++++ core/partitions/src/iggy_partition.rs | 152 ++++-------------------------- core/server/src/partition_helpers.rs | 30 +++++- core/simulator/src/lib.rs | 41 -------- 4 files changed, 135 insertions(+), 177 deletions(-) diff --git a/core/integration/tests/server/http_vsr.rs b/core/integration/tests/server/http_vsr.rs index 2ae84bac2..c97be5a65 100644 --- a/core/integration/tests/server/http_vsr.rs +++ b/core/integration/tests/server/http_vsr.rs @@ -735,6 +735,95 @@ async fn given_ack_none_when_producing_should_return_202_and_commit(harness: &Te } } +/// The FIRST send to a partition that has never minted an offset, on both +/// produce routes, against the only cluster shape where the offset reservation +/// runs at all. +/// +/// `cluster_nodes = 1` is load-bearing, not a speed-up: `request_mint_ceiling` +/// returns `None` above one replica, so this suite's three-node default leaves +/// the whole reservation path as dead code and proves nothing here. +/// +/// The reservation writes the partition's superblock before it hands out an +/// offset, and a first send is where that claim is missing. Neither HTTP route +/// can carry a retryable refusal back to the caller: the acked route has no +/// transient replay loop, and `?ack=none` never reads a reply at all, so a +/// refusal there would answer 202 and drop the message. Both partitions are +/// produced to exactly once, so a per-partition regression cannot hide behind a +/// second send. +#[iggy_harness(cluster_nodes = 1)] +async fn given_a_solo_topic_when_producing_its_first_http_messages_should_commit_them( + harness: &TestHarness, +) { + const ACKED_PARTITION: u32 = 0; + const UNACKED_PARTITION: u32 = 1; + + let http = HttpClient::login_root(harness).await; + http.create_stream_and_topic("http-first-send", "first", 2) + .await; + + let response = http + .produce( + "http-first-send", + "first", + ACKED_PARTITION, + vec![text_message(1, "first-acked".to_string())], + ) + .await; + assert_eq!( + response.status(), + StatusCode::CREATED, + "the first acked send to a never-minted partition must commit, not be refused" + ); + let polled = http + .poll("http-first-send", "first", ACKED_PARTITION, 0, 10) + .await; + assert_eq!(polled.messages.len(), 1, "the first acked send is durable"); + assert_eq!( + polled.messages[0].payload, + bytes::Bytes::from("first-acked") + ); + + let response = http + .produce_with_query( + "http-first-send", + "first", + UNACKED_PARTITION, + vec![text_message(2, "first-unacked".to_string())], + "?ack=none", + ) + .await; + assert_eq!( + response.status(), + StatusCode::ACCEPTED, + "ack=none must answer before the commit" + ); + + // 202 says nothing about the commit, which is the whole hazard: a refusal + // on this route is answered the same way and leaves no trace. Only the poll + // proves the message survived. + let deadline = Instant::now() + ASYNC_COMMIT_TIMEOUT; + loop { + if let Some(polled) = http + .try_poll("http-first-send", "first", UNACKED_PARTITION, 0, 10) + .await + && !polled.messages.is_empty() + { + assert_eq!(polled.messages.len(), 1, "exactly one message was produced"); + assert_eq!( + polled.messages[0].payload, + bytes::Bytes::from("first-unacked"), + "the first ack=none send to a never-minted partition must not be dropped" + ); + break; + } + assert!( + Instant::now() < deadline, + "the first ack=none send never became pollable within {ASYNC_COMMIT_TIMEOUT:?}" + ); + sleep(ASYNC_COMMIT_RETRY_INTERVAL).await; + } +} + /// End-to-end RBAC proof: an ungranted user is 403 on a metadata read and on a /// data-plane produce, root stays 200/201, and the auth-only cluster-metadata /// route is never gated. Exercises the HTTP per-op gates (read + partition diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index ab16b256f..a8976bb1e 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -261,16 +261,6 @@ where /// cost nothing. Kept apart from [`Self::durable_offset_frontier`]: see /// `consensus::VsrState::offset_reserved`. durable_offset_reserved: Cell<u64>, - /// A send arrived for a partition with no reservation on disk yet, and was - /// bounced so the shard tick could claim the first block off the request - /// path. Cleared by the write it asks for. - /// - /// The trigger the tick consults is otherwise gated on a partition that has - /// already minted, deliberately: arming every idle partition at boot would - /// write a superblock per partition for nothing. This bit is what separates - /// "idle" from "wanted", so the cost falls only on partitions someone - /// actually produced to. - offset_reservation_wanted: Cell<bool>, /// Offsets the append fence claims per superblock write; installed by boot /// from `PartitionsConfig`. offset_reservation_lease: u64, @@ -540,7 +530,6 @@ where purge_deferred: false, durable_offset_frontier: Cell::new(0), durable_offset_reserved: Cell::new(0), - offset_reservation_wanted: Cell::new(false), offset_reservation_lease: u64::from(crate::DEFAULT_OFFSET_RESERVATION_LEASE), transfer: None, transfer_attempts: 0, @@ -1425,19 +1414,15 @@ where /// costs nothing but offset space, while arriving late puts the write back on /// the append path. Floored at 1, since validation admits a lease of 1 and /// `1 / 2` would never trigger, leaving every append to pay the inline claim. - /// A partition that has never minted is skipped unless a send has already - /// been bounced for it (`should_defer_first_reservation`): extending - /// every idle partition at boot would write a superblock per partition for - /// nothing, while a partition someone is producing to needs its first block - /// claimed off the request path like every later one. + /// 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. #[must_use] pub fn needs_offset_reservation_extension(&self) -> bool { if self.consensus.replica_count() > 1 || self.superblock.is_none() { return false; } - if self.offset_reservation_wanted.get() { - return true; - } if !self.offset_space.append_live { return false; } @@ -1448,38 +1433,6 @@ where headroom < (self.offset_reservation_lease / 2).max(1) } - /// Whether this send should be BOUNCED so the shard tick claims the - /// partition's first block, rather than paying for it inline. - /// - /// Without this the first append to every untouched solo partition awaits a - /// create, write, file fsync, rename and directory fsync inside the shard's - /// request pump, where the consensus tick is a sibling arm. A - /// high-cardinality first-write burst serializes those fences and delays - /// unrelated group work and heartbeats on the same core. - /// - /// One retry, once in a partition's life: the bounce is - /// `TransientNotAccepted`, which admitted nothing, and by the time the client - /// re-sends, the tick has claimed the block and the fence takes its fast - /// path. - /// - /// `false` once a claim covers the batch, and `false` for a first batch wider - /// than the whole lease -- the tick's claim would not cover that one either, - /// so bouncing it would bounce the same request forever. - /// - /// `false` with no store attached, for the same reason. A storeless partition - /// (in-memory, simulated) reserves nothing at all, and - /// [`Self::needs_offset_reservation_extension`] skips it, so nothing would - /// ever clear the bounce: every first send would be denied for the life of - /// the partition. The gates here and there must agree on which partitions the - /// tick can serve. - #[must_use] - const fn should_defer_first_reservation(&self, end_offset: u64) -> bool { - self.superblock.is_some() - && !self.offset_space.append_live - && self.durable_offset_reserved.get() <= end_offset - && end_offset < self.offset_reservation_lease - } - /// Extend the reservation a full block past the CEILING already on disk. /// /// Pairs with [`Self::needs_offset_reservation_extension`]; the caller is the @@ -1504,14 +1457,7 @@ where } let _superblock_guard = self.superblock_lock.acquire().await; let ceiling = self.durable_offset_reserved.get().max(self.mint_frontier()); - let written = self.write_claim_from(superblock.as_ref(), ceiling).await; - if written { - // Only on success: a failed write leaves the bounce standing so the - // next tick retries it, rather than dropping the partition back to - // paying inline. - self.offset_reservation_wanted.set(false); - } - written + self.write_claim_from(superblock.as_ref(), ceiling).await } /// Upper bound on the offsets a pending `SendMessages` request will mint, for @@ -1602,6 +1548,9 @@ where /// serving. At the mint the op already has its number and its ack is already /// skipped, so `commit_max` can never pass it and nothing later can commit /// either: `on_replicate` fences the partition there and takes the node down. + /// 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. #[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 { @@ -1659,9 +1608,8 @@ where /// and must go no further: the client holds a `TransientNotAccepted`, which /// admitted nothing, so it may re-issue anywhere without double-apply risk. /// - /// Three ways to come back `false`, none of them reaching the mint: a bounced - /// first send, an open superblock backoff window, and a claim that was - /// attempted and failed. + /// Two ways to come back `false`, neither reaching the mint: an open + /// superblock backoff window, and a claim that was attempted and failed. /// /// `waiter` is the submit's in-process reply channel, taken only on a /// refusal: the deny goes there because `header.client` is then the VSR @@ -1678,16 +1626,6 @@ where let Some(ceiling) = self.request_mint_ceiling(message) else { return true; }; - if self.should_defer_first_reservation(ceiling) { - self.offset_reservation_wanted.set(true); - self.deny_unreserved_send( - message.header(), - "bouncing a partition's first send so the tick claims its offset block", - waiter.take(), - ) - .await; - return false; - } if !self.reserve_offsets_through_retryable(ceiling).await { self.deny_unreserved_send( message.header(), @@ -6865,78 +6803,24 @@ mod tests { } } - /// The first send to an untouched partition is BOUNCED so the tick claims the - /// block, rather than awaiting a create, write, fsync, rename and directory - /// fsync inside the shard's request pump. - #[compio::test] - async fn given_an_untouched_partition_when_a_send_arrives_should_bounce_it_to_the_tick() { - let store = Rc::new(RecordingSuperblock::default()); - let mut partition = solo_recording_partition(); - partition.set_superblock(store.clone(), None); - partition.set_offset_reservation_lease(test_lease(16)); - - assert!( - partition.should_defer_first_reservation(0), - "a first send with no claim on disk must not pay for one inline" - ); - assert!( - !partition.should_defer_first_reservation(16), - "a first batch wider than the whole lease must not be bounced: the \ - tick's claim would not cover it either, so it would bounce forever" - ); - - // Arming is what the bounce does; the tick then writes, off this path. - partition.offset_reservation_wanted.set(true); - assert!( - partition.needs_offset_reservation_extension(), - "an armed partition needs the write even though it has never minted" - ); - assert!(partition.extend_offset_reservation().await); - assert_eq!(store.attempts.get(), 1); - assert!( - !partition.needs_offset_reservation_extension(), - "the write it asked for disarms it" - ); - - // The retry finds the block already claimed and writes nothing. - assert!(!partition.should_defer_first_reservation(0)); - assert!(partition.reserve_offsets_through(0).await); - assert_eq!( - store.attempts.get(), - 1, - "the bounced send's retry takes the fence's fast path" - ); - } - - /// A storeless partition reserves nothing, and the tick skips it, so a bounce - /// there is a send denied for the life of the partition with nothing able to - /// clear it. The two gates have to agree on which partitions the tick serves. + /// A storeless partition (in-memory, simulated) reserves nothing at all, so + /// the tick must never reach a write for one. #[test] - fn given_a_storeless_partition_when_a_send_arrives_should_not_bounce_it() { + fn given_a_storeless_partition_when_ticking_should_not_extend() { let mut partition = solo_recording_partition(); partition.set_offset_reservation_lease(test_lease(16)); assert!(partition.superblock.is_none(), "the premise: no store"); - assert!(!partition.offset_space.append_live); - - assert!( - !partition.should_defer_first_reservation(0), - "nothing would ever claim the block this bounce waits for" - ); - assert!( - !partition.needs_offset_reservation_extension(), - "and the tick agrees it has nothing to do here" - ); + assert!(!partition.needs_offset_reservation_extension()); } - /// Idle partitions stay idle: arming is what separates a partition someone - /// produced to from one boot merely materialized, and without that a node - /// with many partitions writes a superblock per partition for nothing. + /// Idle partitions stay idle: the first block is claimed where the partition + /// is created, and a node with many partitions must not write a superblock + /// per partition at boot for nothing. #[test] - fn given_an_unarmed_untouched_partition_when_ticking_should_still_not_extend() { + fn given_an_untouched_partition_when_ticking_should_not_extend() { let mut partition = solo_recording_partition(); partition.set_superblock(Rc::new(RecordingSuperblock::default()), None); assert!(!partition.offset_space.append_live); - assert!(!partition.offset_reservation_wanted.get()); assert!(!partition.needs_offset_reservation_extension()); } diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index f04d0d537..2c88403dd 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -1116,8 +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. Configure empty consumer-offset storage with the on-disk paths set. -/// 4. Provision the initial segment + writers (offset 0). +/// 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). /// /// The namespace arrives packed, so its components are in range by /// construction. Metadata admission is what bounds them. @@ -1296,6 +1297,31 @@ pub async fn build_partition_fresh( // frontier before quarantining, and the boot-path chain refusal to carry // 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. + // + // 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" + ); + } let current_offset = partition.offset.load(Ordering::Acquire); configure_consumer_offsets(&mut partition, config, namespace, current_offset)?; diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 36473b8dc..3d2fd3fea 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1324,43 +1324,6 @@ impl Simulator { Some(u64::from(partition.consensus().view())) } - /// Claim the first offset block for every live replica's copy of a - /// materialised solo partition, standing in for the shard tick that does it - /// in production. - /// - /// The append fence BOUNCES the first send to a partition with no claim on - /// disk, so the superblock write lands on the tick rather than inside the - /// request pump. A real client retries that `TransientNotAccepted`; the - /// simulator has no retry loop, so a scenario that produces to a freshly - /// materialised partition and expects the send to commit has to claim the - /// block first. - /// - /// Not folded into [`Self::init_partition`]: materialising a partition is not - /// the same event as producing to one, and arming every materialised - /// partition would model a superblock write per idle partition that - /// production deliberately does not make. - /// - /// # Panics - /// If the simulated superblock refuses the claim, which no scenario injects: - /// a silent skip would leave the caller's next send bounced with nothing to - /// say why. - #[allow(clippy::cast_possible_truncation)] - pub fn claim_partition_offset_block(&self, namespace: IggyNamespace) { - for (index, replica) in self.replicas.iter().enumerate() { - if self.crashed.contains(&(index as u8)) { - continue; - } - let shard = replica.partition_shard(namespace); - let Some(partition) = shard.plane.partitions().get_by_ns(&namespace) else { - continue; - }; - assert!( - futures::executor::block_on(partition.extend_offset_reservation()), - "the simulated superblock must accept the first offset claim" - ); - } - } - /// One replica's view of a partition group's consensus, or `None` when it does /// not host the namespace. Read by the quiesce oracle to decide whether a group /// has settled into one view, which its leader-relative checks depend on once @@ -2675,10 +2638,6 @@ mod tests { ); sim.init_partition(namespace); - // The tick's job in production. Without it the append fence bounces this - // first send so the superblock write stays off the request pump, and the - // simulator has no client to retry the bounce. - sim.claim_partition_offset_block(namespace); assert_eq!( shard.redispatched_frame_count(), 1,
