fix(partitions): reserve offsets before confirming them - #3975
fix(partitions): reserve offsets before confirming them#3975krishvishal wants to merge 4 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #3975 +/- ##
============================================
+ Coverage 85.00% 85.03% +0.03%
Complexity 1402 1402
============================================
Files 1225 1227 +2
Lines 180283 181359 +1076
Branches 146587 147664 +1077
============================================
+ Hits 153248 154224 +976
- Misses 22993 23060 +67
- Partials 4042 4075 +33
🚀 New features to boost your workflow:
|
| #[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(); |
There was a problem hiding this comment.
offset_frontier() is the append counter, so a boot mint floor that was never consumed is erased here.
Reproduced on this branch: crash, restart, clean stop, restart, produce — offset 4 was ACKed before the SIGKILL and the first send after the clean restart is confirmed at offset 0 again. The graceful stop is what destroys the protection; a second SIGKILL would have preserved it, so the runbook-correct incident response is the one action that defeats the fix. Nothing logs the moment.
Suggest:
let frontier = self
.offset_frontier()
.max(self.armed_mint_floor())
.max(self.durable_offset_frontier.get());The armed_mint_floor term covers the case above. The durable_offset_frontier term is a separate fix: reset_offset_frontier_at is the non-monotone writer, so in the failed-install window this currently regresses the recorded frontier.
| /// 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); |
There was a problem hiding this comment.
The flag is spent at mint time, but every failure path after the mint discards the mint itself: the count == 0 return, the checked_add, stamp_prepare_for_persistence's ?, the reservation fence, and the journal append all return before dirty_offset is stored.
The next append then mints dirty_offset + 1 — the pre-crash value — while reserve_offsets_through short-circuits on durable_offset_reserved > end_offset and writes nothing. So one transient superblock error on the first append after a crash-restart re-opens the exact defect this PR closes, silently.
Suggest disarming where dirty_offset is stored instead. That point is past every early return, and it also covers the ? in stamp_prepare_for_persistence, which sits after the mint and is not covered today.
| // 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 { |
There was a problem hiding this comment.
This two-fsync atomic_replace runs inline in the shard's sequential frame pump, and the consensus tick is a select_biased! arm on that same pump — its own comment bounds tick delay to one main frame body's longest .await. A slow superblock write therefore withholds heartbeats for every group on the shard, not only this partition, and peers can elect around the node. It also sits ahead of send_prepare_ok on the backup path, so it is on the primary's commit-latency path too.
The LOCK ORDER note above is correct about lock ordering but understates the blast radius.
Worth saying that the throughput cost is negligible and the arithmetic in the description checks out: roughly 3 fsyncs/s at 100k msg/s with the 64Ki lease, about 15 ns/message amortized. The concern is only where the stall lands. Extending the reservation off the critical path at a watermark, keeping this fence as the backstop, would preserve the guarantee without putting the write inside the pump.
| 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 { |
There was a problem hiding this comment.
When the unlink loop empties the chain, last() is None, needs_plant is false, and this returns without planting. ensure_initial_segment then plants at offset_frontier() and knows nothing about armed_mint_floor(), so the first mint lands at the reservation inside a segment named for the frontier — the hole-inside-a-segment shape the re-anchor exists to prevent.
It is tolerated while the index survives, so it is invisible in the common case. With a torn index (enforce_fsync = false is the shipped default) the index-less walk hits base_offset != expected_offset and raises OffsetDiscontinuity, tombstoning the partition.
This is the shape given_confirmed_sends_below_flush_threshold_when_a_solo_node_is_killed_should_not_remint_offsets produces; its assertion only checks the offset moved forward, so it passes over the segment shape. build_partition_fresh has the same gap — it arms the floor via set_superblock but never re-anchors before ensure_initial_segment.
Suggest planting here when the chain is empty and frontier > 0, or passing the floor into ensure_initial_segment.
| // the re-mint. The ceiling is what bounds the leniency: a stray file | ||
| // splicing a gap ABOVE anything this replica claimed still refuses. | ||
| if previous.end_offset.checked_add(1) != Some(next.start_offset) | ||
| && next.start_offset > offset_reserved |
There was a problem hiding this comment.
write_superblock_inner clamps offset_reserved up to offset_frontier, and advanced_frontier floors the frontier at held_offset_frontier(). offset_reserved therefore sits at or above the end of every legitimate segment, which makes next.start_offset > offset_reserved false for any real gap — so Hole can no longer fire for a partition that has ever appended. A genuinely lost middle segment now recovers silently as a holed log instead of refusing loudly, and the admitted arm logs nothing.
Narrowing the predicate is harder than it first looks. The gaps accumulate, one per crash. Probed on this branch, segment base offsets across successive crash cycles:
[0]
[0, 65537]
[0, 65537, 131074]
[0, 65537, 131074, 196611]
So an "at most one gap" rule would tombstone a healthy partition by the second crash, and given_a_crash_restarted_node_when_it_flushes_and_crashes_again_should_still_not_remint_offsets already reaches the two-gap state. Pinning next.start_offset == offset_reserved is unstable for a different reason: the first append after the plant extends the reservation past it, so the same pair fails on the following boot.
A single monotone scalar may not be able to separate re-anchor gaps from damage across repeated cycles. Recording the anchor points durably, or making the planted segment self-describing (carrying its predecessor's end offset), would give the guard something stable to check against.
6a90640 to
d21c33a
Compare
hubcio
left a comment
There was a problem hiding this comment.
the crash shapes reproduce on this branch with curl against a solo node: a crash below the flush threshold then a clean restart re-mints offset 0 or tombstones the partition, and a crash, one send, and a clean restart tombstones on the next boot. inline comments have the details.
outside the diff: init_partition in core/shard/src/lib.rs:3636 arms the floor on every simulator restart with no re-anchor, so the simulator diverges from the server here. the poll ceiling at core/partitions/src/iggy_partition.rs:1788-1806 serves an uncommitted offset 0 before the first commit - pre-existing, but it is how the spent-floor re-mint reaches consumers. the primary apply error at iggy_partition.rs:2676-2688 has no rollback_pipelined_prepare on the partitions plane - pre-existing, this fence is the new trigger. the doc on persist_offset_frontier_at at iggy_partition.rs:1122 still describes the install use it lost.
| #[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(); |
There was a problem hiding this comment.
after a crash-restart with no sends the counter is still below the armed floor, so a clean stop writes (0,0) or a reservation under the planted segment - next boot re-mints offset 0 or refuses the chain. collapse to offset_frontier().max(armed_mint_floor()).
| /// | ||
| /// 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 |
There was a problem hiding this comment.
ensure_initial_segment plants at offset_frontier(), still 0 here, while the first mint lands at the floor - the hole inside a segment this doc calls unsurvivable; the boot after the first flush refuses it. plant the empty chain at frontier here too.
| config.partition.evicted_ring_capacity, | ||
| config.partition.evicted_ring_bytes_max.as_bytes_u64(), | ||
| ); | ||
| partition.set_offset_reservation_lease(config.partition.offset_reservation_lease); |
There was a problem hiding this comment.
same hole on this path: set_superblock arms the floor from the surviving offset_reserved, ensure_initial_segment plants at the frontier, and the re-anchor never runs here. the first mint lands a lease block into a segment named below it, no crash needed.
| // 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()); |
There was a problem hiding this comment.
mint_floor() clears the flag before the fence or the journal append can fail, so the retry mints below the reservation and the fence fast path waves it through - a confirmed offset re-minted. clear the flag next to dirty_offset.store instead.
| // the re-mint. The ceiling is what bounds the leniency: a stray file | ||
| // splicing a gap ABOVE anything this replica claimed still refuses. | ||
| if previous.end_offset.checked_add(1) != Some(next.start_offset) | ||
| && next.start_offset > offset_reserved |
There was a problem hiding this comment.
this also admits overlaps and missing middle segments, since every real start sits below the reservation - after a clean stop, the whole chain. only accept a forward gap on the last pair, and pass 0 for replicated groups.
| 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?; |
There was a problem hiding this comment.
purge fsyncs the dir after its plant and says why; this one does not. a line on why the floor makes it unnecessary would save the next reader the question.
| /// 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 |
There was a problem hiding this comment.
seed the solo counter from max(offset_frontier, offset_reserved) here (gated on replica_count() == 1): offset_frontier() is then right for the re-anchor, ensure_initial_segment and the collapse at once, and the mint-floor state machine goes away. keep held_offset_frontier bytes-based so a later 1 -> N topology change does not inherit an inflated frontier.
| 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]; |
There was a problem hiding this comment.
this seal-and-plant is rotate_segment minus reset_read_state. a rotate_segment_at(config, start) shared by both keeps one seal path and its create-then-teardown order.
| /// 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) { |
There was a problem hiding this comment.
u64::from(lease) and drop const - the cast only exists to dodge cast_lossless. keep the zero floor, unit tests and the simulator set this directly.
| assert_eq!(bytes_of(&next_index_path), next_index); | ||
| } | ||
|
|
||
| #[compio::test] |
There was a problem hiding this comment.
this pins the over-broad rule (any hole below the reservation); once the guard only admits the last pair's forward gap, this test needs the tighter shape, plus an overlap case and a mid-chain gap case.
`restore_offset_frontier` is public, so its intra-doc link to the private `mint_floor` is a hard rustdoc error under `-D warnings`. It failed the pre-merge Rust lane before clippy, machete or any test job got a runner, hiding two more denials behind it: the lease default widened with an `as` cast, and `armed_mint_floor` reads only Copy state, so both trip the pedantic set.
`reanchor_to_offset_frontier` seals the recovered tail and plants the next segment at the restored frontier, betting that "every reader copes" with a hole on a segment boundary. `ensure_contiguous_chain` does not: it walks consecutive planned bounds and demands `end + 1 == next start`, so the boot AFTER the one that stopped the re-mint refused its own chain and tombstoned the partition, which sends then time out against. `offset_reserved` is what separates the two shapes. It already means "offsets this replica may have minted", so a gap ending inside it names offsets no file was ever meant to hold. A gap reaching past the claimed ceiling is still a stray file, and still refuses. Threading the ceiling in put `load_persisted_segments` one argument over the lint ceiling and `load_partition` one line over it, so the namespace triple collapses into the `IggyNamespace` every caller already holds and the offset-counter restore moves out to its own function.
2bfc229 to
7b66cfc
Compare
| @@ -2601,7 +2618,6 @@ async fn load_partition( | |||
| .max() | |||
| .filter(|&start| sized_end.is_none() && start > 0); | |||
| let current_offset = sized_end.or_else(|| empty_frontier.map(|start| start - 1)); | |||
There was a problem hiding this comment.
an empty chain's name used to be a committed frontier; the re-anchor and ensure_initial_segment now plant it at mint_frontier(), which is a reservation. two crashes below the flush threshold then store committed offset 65537 for a partition holding nothing, and store_consumer_offset admits the whole hole. bound the promotion by recovered_state.offset_frontier, which a real install writes at the group frontier and a re-anchor plant leaves far below.
| ); | ||
| self.offset.store(recovered_end, Ordering::Release); | ||
| self.dirty_offset.store(recovered_end, Ordering::Relaxed); | ||
| self.should_increment_offset = true; |
There was a problem hiding this comment.
should_increment_offset now means the append counter is live, but offset_frontier() still reads it as the committed counter naming data, so it returns 1 for a partition holding nothing. a separate bit for the committed seed stops that; it does not help the boot path above, which rebuilds from the file name.
| sealed_end, | ||
| }; | ||
| if let Err(error) = | ||
| crate::segment_anchor::write_anchor(&partition_dir, frontier, anchor).await |
There was a problem hiding this comment.
anchors have no lifecycle - every unlink site omits .anchor, including the quarantine and converge sweeps that filter a directory listing to .log/.index/.staging. purge resets the offset space to 0 and leaves the anchor there, where two of covers's three checks match for free and only the far-side offset has to coincide. add ANCHOR_EXTENSION to those sets.
| /// to the guard, so the caller needs no distinction. | ||
| pub async fn read_anchor(partition_dir: &str, start_offset: u64) -> Option<SegmentAnchor> { | ||
| let path = anchor_path(partition_dir, start_offset); | ||
| let bytes = compio::fs::read(&path).await.ok()?; |
There was a problem hiding this comment.
.ok()? turns an EACCES or EIO into no gap intended, so a healthy planted chain is refused as Hole - a solo group goes dark for the life of the process and every boot while the error lasts, a replicated one re-transfers the whole partition over one bad read. file_len fail-stops on exactly this class; do the same and treat NotFound separately.
| let Some(superblock) = self.superblock.as_ref().map(Rc::clone) else { | ||
| return true; | ||
| }; | ||
| if self.superblock_write_is_backed_off() { |
There was a problem hiding this comment.
one failed superblock write by any writer on this partition - a purge frontier reset, the tick's own extension - opens a 20 ms window, up to 1 s after repeats, where an already-admitted send whose mint crosses the lease boundary sets fatal and takes the node down. rare at the default lease, routine at a small one. give it the grace superblock_wedged grants.
| padded[..ENCODED_LEN_WITHOUT_FRONTIER].copy_from_slice(bytes); | ||
| ENCODED_LEN_WITHOUT_RESERVATION => { | ||
| padded[..ENCODED_LEN_WITHOUT_RESERVATION].copy_from_slice(bytes); | ||
| padded[66..74].copy_from_slice(&bytes[58..66]); |
There was a problem hiding this comment.
the 66 here is a literal two lines after ENCODED_LEN_WITHOUT_RESERVATION names it. padded[ENCODED_LEN_WITHOUT_RESERVATION..ENCODED_LEN] is the whole fix - the 58 has no constant and numeric offsets are this file's convention.
| .await | ||
| } else { | ||
| self.persist_offset_frontier_at(offsets_wire.next_offset) | ||
| self.install_offset_frontier_at(offsets_wire.next_offset) |
There was a problem hiding this comment.
does this reset buy anything today? the fence never runs above one replica, so on every group that can receive an offer the reservation already equals the frontier and advanced_frontier(offer) is never below it - the write is identical to the advancing form's.
| match intended { | ||
| Some(frontier) => { | ||
| self.write_superblock_inner(superblock.as_ref(), frontier) | ||
| self.write_superblock_inner(superblock.as_ref(), frontier, frontier) |
There was a problem hiding this comment.
this writes (f, f) with no max against durable_offset_reserved, which contradicts the monotone ceiling vsr_state.rs:106 claims. inert today - the only Some caller is the replicated ConvergeFailed arm, where the fence never ran and reserved == frontier. carry the max anyway?
| ); | ||
| return Err(ConfigurationError::InvalidConfigurationValue); | ||
| } | ||
| if self.offset_reservation_lease == 0 |
There was a problem hiding this comment.
the validation landed but mod tests has no zero or above-ceiling case for this knob, while prepare_queue_depth and evicted_ring_capacity each keep theirs.
| } | ||
|
|
||
| let client = wait_until_serving(harness, SERVE_TIMEOUT).await; | ||
| #[iggy_harness(cluster_nodes = 1)] |
There was a problem hiding this comment.
no test combines a flush with a graceful stop - :220 stops cleanly but takes no traffic between lives, and :299 flushes then re-enters through SIGKILL. ensure_initial_segment also never runs with a non-zero frontier: the only superblock fixture hardcodes offset_reserved: 0.
| // wasteful, but the weaker code would claim an unknown outcome | ||
| // for a request that provably has none. | ||
| if message.header().operation == Operation::SendMessages | ||
| && let Some(ceiling) = self.request_mint_ceiling(&message) |
There was a problem hiding this comment.
this never fires on an ordinary send - the convert above passes ChecksumMode::Skip, which zeroes batch_checksum, so request_mint_ceiling's verifying decode returns None. only the encrypt re-entry reaches the fence, so with encryption off every fence failure lands at the mint instead, where on_replicate takes the node down.
| let body = message | ||
| .as_slice() | ||
| .get(std::mem::size_of::<RoutedRequestHeader>()..message.header().size as usize)?; | ||
| let count = decode_batch_slice(body).ok()?.message_count(); |
There was a problem hiding this comment.
decode_batch_slice verifies, so this hashes every message body again just to read message_count, a u32 sitting in the batch header - a second full pass per send on top of the one admit_wire_request already did. use BatchHeader::decode(body), and check replica_count() before any of it.
The defect
A solo node ACKs sends from its in-memory journal. The client receives a concrete base offset before the threshold-gated flush writes it to a segment. A SIGKILL in this window loses the only record that the offset was issued.
Two messages now share offset 4. Consumers positioned above the reissued range also miss the new messages.
offset_frontieralready existed in the superblock, but stable-view traffic never persisted it because its write gate only ran on view changes.The fix
Add
offset_reserved, a monotonic ceiling on offsets that may have been issued. Before an offset can escape, the append fence reserves its block in the superblock. On boot, minting starts at this ceiling.The reservation covers both primary mints and backup re-stamps. It requires one write per block, not per batch. With the default 64 Ki lease at 100,000 messages per second, this is about three fsyncs per second. A crash wastes at most one block of the
u64offset space.The lease is configurable through
[partition] offset_reservation_lease. A failed reservation rejects the append.Why a second field?
Combining the fields would make valid transfer offers inside the reserved block look like rewinds. The rewind guard must instead compare against stored data: sized segments plus the resident journal. The append counter may be one lease block ahead after recovery.
Segment re-anchoring
A gap inside a segment is not recoverable.
recover_segment_boundsexpects contiguous offsets and truncates everything after a gap, allowing another crash to reissue confirmed offsets.An empty tail that falsely claims a range is removed. A sized tail is sealed, and a new segment is created at the frontier. Graceful shutdown collapses the reservation to the frontier, so only crashes spend offsets.
This applies only to solo groups. A backup rejects prepares whose
base_offsetdoes not continue its counter.Upgrading and rolling back
The superblock record grows from 66 to 74 bytes. Upgrading is transparent: a 66-byte record decodes with the reservation seeded from the frontier it carries. Rolling back is not, since a build that predates the field rejects a 74-byte record, so a downgrade needs the data directory wiped on every node.
Anyone running a build from an EARLIER push of this branch must also wipe their data directory before running this one. Those builds planted re-anchor gaps without the
.anchorrecord the chain guard now requires, so this build reads such a chain as a lost segment and the solo arm tombstones the partition.masterplants no gaps and edge images come only frommaster, so nothing deployed is affected.