diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 148c4078b9..945c0739e7 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -770,7 +770,24 @@ where self.offset.store(recovered_end, Ordering::Release); self.dirty_offset.store(recovered_end, Ordering::Relaxed); self.should_increment_offset = true; - self.stats.set_current_offset(recovered_end); + } + + /// Copy this incarnation's offset counter into the shared + /// [`PartitionStats`], making it the value readers (offset validation, + /// `get_topic`, `get_stats`) see. + /// + /// Called from [`IggyPartitions::insert`](crate::IggyPartitions::insert) + /// only: when the instance BECOMES the addressable one, never while + /// building it. The stats registry keys on the namespace, not the + /// incarnation, so every build of a namespace holds the same `Arc` as + /// whatever is already serving it -- and a build is not guaranteed to be + /// adopted. Seeding from the build instead leaves a zeroed `current_offset` + /// on the live incarnation, which then rejects every + /// `store_consumer_offset` above 0 with `InvalidOffset` until the next send + /// re-seeds it. + pub(crate) fn publish_current_offset(&self) { + self.stats + .set_current_offset(self.offset.load(Ordering::Acquire)); } /// The next message offset this replica will mint, `0` while the offset diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index a25aed657b..5fac54528a 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -200,6 +200,12 @@ where /// Insert a new partition and return its local index. /// + /// Insertion is the moment a build becomes the addressable incarnation, + /// so this is also where its offset counter is published into the shared + /// `PartitionStats` ([`IggyPartition::publish_current_offset`]). No + /// earlier point is safe: a build that is never adopted must leave the + /// live incarnation's counters alone. + /// /// # Safety discipline (compiler cannot enforce) /// /// Must only be called from the shard's pump task (i.e. inside @@ -220,6 +226,7 @@ where 0, "IggyPartitions::insert while a with_partition borrow is live" ); + partition.publish_current_offset(); // Safety: pump-only invariant, caller responsibility. let partitions = unsafe { &mut *self.partitions.get() }; let local_idx = LocalIdx::new(partitions.len()); diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index ad1671b201..b3bc66b353 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -2413,7 +2413,6 @@ async fn load_partition( partition.offset.store(counter, Ordering::Release); partition.dirty_offset.store(counter, Ordering::Relaxed); partition.should_increment_offset = current_offset.is_some(); - partition.stats.set_current_offset(counter); // The durable frontier is a LOWER BOUND on top of what the segments proved: // it is the only carrier left when the segments that named the frontier are // gone (an all-GC'd origin's install, a crash inside the swap window), and diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs index a7a291eb76..129ffb537c 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -694,7 +694,6 @@ pub async fn build_partition_fresh( partition.offset.store(0, Ordering::Release); partition.dirty_offset.store(0, Ordering::Relaxed); partition.should_increment_offset = false; - partition.stats.set_current_offset(0); debug_assert!( !partition.log.has_segments(), "fresh partition must not carry recovered segments" diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs index 63d6c2a5cb..cf6c86de0d 100644 --- a/core/server-ng/src/partition_reconciler.rs +++ b/core/server-ng/src/partition_reconciler.rs @@ -413,6 +413,10 @@ struct PassCounters { /// tombstone and re-wakes us without bumping `Streams::revision`, so an /// armed skip would swallow that wake and strand the rebuild forever. deferred: usize, + /// Namespaces an earlier pass already built, whose `InsertOwned` the pump + /// has not applied yet. Counted so the pass does not arm the fast-skip + /// while work is in flight; applying it bumps no revision. + already_staged: usize, } impl PassCounters { @@ -428,6 +432,7 @@ impl PassCounters { + self.purges_staged + self.deferred + self.parked_reclaimed + + self.already_staged } } @@ -476,9 +481,9 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { let target_set: AHashSet = target.iter().map(|(ns, _)| *ns).collect(); let mut counters = PassCounters::default(); - let staged = reconcile_additions(ctx, target, &mut counters).await; + reconcile_additions(ctx, target, &mut counters).await; reconcile_removals(ctx, &target_set, &mut counters).await; - reconcile_parked_frames(ctx, &staged, &mut counters); + reconcile_parked_frames(ctx, &mut counters); reconcile_consumer_group_offsets(ctx, &mut counters).await; reconcile_segment_truncations(ctx, &mut counters); reconcile_partition_purges(ctx, &mut counters); @@ -506,6 +511,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { backoff_skipped = counters.backoff_skipped, stale = counters.stale, deferred = counters.deferred, + already_staged = counters.already_staged, parked_reclaimed = counters.parked_reclaimed, purges_staged = counters.purges_staged, trims_pending = counters.trims_pending, @@ -521,19 +527,14 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { true } -/// Returns the namespaces whose `ReconcileOp::InsertOwned` this pass staged. The -/// pump applies the op on its own task, so they are not in `IggyPartitions` yet -/// and [`reconcile_parked_frames`] would read them as un-materialised, aging -/// their frames on the pass that built them. async fn reconcile_additions( ctx: &ReconcilerCtx, target: Vec<(IggyNamespace, u64)>, counters: &mut PassCounters, -) -> AHashSet { +) { let shard_id = ctx.shard.id; let partitions = ctx.shard.plane.partitions(); let total_shards = u32::from(ctx.total_shards); - let mut staged = AHashSet::new(); for (ns, epoch) in target { if partitions.contains(&ns) { @@ -583,7 +584,7 @@ async fn reconcile_additions( // means the local partition is a prior incarnation carrying // stale segments/offsets/log. Tear it down; the // post-ConfirmRemove wake rebuilds it fresh next pass. - if ctx.shard.shards_table().epoch_for(ns) == Some(epoch) { + if shards_table_has_epoch(ctx, ns, epoch) { continue; } trace!( @@ -599,10 +600,15 @@ async fn reconcile_additions( let owning_shard = calculate_shard_assignment(&ns, total_shards); if owning_shard != shard_id { - // Compare the epoch, not just presence: a delete + recreate recycles - // the slab keys, so the row survives with the DEAD incarnation's - // `created_revision`. A presence-only gate never refreshes it, and - // nothing else writes a non-owner's row. + // Compare the epoch, not just presence: a delete + recreate + // recycles the slab keys, so the row survives with the DEAD + // incarnation's `created_revision`. A presence-only gate never + // refreshes it, and nothing else writes a non-owner's row. + // + // No mirror of the staged-`InsertOwned` guard below, deliberately: + // a lagging pump costs one duplicate `InsertRouted` per pass, and + // the apply is an idempotent row overwrite, while scanning the op + // queue per routed namespace would go quadratic. if !shards_table_has_epoch(ctx, ns, epoch) { ctx.shard.enqueue_reconcile_op(ReconcileOp::InsertRouted { namespace: ns, @@ -614,6 +620,17 @@ async fn reconcile_additions( continue; } + // An earlier pass already built this one and the pump has not applied it + // yet, so the `contains` test above reads false for finished work. + // Rebuilding is not a wasted-effort question: the second build shares + // the namespace's `PartitionStats` with the queued sibling and re-opens + // segment 0 with `file_exists = false`, truncating the file that + // sibling is about to serve. + if ctx.shard.has_staged_insert_owned(ns) { + counters.already_staged += 1; + continue; + } + let now = Instant::now(); if ctx.is_backed_off(ns, FailureCause::Add, now) { counters.backoff_skipped += 1; @@ -648,7 +665,6 @@ async fn reconcile_additions( }); ctx.record_success(ns, FailureCause::Add); counters.materialised += 1; - staged.insert(ns); } Err(err) => { ctx.record_failure(ns, FailureCause::Add, now); @@ -664,8 +680,6 @@ async fn reconcile_additions( } } } - - staged } /// Retire parked frames the shard cannot serve, age the ones it might. @@ -705,18 +719,15 @@ async fn reconcile_additions( /// timeout and no committed op dies on a local-convergence signal. Residency /// only; see `ParkedFrame::passes`. /// -/// `staged_this_pass` is exempt: its `InsertOwned` is queued but not applied, so -/// it reads as un-materialised here. Not a one-pass concession. -/// `reconcile_additions` has no cross-pass guard against a queued-but-unapplied -/// op (it tests `partitions.contains`, false the whole time it sits in the -/// queue), so it re-stages every pass until the pump drains. The exemption -/// therefore covers arbitrary pump lag; dropping it ages frames on every -/// commit-driven pass the pump falls behind. -fn reconcile_parked_frames( - ctx: &ReconcilerCtx, - staged_this_pass: &AHashSet, - counters: &mut PassCounters, -) { +/// A namespace with a staged, unapplied `InsertOwned` is exempt: its partition +/// is on the way but reads as un-materialised here. The queue is asked per +/// parked namespace ([`shard::IggyShard::has_staged_insert_owned`]) rather than +/// carrying a set over from the additions pass, so the answer cannot go stale +/// across `reconcile_removals`' awaits; `parked` is empty on the steady path, +/// so the scan costs nothing there. The exemption spans passes, not just the +/// one that built the namespace, covering arbitrary pump lag; dropping it ages +/// frames on every commit-driven pass the pump falls behind. +fn reconcile_parked_frames(ctx: &ReconcilerCtx, counters: &mut PassCounters) { let parked = ctx.shard.parked_namespaces(); if parked.is_empty() { return; @@ -724,7 +735,7 @@ fn reconcile_parked_frames( let partitions = ctx.shard.plane.partitions(); let total_shards = u32::from(ctx.total_shards); for ns in parked { - if staged_this_pass.contains(&ns) { + if ctx.shard.has_staged_insert_owned(ns) { continue; } // Tombstoned namespaces are still in the map, so `contains` below reads @@ -1147,7 +1158,8 @@ pub fn install_tick_handler(shard: &Rc, wake_tx: WakeTx) { #[cfg(test)] mod tests { use super::{ - FailureCause, FailureRecord, ReconcilerCtx, delete_partitions_from_disk, reconcile_once, + FailureCause, FailureRecord, ReconcilerCtx, build_partition_fresh, + delete_partitions_from_disk, fetch_partition_stats, reconcile_once, }; use configs::server_ng::{NgSystemConfig, ServerNgConfig}; use consensus::{MetadataHandle, PartitionsHandle}; @@ -1181,6 +1193,7 @@ mod tests { use std::mem::size_of; use std::rc::Rc; use std::sync::Arc; + use std::sync::atomic::Ordering; use std::time::Instant; use tempfile::TempDir; @@ -1595,61 +1608,149 @@ mod tests { ); } - /// Regression (deferred-apply window): the reconciler stages - /// `ReconcileOp::InsertOwned` from a task separate from the pump that - /// applies it, so under a commit burst it can run a second pass before - /// the pump drains the first pass's staged ops. Both passes then - /// observe `!contains(ns)` and build the same namespace. The pump's - /// apply must be idempotent, else the second `insert` orphans the first - /// partition (leaked VSR group + writers) and inflates `len`. - /// `reconcile_pass` applies inline and cannot surface this, so here we - /// run two passes and only then drain once. + /// The cross-pass guard: a pass must not rebuild a namespace an earlier pass + /// already built and left queued. Rebuilding is not merely wasted work -- + /// the second build shares the namespace's `PartitionStats` with the queued + /// sibling and re-opens segment 0 truncating -- so a pass has to recognise + /// the staged op, not just `partitions.contains`. #[compio::test] - async fn deferred_apply_window_does_not_duplicate_owned_partition() { + async fn second_pass_does_not_rebuild_a_namespace_already_staged() { let tmp = TempDir::new().expect("tempdir for system path"); let config = test_config(&tmp); let mux = TestMux::default(); seed_stream(&mux, 1, "stream-a"); - seed_topic( - &mux, - 2, - 0, - "topic-a", - vec![assignment(0, 1), assignment(1, 2)], - ); + seed_topic(&mux, 2, 0, "topic-a", vec![assignment(0, 1)]); let shard = build_test_shard(0, &config, mux); let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); - // Two passes with no pump drain in between: models the reconciler, - // woken by a second commit tick, running pass N+1 before the pump - // applies pass N's `InsertOwned`. Both passes see the namespaces as - // unmaterialised and stage a build for each, so the queue holds two - // `InsertOwned` per namespace when the pump finally drains. reconcile_once(&ctx).await; + assert!( + ctx.shard.has_staged_insert_owned(ns), + "the first pass must leave an unapplied InsertOwned to guard against" + ); + + // Second pass while that op is still queued: `partitions.contains(ns)` + // is false, so only the staged-op guard can stop the rebuild. reconcile_once(&ctx).await; ctx.shard.apply_reconcile_ops(); + assert_eq!( + shard.plane.partitions().len(), + 1, + "the namespace must materialise exactly once" + ); - let partitions = shard.plane.partitions(); + // Counting builds, not partitions: the pump discards the redundant + // `InsertOwned` either way, so `len` cannot tell one build from two. + // `ensure_initial_segment` plants exactly one segment per build and + // folds it into the namespace's shared stats, so this counter is the + // observable that separates them. + let stats = fetch_partition_stats(&ctx, ns).expect("materialised namespace has stats"); assert_eq!( - partitions.len(), - 2, - "deferred-apply window must not duplicate partitions: \ - each namespace materialises exactly once" + stats.segments_count_inconsistent(), + 1, + "a second build ran: its initial segment was folded into the \ + namespace's shared stats on top of the live incarnation's" + ); + } + + /// The stats registry keys on the namespace, not the incarnation, so + /// `current_offset` moves on adoption only: the pump seeds it from the + /// incarnation it inserts, and a build that never becomes addressable + /// leaves it alone. Seeding from the build instead zeroed it under the live + /// incarnation, after which the partition plane's admission check read an + /// empty offset space and answered every `store_consumer_offset` above 0 + /// with `InvalidOffset` (error 4100) until the next send re-seeded it. + /// + /// Both incarnations are built and staged by hand: the adopted one so its + /// counter is non-zero BEFORE insertion (a reconciler build always adopts + /// at 0, where the publish is indistinguishable from a no-op), and the + /// redundant one because `has_staged_insert_owned` now stops a pass from + /// producing it. + #[compio::test] + async fn discarded_build_leaves_live_partition_offset_intact() { + const COMMITTED_OFFSET: u64 = 3; + const LIVE_EPOCH: u64 = 1; + + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-a"); + seed_topic(&mux, 2, 0, "topic-a", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config.clone())); + let ns = IggyNamespace::new(0, 0, 0); + + let stats = fetch_partition_stats(&ctx, ns).expect("committed namespace has stats"); + let live = build_partition_fresh( + &config, + ns, + Arc::clone(&stats), + LIVE_EPOCH, + CLUSTER_ID, + 0, + 1, + Rc::clone(&ctx.shard.bus), + ) + .await + .expect("live build succeeds"); + // What a recovery leaves behind: an incarnation whose own counter is + // ahead of the zeroed shared stats until adoption publishes it. + live.offset.store(COMMITTED_OFFSET, Ordering::Release); + ctx.shard.enqueue_reconcile_op(ReconcileOp::InsertOwned { + namespace: ns, + partition: Box::new(live), + epoch: LIVE_EPOCH, + }); + ctx.shard.apply_reconcile_ops(); + + assert_eq!( + stats.current_offset(), + COMMITTED_OFFSET, + "adoption must publish the incarnation's offset into the shared stats" + ); + + let redundant = build_partition_fresh( + &config, + ns, + Arc::clone(&stats), + LIVE_EPOCH + 1, + CLUSTER_ID, + 0, + 1, + Rc::clone(&ctx.shard.bus), + ) + .await + .expect("redundant build succeeds over the live incarnation's path"); + ctx.shard.enqueue_reconcile_op(ReconcileOp::InsertOwned { + namespace: ns, + partition: Box::new(redundant), + epoch: LIVE_EPOCH + 1, + }); + ctx.shard.apply_reconcile_ops(); + + assert_eq!( + shard.plane.partitions().len(), + 1, + "the redundant build must be discarded, not adopted: a second insert \ + overwrites the ns -> idx entry and orphans the first partition, \ + leaking its VSR group and segment writers" + ); + // The epoch, not `shard_for`: on a single shard an adopt would write + // `ShardId::new(0)` too, but it would stamp the redundant op's epoch. + assert_eq!( + shard.shards_table().epoch_for(ns), + Some(LIVE_EPOCH), + "the discarded op must not rewrite the routing row" + ); + assert_eq!( + stats.current_offset(), + COMMITTED_OFFSET, + "a discarded build must not reset the live incarnation's current_offset" ); - for partition_id in 0..2 { - let ns = IggyNamespace::new(0, 0, partition_id); - assert!( - partitions.contains(&ns), - "namespace {ns:?} must be addressable exactly once" - ); - assert_eq!( - shard.shards_table().shard_for(ns), - Some(0), - "shards_table must point at the owning shard" - ); - } } /// Multi-shard scenario: only the partition whose hash maps to diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 28537d1c5f..9a0ca3fc99 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -1775,6 +1775,34 @@ where let _ = sender.try_send(ShardFrame::lifecycle(LifecycleFrame::ReconcileApply)); } + /// `true` when an `InsertOwned` for `namespace` is built and queued but not + /// yet applied. + /// + /// The reconciler's own "already handled" test is `IggyPartitions::contains`, + /// which only turns true once the pump applies, so without this a pass run + /// during that lag rebuilds a namespace an earlier pass already built. The + /// queue IS the record of that in-flight work, so asking it cannot drift + /// from reality the way a parallel set would: every op leaves the queue + /// through `apply_reconcile_ops`, which either inserts or discards. + /// + /// Deliberately blind to `epoch`. Matching it would let a delete + recreate + /// landing inside the lag build a second incarnation over the queued one's + /// on-disk path, which is the case this exists to prevent; the recreate is + /// not lost, it costs one pass. The queued (dead-epoch) op applies, and the + /// next pass reads the epoch mismatch off the routing row and takes the + /// stale-incarnation teardown into a clean rebuild. + pub fn has_staged_insert_owned(&self, namespace: IggyNamespace) -> bool { + self.reconcile_queue.borrow().iter().any(|op| { + matches!( + op, + ReconcileOp::InsertOwned { + namespace: staged_namespace, + .. + } if *staged_namespace == namespace + ) + }) + } + /// Stage a segment-cleaner pass for `namespace` on this shard's pump. The /// timer task resolves retention config off-pump and stamps `now`; the pump /// is the single writer of partition state, so the deletion runs there, @@ -1882,18 +1910,30 @@ where epoch, } => { // Idempotent apply, mirroring `ConfirmRemove` (idempotent - // via `remove`'s `None` early-return). The reconciler - // stages this from a task separate from the pump, so under - // a commit burst two passes can each observe - // `!contains(ns)` and build the same namespace before - // either drains here. A second unconditional `insert` - // would push a duplicate partition and overwrite the - // `ns -> idx` entry, orphaning the first (its VSR group + - // segment writers leak and `len` inflates). The discarded - // build is a fresh empty incarnation over the same on-disk - // path the kept one owns, so dropping it just closes a few - // fds. + // via `remove`'s `None` early-return). An unconditional + // `insert` over a live namespace would push a duplicate + // partition and overwrite the `ns -> idx` entry, orphaning + // the first: its VSR group + segment writers leak and `len` + // inflates. + // + // A backstop, not the mechanism. `reconcile_additions` + // skips a namespace whose `InsertOwned` is already staged + // ([`Self::has_staged_insert_owned`]), so a second op for a + // live namespace should not be built at all. Dropping one + // here is damage control rather than a free no-op: the + // build already planted its initial segment over the live + // incarnation's path and folded that into the namespace's + // shared stats. if partitions.contains(&namespace) { + tracing::error!( + shard = self_shard_id, + ns_raw = namespace.inner(), + epoch, + "discarding duplicate InsertOwned for a live namespace: the \ + staged-op guard was bypassed and the build re-planted segment 0 \ + over the live incarnation's path" + ); + self.metrics.record_duplicate_partition_build_discarded(); drop(partition); continue; } diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 659550217e..2df15c357e 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -186,6 +186,7 @@ pub struct ShardMetrics { partitions_materialised_total: Counter, partitions_removed_total: Counter, partitions_reconcile_failures_total: Counter, + partitions_duplicate_builds_discarded_total: Counter, partition_transfer_refusals_total: Counter, partition_frames_rejected_stale_total: Counter, partition_frames_rejected_ahead_total: Counter, @@ -220,6 +221,7 @@ impl ShardMetrics { partitions_materialised_total: Counter::default(), partitions_removed_total: Counter::default(), partitions_reconcile_failures_total: Counter::default(), + partitions_duplicate_builds_discarded_total: Counter::default(), partition_transfer_refusals_total: Counter::default(), partition_frames_rejected_stale_total: Counter::default(), partition_frames_rejected_ahead_total: Counter::default(), @@ -259,6 +261,16 @@ impl ShardMetrics { self.partitions_removed_total.inc(); } + /// Bumped when the pump discards a duplicate `InsertOwned` for a namespace + /// that is already live. The reconciler's staged-op guard should make this + /// unreachable, so a non-zero value is a caught correctness anomaly, not + /// routine churn: the discarded build re-planted segment 0 over the live + /// incarnation's path and folded its initial segment into the shared stats + /// before the pump caught it. + pub fn record_duplicate_partition_build_discarded(&self) { + self.partitions_duplicate_builds_discarded_total.inc(); + } + /// Bumped each time `build_partition_fresh` or /// `delete_partitions_from_disk` returns `Err`. The reconciler retries /// next tick, but a sustained climb surfaces a stuck partition (disk