hubcio commented on code in PR #3846:
URL: https://github.com/apache/iggy/pull/3846#discussion_r3741058186
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -1595,61 +1627,122 @@ 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 incarnation shares the namespace's `PartitionStats` with the
+ /// live one 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(),
+ 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 a
build
+ /// that never becomes addressable must leave those counters alone. Seeding
+ /// `current_offset` 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.
+ #[compio::test]
+ async fn discarded_build_leaves_live_partition_offset_intact() {
+ const COMMITTED_OFFSET: u64 = 3;
+
+ 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);
+
+ reconcile_pass(&ctx).await;
+ let stats = fetch_partition_stats(&ctx, ns).expect("materialised
namespace has stats");
+
+ // What a committed send leaves behind: `commit_partition_entry`
+ // advances the visible offset on this same shared `Arc`.
+ stats.set_current_offset(COMMITTED_OFFSET);
+
+ // Built and staged by hand rather than by a second `reconcile_once`:
+ // `has_staged_insert_owned` now stops a pass from getting here, and
the
+ // invariant under test is the one that has to hold anyway -- a build
+ // that does not become the addressable incarnation leaves the
+ // namespace's counters untouched.
+ let redundant = build_partition_fresh(
+ &config,
+ ns,
+ Arc::clone(&stats),
2,
- "deferred-apply window must not duplicate partitions: \
- each namespace materialises exactly once"
+ 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: 2,
+ });
+ 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"
+ );
+ assert_eq!(
+ shard.shards_table().shard_for(ns),
Review Comment:
can't fail - single shard, adopt would write `ShardId::new(0)` too. capture
`epoch_before` after the first pass, stage the op with `epoch_before + 1`,
assert `epoch_for(ns) == epoch_before`.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -599,18 +606,20 @@ 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.
- if !shards_table_has_epoch(ctx, ns, epoch) {
- ctx.shard.enqueue_reconcile_op(ReconcileOp::InsertRouted {
- namespace: ns,
- owner: ShardId::new(owning_shard),
- epoch,
- });
- counters.routed += 1;
- }
+ stage_routing_row(ctx, ns, epoch, owning_shard, counters);
+ 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 incarnation
+ // shares the namespace's `PartitionStats` with the live one and
re-opens
+ // segment 0 with `file_exists = false`, which truncates. It joins
+ // `staged` so the parked-frame sweep still reads the namespace as
+ // building instead of aging its frames.
+ if ctx.shard.has_staged_insert_owned(ns) {
+ counters.already_staged += 1;
+ staged.insert(ns);
Review Comment:
`staged` duplicates what `has_staged_insert_owned` answers and goes stale
across the `reconcile_removals` awaits. drop the set + `staged_unapplied` param
and query the queue per parked namespace; `parked` early-returns when empty.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -1053,6 +1059,31 @@ fn fetch_partition_stats(
})
}
+/// Point this shard's routing row for `ns` at its owner, so a frame that lands
+/// here reaches the shard holding the partition.
+///
+/// Compares 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.
+fn stage_routing_row(
Review Comment:
single caller, wrapping `shards_table_has_epoch` (also single caller), while
593 spells the same predicate inline. inline this, keep
`shards_table_has_epoch` and use it at 593.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -1595,61 +1627,122 @@ 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 incarnation shares the namespace's `PartitionStats` with the
+ /// live one 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(),
+ 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 a
build
+ /// that never becomes addressable must leave those counters alone. Seeding
+ /// `current_offset` 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.
+ #[compio::test]
+ async fn discarded_build_leaves_live_partition_offset_intact() {
+ const COMMITTED_OFFSET: u64 = 3;
+
+ 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);
+
+ reconcile_pass(&ctx).await;
+ let stats = fetch_partition_stats(&ctx, ns).expect("materialised
namespace has stats");
+
+ // What a committed send leaves behind: `commit_partition_entry`
+ // advances the visible offset on this same shared `Arc`.
+ stats.set_current_offset(COMMITTED_OFFSET);
+
+ // Built and staged by hand rather than by a second `reconcile_once`:
+ // `has_staged_insert_owned` now stops a pass from getting here, and
the
+ // invariant under test is the one that has to hold anyway -- a build
+ // that does not become the addressable incarnation leaves the
+ // namespace's counters untouched.
+ let redundant = build_partition_fresh(
+ &config,
+ ns,
+ Arc::clone(&stats),
2,
- "deferred-apply window must not duplicate partitions: \
- each namespace materialises exactly once"
+ 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: 2,
+ });
+ 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"
+ );
+ assert_eq!(
+ shard.shards_table().shard_for(ns),
+ Some(0),
+ "the discarded op must not repoint the routing row"
+ );
+ assert_eq!(
Review Comment:
stays green even with all three `publish_current_offset()` calls deleted -
every build here adopts at offset 0 and the hand `set_current_offset(3)` lands
after the pass. nothing else covers the publish half (vsr-off CI spawns the
legacy binary). fix: `partition.offset.store(N, Ordering::Release)` before
enqueuing `InsertOwned`, then assert `stats.current_offset() == N` - must fail
when the publish is removed.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -599,18 +606,20 @@ 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.
- if !shards_table_has_epoch(ctx, ns, epoch) {
- ctx.shard.enqueue_reconcile_op(ReconcileOp::InsertRouted {
- namespace: ns,
- owner: ShardId::new(owning_shard),
- epoch,
- });
- counters.routed += 1;
- }
+ stage_routing_row(ctx, ns, epoch, owning_shard, counters);
+ 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 incarnation
+ // shares the namespace's `PartitionStats` with the live one and
re-opens
Review Comment:
nothing is live here (`contains` was false) - the sibling is queued for
adoption. same wording in the test doc at 1632; the copy at shard/lib.rs:1922
is fine.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -770,7 +770,23 @@ 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.
+ ///
+ /// Call it 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 fn publish_current_offset(&self) {
Review Comment:
all three call sites funnel through `IggyPartitions::insert` - moving the
publish inside makes the contract unforgettable, guards the unit-unreachable
boot site through the same test, and lets this go crate-private.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -1053,6 +1059,31 @@ fn fetch_partition_stats(
})
}
+/// Point this shard's routing row for `ns` at its owner, so a frame that lands
+/// here reaches the shard holding the partition.
+///
+/// Compares 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.
+fn stage_routing_row(
+ ctx: &ReconcilerCtx,
+ ns: IggyNamespace,
+ epoch: u64,
+ owning_shard: u16,
+ counters: &mut PassCounters,
+) {
+ if shards_table_has_epoch(ctx, ns, epoch) {
Review Comment:
no mirror of the `InsertOwned` guard, so a lagging pump gets one duplicate
`InsertRouted` per namespace per pass and the fast-skip stays disarmed.
harmless - just worth a doc line that the asymmetry is deliberate (a queue scan
here would go quadratic).
##########
core/shard/src/lib.rs:
##########
@@ -1882,21 +1910,26 @@ 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) {
Review Comment:
this drop is silent, but per the comment above it only fires after the build
truncated the live writer's file - worth `tracing::error!` + a metric.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
