krishvishal commented on code in PR #3786: URL: https://github.com/apache/iggy/pull/3786#discussion_r3706451025
########## core/integration/tests/cluster/multi_shard_partition_convergence.rs: ########## @@ -0,0 +1,201 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Partition convergence across shards on one node. +//! +//! `create_topic` returns on metadata commit, before the owning shard has run +//! `build_partition_fresh`, so a produce issued immediately after races +//! materialisation. Nothing about that race is resolved on the shard the client +//! is connected to: `shards_table` is a cache of the deterministic hash +//! assignment and may hold a row before the partition exists. The owning shard +//! resolves it, by parking the frame until its partition lands +//! (`park_if_unmaterialised`) and by fencing a mismatched incarnation +//! (`serves_committed_incarnation`), answering anything it cannot yet serve +//! with a retriable status the SDK replays. +//! +//! Forcing two shards is the point. Existing coverage of these fences is +//! single-shard, where the shard that admits the request is also the one that +//! owns the partition, so the cross-core path never runs: the hash fallback in +//! `router::route_typed`, the frame crossing into a peer's inbox, and that +//! peer's park queue draining on its own pump. Roughly half of the topics below +//! hash to a shard other than the one homing the connection. +//! +//! That split is a murmur3 outcome and is invisible from here -- this test would +//! stay green while silently degrading to single-shard if the hash or the shard +//! count changed. It is pinned instead where the assignment is a pure function, +//! by `partition_reconciler::tests::integration_topic_set_straddles_both_shards`, +//! which asserts over the same namespaces this test creates. Change the topic +//! count or the stream here and that guard has to move with it. +//! +//! Scope, stated plainly: this is a convergence test, not a barrier test. It +//! cannot tell a request served straight through from one that parked and was +//! re-dispatched, so it does not pin *which* mechanism carried it. What it does +//! catch is that mechanism failing outright -- a frame that never reaches the +//! owner, a park queue that never drains, or a fence that denies forever -- +//! since every one of those surfaces as a failed send or a short poll. + +#![cfg(feature = "vsr")] + +use iggy::prelude::*; +use integration::iggy_harness; + +const STREAM: &str = "convergence-stream"; +const PARTITION_ID: u32 = 0; +/// Enough topics that the murmur3 assignment lands on both shards; every one is +/// asserted, so which side each falls on does not matter. +const TOPICS: u32 = 8; + +fn topic_name(index: u32) -> String { + format!("convergence-topic-{index}") +} + +async fn create_topic(client: &IggyClient, stream: &Identifier, name: &str) { + client + .create_topic( + stream, + name, + 1, + CompressionAlgorithm::default(), + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .unwrap_or_else(|error| panic!("create_topic {name}: {error}")); +} + +async fn produce(client: &IggyClient, stream: &Identifier, topic: &Identifier, payload: &str) { + let mut messages = vec![ + IggyMessage::builder() + .payload(payload.to_owned().into()) + .build() + .expect("message build"), + ]; + client + .send_messages( + stream, + topic, + &Partitioning::partition_id(PARTITION_ID), + &mut messages, + ) + .await + .unwrap_or_else(|error| panic!("send_messages {payload}: {error}")); +} + +async fn poll_payloads( + client: &IggyClient, + stream: &Identifier, + topic: &Identifier, +) -> Vec<String> { + client + .poll_messages( + stream, + topic, + Some(PARTITION_ID), + &Consumer::default(), + &PollingStrategy::offset(0), + 16, + false, + ) + .await + .unwrap_or_else(|error| panic!("poll_messages: {error}")) + .messages + .iter() + .map(|message| String::from_utf8_lossy(&message.payload).into_owned()) + .collect() +} + +/// Topics are created in a batch first, so several materialisations are in +/// flight at once when the produces start. +#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation = "2"))] +async fn given_two_shards_when_producing_right_after_create_topic_should_round_trip( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + let stream = Identifier::named(STREAM).unwrap(); + client.create_stream(STREAM).await.unwrap(); + + for index in 0..TOPICS { + create_topic(&client, &stream, &topic_name(index)).await; + } + for index in 0..TOPICS { + let topic = Identifier::named(&topic_name(index)).unwrap(); + produce(&client, &stream, &topic, &format!("payload-{index}")).await; + } + for index in 0..TOPICS { + let topic = Identifier::named(&topic_name(index)).unwrap(); + assert_eq!( + poll_payloads(&client, &stream, &topic).await, + vec![format!("payload-{index}")], + "topic {index} must return the message produced right after its creation" + ); + } +} + +/// Delete + recreate reuses the freed slab keys, so the namespace is +/// byte-identical across incarnations and only `created_revision` separates +/// them. This pins the observable outcome across that transition on a +/// multi-shard node: the recreated topic serves its own data and none of the +/// dead incarnation's. +/// +/// It does NOT exercise the incarnation fence. Each step here is a completed +/// round trip, so the reconciler converges before the next one starts and +/// `serves_committed_incarnation` never denies (verified: zero "unverified Review Comment: Agreed, that parenthetical could not have verified anything. Removed the claim; the doc now states the sequencing argument and says outright that the deny logs at `debug` while the harness runs at `info`, so its absence proves nothing. Added `assert_no_degraded_park_paths`, counting the three `warn` markers via `stdout_occurrences` at the end of both tests. Both pass at zero, so the park path is now positively pinned. ########## core/server-ng/src/partition_reconciler.rs: ########## @@ -2104,4 +2423,918 @@ mod tests { "survivor must take over the disconnected member's partitions" ); } + + /// A namespace deleted before its build finished is named by nothing: it is + /// absent from `IggyPartitions`, so the removals pass sees no owned ghost, + /// and absent from `shards_table`, since the owner seeds a row only via + /// `InsertOwned`. Neither `ConfirmRemove` nor `RemoveRouted` can therefore + /// reach its parked frames, and without the sweep they are held for the + /// process lifetime while every waiting client burns its read timeout. + /// + /// Reclaim is via the age bound, not on sight of the namespace leaving the + /// target set: "absent from committed metadata" reads identically for a + /// deleted namespace and for one a metadata-lagging replica has not applied + /// yet, so reclaiming on that would destroy live in-flight traffic. The first + /// pass must therefore hold the frames, and a few passes later they are gone. + #[compio::test] + async fn parked_frames_are_reclaimed_when_the_namespace_leaves_metadata() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-reclaim"); + seed_topic(&mux, 2, 0, "topic-reclaim", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + // No pass yet, so the namespace is committed but unmaterialised. + park_one_request(&shard, ns).await; + assert_eq!( + shard.parked_namespaces(), + vec![ns], + "a request for an unmaterialised namespace must park" + ); + + // Delete before any pass builds it: the reconciler never materialises + // it, so nothing drains the entry the normal way. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + reconcile_pass(&ctx).await; + assert_eq!( + shard.parked_frame_count(ns), + 1, + "the first pass must not destroy the frame: absence from the target set \ + is also what a not-yet-applied commit looks like" + ); + + // Every subsequent pass ages it, and the park buffer keeps defeating the + // revision fast-skip until it drains. + for _ in 0..=PARK_MAX_PASSES { + reconcile_pass(&ctx).await; + } + + assert!( + shard.parked_namespaces().is_empty(), + "frames for a namespace that left metadata must be answered and reclaimed" + ); + assert_eq!( + drain_staged_client_sends(&inbox), + 1, + "and the waiting client must get a retriable answer" + ); + } + + /// A frame parked before this node's metadata knew the namespace carries no + /// epoch stamp, and `None` must NOT read as "prior incarnation": on a + /// metadata-lagging replica it is the ordinary case, since the partition + /// primary materialises and replicates as soon as its own metadata commits. + /// Rejecting it destroys live traffic -- silently for a replicated prepare, + /// which has no client to answer -- and the pre-stamp code served it. + #[compio::test] + async fn unstamped_parked_frame_is_served_not_rejected_as_stale() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-unstamped"); + seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Topic slab 1 does not exist yet, so there is no committed + // `created_revision` to stamp: the frame parks with `epoch: None`. + let unknown = IggyNamespace::new(0, 1, 0); + park_one_request(&shard, unknown).await; + assert_eq!( + shard.parked_frame_count(unknown), + 1, + "a request for a namespace this node has not applied must park" + ); + + // The commit this node was lagging behind now lands, and the pass + // materialises the namespace. + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + assert_eq!( + shard.parked_frame_count(unknown), + 0, + "materialisation must drain the park entry" + ); + let (served, answered) = drain_inbox(&inbox); + assert_eq!( + served, 1, + "the unstamped frame must be re-dispatched onto the pump, not rejected" + ); + assert_eq!( + answered, 0, + "and it must not be answered with a deny instead of served" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 0, + "an absent stamp is not evidence of a prior incarnation" + ); + } + + /// The shard-wide byte budget is a running total maintained at each mutation + /// site rather than rescanned per arriving frame. A total that fails to debit + /// on drain silently wedges the budget: the shard would shed every namespace's + /// frames while nothing is actually parked. Exercise each way frames leave -- + /// re-dispatch on materialisation, the age bound, and reclaim -- and assert the + /// total returns to empty. + #[compio::test] + async fn park_byte_total_returns_to_zero_on_every_drain_path() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-bytes"); + seed_topic(&mux, 2, 0, "topic-bytes", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 16); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Drain path 1: materialisation re-dispatches. + let late = IggyNamespace::new(0, 1, 0); + park_one_request(&shard, late).await; + assert!(shard.has_parked_partition_frames()); + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-bytes-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + assert!( + !shard.has_parked_partition_frames(), + "re-dispatch must debit the parked-byte total" + ); + + // Drain path 2: the age bound answers the frame. + let never = IggyNamespace::new(0, 9, 0); + park_one_request(&shard, never).await; + assert!(shard.has_parked_partition_frames()); + for _ in 0..=PARK_MAX_PASSES { + shard.age_parked_partition_frames(never); + } + assert!( + !shard.has_parked_partition_frames(), + "aging out must debit the parked-byte total" + ); + + // Drain path 3: an explicit reclaim. + park_one_request(&shard, never).await; + assert!(shard.has_parked_partition_frames()); + shard.reclaim_parked_partition_frames(never); + assert!( + !shard.has_parked_partition_frames(), + "reclaim must debit the parked-byte total" + ); + + drop(inbox); + } + + /// The replicated-prepare shape, which no other test covers and where both + /// park critical are worst: a prepare has no client, so `deny_parked_frame` + /// no-ops on it and anything that discards it loses committed data silently, + /// with no normal-status repair driver to refetch it. + /// + /// A backup receives the prepare before its own metadata commits (so the frame + /// parks unstamped), then applies the commit and materialises. The prepare must + /// be re-dispatched, not rejected as a prior incarnation. + #[compio::test] + async fn unstamped_parked_prepare_is_served_after_materialisation() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-prepare"); + seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // The primary replicates ahead of this node's metadata: topic slab 1 is + // not committed here yet, so the prepare parks with `epoch: None`. + let lagging = IggyNamespace::new(0, 1, 0); + park_one_prepare(&shard, lagging, 7).await; + assert_eq!( + shard.parked_frame_count(lagging), + 1, + "a prepare for a namespace this backup has not applied must park" + ); + + // The metadata commit catches up and the pass materialises the namespace. + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + let (served, answered) = drain_inbox(&inbox); + assert_eq!( + served, 1, + "the parked prepare must be re-dispatched; discarding it is silent \ + committed-data loss, since a prepare has no client to answer" + ); + assert_eq!(answered, 0, "a prepare has no client deny to send"); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 0, + "an unstamped prepare is not a prior incarnation" + ); + } + + /// A parked prepare whose stamp names a DIFFERENT incarnation must still be + /// dropped: applying a dead topic's op into the topic that recycled its slab + /// keys diverges this replica. This is the half of the fence that stays. + #[compio::test] + async fn stamped_parked_prepare_from_a_prior_incarnation_is_rejected() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-stale"); + seed_topic(&mux, 2, 0, "topic-first", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + // Parked against the FIRST incarnation, so it carries that revision. + park_one_prepare(&shard, ns, 7).await; + assert_eq!(shard.parked_frame_count(ns), 1); + + // Delete and recreate: same namespace keys, new committed revision. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-recreated", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + let (served, _answered) = drain_inbox(&inbox); + assert_eq!( + served, 0, + "a prepare stamped with the dead incarnation must not be served against \ + its replacement" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 1, + "and the reject must be counted" + ); + } + + /// Parking does not bump `Streams::revision` and does not wake the reconciler, + /// so a frame that parks in a converged steady state would be held for the + /// process lifetime if the revision fast-skip could still fire. A non-empty + /// park buffer must therefore defeat the skip. + #[compio::test] + async fn park_buffer_defeats_the_revision_fast_skip() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-skip"); + seed_topic(&mux, 2, 0, "topic-skip", vec![assignment(0, 1)]); + + let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Converge: the first pass materialises, the verify pass after it arms + // the skip (same sequence as `reconcile_fast_skips_when_revision_unchanged`). + assert!(reconcile_once(&ctx).await, "first pass runs the full diff"); + ctx.shard.apply_reconcile_ops(); + assert!( + reconcile_once(&ctx).await, + "the verify pass after a working pass must still run" + ); + ctx.shard.apply_reconcile_ops(); + assert!( + !reconcile_once(&ctx).await, + "a converged pass with an unchanged revision must fast-skip" + ); + + // Park a frame for a namespace that is NOT materialised, without touching + // the revision, and the skip must stop firing. + let unbuilt = IggyNamespace::new(0, 0, 7); + park_one_request(&shard, unbuilt).await; + assert!( + shard.has_parked_partition_frames(), + "the frame must be parked for this test to mean anything" + ); + assert!( + reconcile_once(&ctx).await, + "a non-empty park buffer must defeat the fast-skip so the sweep can run" + ); + } + + /// Same hole, reached the other way: the namespace stays committed but + /// `build_partition_fresh` keeps failing. The `FailureCause::Add` backoff + /// clamps at 60s, twice the client's read timeout, so holding the frames + /// cannot help - answer them and let the client re-issue. + #[compio::test] + async fn parked_frames_are_reclaimed_while_the_build_is_backed_off() { Review Comment: Rewritten as `a_backed_off_build_ages_requests_and_retains_prepares`: `build_test_shard_with_inbox` + `drain_staged_client_sends` so the answer is observed rather than assumed, both frame classes parked, and it asserts that one failed build destroys nothing. The behaviour it used to pin is gone with the backoff-reclaim fix. ########## core/server-ng/src/partition_reconciler.rs: ########## @@ -2104,4 +2423,918 @@ mod tests { "survivor must take over the disconnected member's partitions" ); } + + /// A namespace deleted before its build finished is named by nothing: it is + /// absent from `IggyPartitions`, so the removals pass sees no owned ghost, + /// and absent from `shards_table`, since the owner seeds a row only via + /// `InsertOwned`. Neither `ConfirmRemove` nor `RemoveRouted` can therefore + /// reach its parked frames, and without the sweep they are held for the + /// process lifetime while every waiting client burns its read timeout. + /// + /// Reclaim is via the age bound, not on sight of the namespace leaving the + /// target set: "absent from committed metadata" reads identically for a + /// deleted namespace and for one a metadata-lagging replica has not applied + /// yet, so reclaiming on that would destroy live in-flight traffic. The first + /// pass must therefore hold the frames, and a few passes later they are gone. + #[compio::test] + async fn parked_frames_are_reclaimed_when_the_namespace_leaves_metadata() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-reclaim"); + seed_topic(&mux, 2, 0, "topic-reclaim", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + // No pass yet, so the namespace is committed but unmaterialised. + park_one_request(&shard, ns).await; + assert_eq!( + shard.parked_namespaces(), + vec![ns], + "a request for an unmaterialised namespace must park" + ); + + // Delete before any pass builds it: the reconciler never materialises + // it, so nothing drains the entry the normal way. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + reconcile_pass(&ctx).await; + assert_eq!( + shard.parked_frame_count(ns), + 1, + "the first pass must not destroy the frame: absence from the target set \ + is also what a not-yet-applied commit looks like" + ); + + // Every subsequent pass ages it, and the park buffer keeps defeating the + // revision fast-skip until it drains. + for _ in 0..=PARK_MAX_PASSES { + reconcile_pass(&ctx).await; + } + + assert!( + shard.parked_namespaces().is_empty(), + "frames for a namespace that left metadata must be answered and reclaimed" + ); + assert_eq!( + drain_staged_client_sends(&inbox), + 1, + "and the waiting client must get a retriable answer" + ); + } + + /// A frame parked before this node's metadata knew the namespace carries no + /// epoch stamp, and `None` must NOT read as "prior incarnation": on a + /// metadata-lagging replica it is the ordinary case, since the partition + /// primary materialises and replicates as soon as its own metadata commits. + /// Rejecting it destroys live traffic -- silently for a replicated prepare, + /// which has no client to answer -- and the pre-stamp code served it. + #[compio::test] + async fn unstamped_parked_frame_is_served_not_rejected_as_stale() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-unstamped"); + seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Topic slab 1 does not exist yet, so there is no committed + // `created_revision` to stamp: the frame parks with `epoch: None`. + let unknown = IggyNamespace::new(0, 1, 0); + park_one_request(&shard, unknown).await; + assert_eq!( + shard.parked_frame_count(unknown), + 1, + "a request for a namespace this node has not applied must park" + ); + + // The commit this node was lagging behind now lands, and the pass + // materialises the namespace. + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + assert_eq!( + shard.parked_frame_count(unknown), + 0, + "materialisation must drain the park entry" + ); + let (served, answered) = drain_inbox(&inbox); + assert_eq!( + served, 1, + "the unstamped frame must be re-dispatched onto the pump, not rejected" + ); + assert_eq!( + answered, 0, + "and it must not be answered with a deny instead of served" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 0, + "an absent stamp is not evidence of a prior incarnation" + ); + } + + /// The shard-wide byte budget is a running total maintained at each mutation + /// site rather than rescanned per arriving frame. A total that fails to debit + /// on drain silently wedges the budget: the shard would shed every namespace's + /// frames while nothing is actually parked. Exercise each way frames leave -- + /// re-dispatch on materialisation, the age bound, and reclaim -- and assert the + /// total returns to empty. + #[compio::test] + async fn park_byte_total_returns_to_zero_on_every_drain_path() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-bytes"); + seed_topic(&mux, 2, 0, "topic-bytes", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 16); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Drain path 1: materialisation re-dispatches. + let late = IggyNamespace::new(0, 1, 0); + park_one_request(&shard, late).await; + assert!(shard.has_parked_partition_frames()); + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-bytes-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + assert!( + !shard.has_parked_partition_frames(), + "re-dispatch must debit the parked-byte total" + ); + + // Drain path 2: the age bound answers the frame. + let never = IggyNamespace::new(0, 9, 0); + park_one_request(&shard, never).await; + assert!(shard.has_parked_partition_frames()); + for _ in 0..=PARK_MAX_PASSES { + shard.age_parked_partition_frames(never); + } + assert!( + !shard.has_parked_partition_frames(), + "aging out must debit the parked-byte total" + ); + + // Drain path 3: an explicit reclaim. + park_one_request(&shard, never).await; + assert!(shard.has_parked_partition_frames()); + shard.reclaim_parked_partition_frames(never); + assert!( + !shard.has_parked_partition_frames(), + "reclaim must debit the parked-byte total" + ); + + drop(inbox); + } + + /// The replicated-prepare shape, which no other test covers and where both + /// park critical are worst: a prepare has no client, so `deny_parked_frame` + /// no-ops on it and anything that discards it loses committed data silently, + /// with no normal-status repair driver to refetch it. + /// + /// A backup receives the prepare before its own metadata commits (so the frame + /// parks unstamped), then applies the commit and materialises. The prepare must + /// be re-dispatched, not rejected as a prior incarnation. + #[compio::test] + async fn unstamped_parked_prepare_is_served_after_materialisation() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-prepare"); + seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // The primary replicates ahead of this node's metadata: topic slab 1 is + // not committed here yet, so the prepare parks with `epoch: None`. + let lagging = IggyNamespace::new(0, 1, 0); + park_one_prepare(&shard, lagging, 7).await; + assert_eq!( + shard.parked_frame_count(lagging), + 1, + "a prepare for a namespace this backup has not applied must park" + ); + + // The metadata commit catches up and the pass materialises the namespace. + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "topic-late", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + let (served, answered) = drain_inbox(&inbox); + assert_eq!( + served, 1, + "the parked prepare must be re-dispatched; discarding it is silent \ + committed-data loss, since a prepare has no client to answer" + ); + assert_eq!(answered, 0, "a prepare has no client deny to send"); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 0, + "an unstamped prepare is not a prior incarnation" + ); + } + + /// A parked prepare whose stamp names a DIFFERENT incarnation must still be + /// dropped: applying a dead topic's op into the topic that recycled its slab + /// keys diverges this replica. This is the half of the fence that stays. + #[compio::test] + async fn stamped_parked_prepare_from_a_prior_incarnation_is_rejected() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-stale"); + seed_topic(&mux, 2, 0, "topic-first", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + // Parked against the FIRST incarnation, so it carries that revision. + park_one_prepare(&shard, ns, 7).await; + assert_eq!(shard.parked_frame_count(ns), 1); + + // Delete and recreate: same namespace keys, new committed revision. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-recreated", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + + let (served, _answered) = drain_inbox(&inbox); + assert_eq!( + served, 0, + "a prepare stamped with the dead incarnation must not be served against \ + its replacement" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 1, + "and the reject must be counted" + ); + } + + /// Parking does not bump `Streams::revision` and does not wake the reconciler, + /// so a frame that parks in a converged steady state would be held for the + /// process lifetime if the revision fast-skip could still fire. A non-empty + /// park buffer must therefore defeat the skip. + #[compio::test] + async fn park_buffer_defeats_the_revision_fast_skip() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-skip"); + seed_topic(&mux, 2, 0, "topic-skip", vec![assignment(0, 1)]); + + let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + + // Converge: the first pass materialises, the verify pass after it arms + // the skip (same sequence as `reconcile_fast_skips_when_revision_unchanged`). + assert!(reconcile_once(&ctx).await, "first pass runs the full diff"); + ctx.shard.apply_reconcile_ops(); + assert!( + reconcile_once(&ctx).await, + "the verify pass after a working pass must still run" + ); + ctx.shard.apply_reconcile_ops(); + assert!( + !reconcile_once(&ctx).await, + "a converged pass with an unchanged revision must fast-skip" + ); + + // Park a frame for a namespace that is NOT materialised, without touching + // the revision, and the skip must stop firing. + let unbuilt = IggyNamespace::new(0, 0, 7); + park_one_request(&shard, unbuilt).await; + assert!( + shard.has_parked_partition_frames(), + "the frame must be parked for this test to mean anything" + ); + assert!( + reconcile_once(&ctx).await, + "a non-empty park buffer must defeat the fast-skip so the sweep can run" + ); + } + + /// Same hole, reached the other way: the namespace stays committed but + /// `build_partition_fresh` keeps failing. The `FailureCause::Add` backoff + /// clamps at 60s, twice the client's read timeout, so holding the frames + /// cannot help - answer them and let the client re-issue. + #[compio::test] + async fn parked_frames_are_reclaimed_while_the_build_is_backed_off() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-backoff"); + seed_topic(&mux, 2, 0, "topic-backoff", 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); + + park_one_request(&shard, ns).await; + assert_eq!(shard.parked_namespaces(), vec![ns]); + + // Stand in for a failed build (ENOSPC / EPERM): the additions pass skips + // a backed-off namespace, so it stays committed and unmaterialised. + ctx.record_failure(ns, FailureCause::Add, Instant::now()); + reconcile_pass(&ctx).await; + + assert!( + !ctx.shard.plane.partitions().contains(&ns), + "a backed-off namespace must not have been built" + ); + assert!( + shard.parked_namespaces().is_empty(), + "frames waiting on a backed-off build must be answered, not held" + ); + } + + /// Delete + recreate recycles the slab keys, so a frame parked against the + /// dead incarnation is byte-identical to one for its replacement. Draining + /// it into the new partition would land a dead topic's write inside the live + /// one, and the incarnation fence cannot catch it: that compares the + /// committed revision against the routing row, both of which describe the + /// NEW incarnation. Only the epoch stamped at park time separates them. + #[compio::test] + async fn parked_frames_from_a_prior_incarnation_are_not_served_by_its_replacement() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-epoch"); + seed_topic(&mux, 2, 0, "topic-epoch", 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); + + // Parked against the first incarnation, before any pass builds it. + park_one_request(&shard, ns).await; + assert_eq!(shard.parked_namespaces(), vec![ns]); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 0, + "nothing rejected yet" + ); + + // Recreate the same tuple. The namespace is unchanged; only + // `created_revision` moves. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-epoch", + vec![assignment(0, 1)], + ); + + // The pass builds the SECOND incarnation and drains the park entry. + reconcile_pass(&ctx).await; + + assert!( + shard.plane.partitions().contains(&ns), + "the recreated incarnation must materialise" + ); + assert!( + shard.parked_namespaces().is_empty(), + "the park entry must be drained by the materialisation" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 1, + "the frame stamped with the dead incarnation must be rejected, not \ + re-dispatched into its replacement" + ); + } + + /// Past the per-namespace cap the frame is gone either way, but a client + /// request must still be answered: the transports decode replies in + /// lockstep, so a silent shed leaves the connection waiting out its full + /// response read-timeout. + /// + /// Driven against a registered in-process client, so the assertion is that a + /// reply reached a waiter -- not that a counter moved. With no client on the + /// bus every send fails as `ClientNotFound` and a counter bumped before the + /// send reports an answer nobody received, which is the failure this test + /// exists to catch. + #[compio::test] + async fn park_overflow_answers_the_client_instead_of_shedding_silently() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-overflow"); + seed_topic(&mux, 2, 0, "topic-overflow", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ns = IggyNamespace::new(0, 0, 0); + let (_slot, reply_rx) = register_waiting_client(&shard); + + // Fill the buffer to its cap, then one more. + for _ in 0..PARK_CAP { + park_one_request(&shard, ns).await; + } + assert_eq!( + park_overflow_count(&shard), + 0, + "everything up to the cap parks without shedding" + ); + + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 0, + "nothing has been answered yet; the parked frames are still waiting" + ); + + park_one_request(&shard, ns).await; + assert_eq!( + park_overflow_count(&shard), + 1, + "the frame past the cap must be shed and counted, not parked" + ); + assert_eq!( + shard.parked_frame_count(ns), + PARK_CAP, + "the shed frame must not have grown the buffer past its cap" + ); + // The point of the fix: shedding is unavoidable at the cap, silence is + // not. Without the deny the connection waits out its whole response + // read-timeout on a frame that is already gone. + let reply = reply_rx + .await + .expect("the shed request must reach the waiting client"); + assert_eq!( + reply_status(&reply), + iggy_common::IggyError::TransientNotAccepted.as_code(), + "the shed request must be answered with a retriable status" + ); + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 1, + "and the counter must credit that delivered answer" + ); + } + + /// The counter must credit only denies the bus delivered. It previously + /// incremented before `send_to_client`, so a shard with no client registered + /// still reported the request answered - which blinded + /// `park_overflow_answers_the_client_instead_of_shedding_silently`, the one + /// test that asserts the client hears back. + #[compio::test] + async fn overflow_deny_is_not_counted_when_the_client_is_gone() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-overflow-gone"); + seed_topic(&mux, 2, 0, "topic-overflow-gone", vec![assignment(0, 1)]); + + // No client registered: every `send_to_client` fails `ClientNotFound`. + let shard = build_test_shard(0, &config, mux); + let ns = IggyNamespace::new(0, 0, 0); + + for _ in 0..=PARK_CAP { + park_one_request(&shard, ns).await; + } + + assert_eq!( + park_overflow_count(&shard), + 1, + "the frame past the cap is still shed and counted" + ); + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 0, + "a deny the bus could not deliver must not be counted as an answer" + ); + assert_eq!( + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::PARTITION, + shard::metrics::frame_drop_reason::DELIVERY_FAILED, + ), + 1, + "it must be counted as an undelivered reply instead" + ); + } + + /// A parked prepare that ages out has no client to answer, so nothing is + /// staged and `partition_requests_denied_transient_total` stays put. The op + /// is destroyed all the same - the primary retransmits only what has not + /// reached quorum - so `park_dropped` is the only record it existed. + #[compio::test] + async fn aged_out_prepare_is_counted_even_though_nobody_can_be_answered() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-prepare-age"); + seed_topic(&mux, 2, 0, "topic-prepare-age", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ns = IggyNamespace::new(0, 0, 0); + + park_one_prepare(&shard, ns, 7).await; + for _ in 0..=PARK_MAX_PASSES { + shard.age_parked_partition_frames(ns); + } + + assert_eq!(shard.parked_frame_count(ns), 0, "the prepare aged out"); + assert_eq!( + drain_staged_client_sends(&inbox), + 0, + "a prepare has no client to answer" + ); + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 0, + "and must not be reported as an answered request" + ); + assert_eq!( + park_dropped_count(&shard), + 1, + "the destroyed op must leave a record; silence here is invisible loss" + ); + } + + /// A frame larger than the per-namespace byte cap used to fail the check + /// even against an empty entry, so it could never park on any attempt. For a + /// replicated prepare that is unrecoverable: `retransmit_targets` skips an op + /// that already reached quorum, so the backup stays permanently short of it. + #[compio::test] + async fn a_frame_over_the_namespace_byte_cap_still_parks_into_an_empty_entry() { + const OVER_NAMESPACE_CAP: usize = 5 * 1024 * 1024; + + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-oversize"); + seed_topic(&mux, 2, 0, "topic-oversize", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ns = IggyNamespace::new(0, 0, 0); + + shard + .on_message(build_partition_request_sized(ns, OVER_NAMESPACE_CAP)) + .await; + assert_eq!( + shard.parked_frame_count(ns), + 1, + "the first frame of an empty entry must park regardless of the per-namespace cap" + ); + assert_eq!( + park_overflow_count(&shard), + 0, + "and must not be shed doing it" + ); + + // The waiver is for the first frame only; the cap still applies after. + shard + .on_message(build_partition_request_sized(ns, OVER_NAMESPACE_CAP)) + .await; + assert_eq!( + shard.parked_frame_count(ns), + 1, + "a second oversize frame must be shed, or one namespace eats the shard budget" + ); + assert_eq!(park_overflow_count(&shard), 1); + } + + /// A namespace whose build is still in flight keeps its frames -- but not + /// forever, or the park buffer grows with a namespace that never materialises. + /// The bound is in reconciler passes so the simulator's virtual clock governs + /// it. + /// + /// Driven through `age_parked_partition_frames` directly. The sweep calls it + /// once per pass for a namespace still building, and that branch is the only + /// way a committed, non-backed-off namespace reaches the bound - which a unit + /// test cannot stage, since its build completes on the first pass. + /// + /// Uses a shard with a live inbox: the deny is staged onto the pump, so a + /// shard with no sender would report the frame answered while nothing was + /// ever handed anywhere. + #[compio::test] + async fn parked_frames_are_answered_once_they_outlive_their_admission_window() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-age"); + seed_topic(&mux, 2, 0, "topic-age", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ns = IggyNamespace::new(0, 0, 0); + + park_one_request(&shard, ns).await; + assert_eq!(shard.parked_frame_count(ns), 1); + + // Each pass ages the frame; it survives until it is over the bound. + for pass in 0..PARK_MAX_PASSES { + assert_eq!( + shard.age_parked_partition_frames(ns), + 0, + "pass {pass} is still inside the admission window" + ); + assert_eq!(shard.parked_frame_count(ns), 1); + } + assert_eq!( + shard.age_parked_partition_frames(ns), + 1, + "the pass past the bound must answer the frame" + ); + assert_eq!(shard.parked_frame_count(ns), 0); + assert_eq!( + drain_staged_client_sends(&inbox), + 1, + "the answer must actually reach the pump, not just the counter" + ); + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 1, + "and it must be answered with a retriable status, not dropped" + ); + } + + /// The counter must credit only denies the pump accepted. It previously + /// incremented before the `try_send`, so a shard whose inbox refused the frame + /// (or had no sender at all) still reported the client answered. + #[compio::test] + async fn transient_deny_is_not_counted_when_the_inbox_cannot_take_it() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-deny-drop"); + seed_topic(&mux, 2, 0, "topic-deny-drop", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ns = IggyNamespace::new(0, 0, 0); + park_one_request(&shard, ns).await; + + // Kill the pump side, so every staged frame is refused. + drop(inbox); + + for _ in 0..=PARK_MAX_PASSES { + shard.age_parked_partition_frames(ns); + } + + assert_eq!(shard.parked_frame_count(ns), 0, "the frame still ages out"); + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 0, + "a deny the inbox refused must not be counted as an answer" + ); + } + + /// The frame cap bounds count, not residency: `Message::into_generic` is a + /// retag, so each parked entry keeps its whole buffer -- up to 64 MiB. With + /// only a frame cap, one namespace could pin 128 x 64 MiB and nothing capped + /// the namespace count. The shard-wide byte budget is what actually bounds + /// it, so large frames must shed well before the frame cap. + #[compio::test] + async fn park_byte_budget_sheds_large_frames_before_the_frame_cap() { Review Comment: Correct, that was the per-namespace cap. Renamed to `park_namespace_byte_budget_sheds_large_frames_before_the_frame_cap` and made the assertion exact (`NAMESPACE_BUDGET / MIB_FOOTPRINT` frames, then one shed). Added `park_shard_wide_byte_budget_sheds_a_namespace_that_would_cross_it` with the shape you described, so `MAX_PARKED_BYTES` is bound by a test for the first time. ########## core/shard/src/lib.rs: ########## @@ -1781,29 +1958,340 @@ where /// decode replies in lockstep, so silence wedges the connection until the /// SDK's response read-timeout. fn discard_parked_partition_frames(&self, namespace: IggyNamespace) { - if let Some(frames) = self - .pending_partition_frames - .borrow_mut() - .remove(&namespace) + // Bound the borrow to this statement: the guard in an `if let` + // scrutinee otherwise lives to the end of the then-block, holding a + // shard-global map locked across the outbound sends below. + let parked = self.take_parked_partition_frames(namespace); + if let Some(frames) = parked && !frames.is_empty() { + let total = frames.len(); + let mut answered = 0; + for frame in frames { + if self.deny_parked_client_request(frame) { + answered += 1; + } else { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::metrics::frame_drop_reason::PARK_DROPPED, + ); + } + } tracing::debug!( shard = self.id, namespace_raw = namespace.inner(), - count = frames.len(), + answered, + dropped = total - answered, "discarding parked partition frames for removed namespace" ); - for frame in frames { - if frame.header().command == Command2::Request - && let Ok(request) = frame.try_into_typed::<RequestHeader>() + } + } + + /// Remove a namespace's park entry, debiting its bytes from + /// [`Self::parked_partition_bytes`]. The single place entries leave the map, + /// so the running total and the pending-retry set cannot drift out of step + /// with it. + fn take_parked_partition_frames(&self, namespace: IggyNamespace) -> Option<Vec<ParkedFrame>> { + self.reparked_partition_namespaces + .borrow_mut() + .remove(&namespace); + let frames = self + .pending_partition_frames + .borrow_mut() + .remove(&namespace)?; + let freed: usize = frames + .iter() + .map(|frame| parked_footprint(frame.message.as_slice().len())) + .sum(); + self.parked_partition_bytes + .set(self.parked_partition_bytes.get().saturating_sub(freed)); + Some(frames) + } + + /// Whether any frame is parked. Cheap enough for the reconciler's per-tick + /// fast-skip guard: a non-empty buffer means the shard is by definition not + /// converged, so the skip must not fire. + #[must_use] + pub const fn has_parked_partition_frames(&self) -> bool { + self.parked_partition_bytes.get() > 0 + } + + /// Namespaces currently holding parked frames. The reconciler pairs this + /// against committed metadata to find the ones that will never materialise, + /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that was + /// never built is in neither `IggyPartitions` nor the routing table, so + /// nothing else names it. + #[must_use] + pub fn parked_namespaces(&self) -> Vec<IggyNamespace> { + self.pending_partition_frames + .borrow() + .keys() + .copied() + .collect() + } + + /// Re-queue the frames parked for `namespace` now that its partition exists + /// at `epoch`, onto this shard's own inbox so the pump serves them after the + /// current drain. + /// + /// A frame stamped with a DIFFERENT incarnation never makes it back: the + /// namespace is byte-identical across incarnations, so serving it would land + /// a dead topic's write inside the topic that recycled its keys, and the + /// downstream fence cannot see it -- that compares the committed revision + /// against the routing row, both of which now describe THIS incarnation. + /// + /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's + /// metadata held no committed partition for the namespace when the frame + /// arrived, which on a metadata-lagging backup is the ordinary case the park + /// buffer exists to absorb -- the partition primary materialises and + /// replicates as soon as its own metadata commits, well before a lagging + /// backup applies the same commit. Treating that as "prior incarnation" + /// destroys live traffic: a replicated prepare has no client to answer, so + /// it would be dropped with no recovery until an unrelated view change. + /// The residual is unchanged from before the stamp existed -- a frame parked + /// while the namespace was absent, then recreated under a new incarnation, + /// is served against the replacement -- and closing it needs a wire-level + /// discriminator (see the `TODO(krishna)` in + /// `partition_reconciler`'s module docs), not a `None`-means-stale rule. + /// + /// A frame the inbox refuses is re-parked, not answered. Re-queuing appends, + /// so a pass that materialises many namespaces at once can overrun the inbox; + /// staging a deny there is futile because the deny rides that same sender + /// with no await in between, so nothing can have drained a slot. + /// + /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the reconciler + /// sweep ages a namespace only while it is un-materialised, and by here it + /// is materialised. [`Self::repark_partition_frames`] arms the pump-side + /// retry instead, and the sweep's backstop for an inbox that never drains is + /// `partition_reconciler::reconcile_parked_frames`, which now ages a + /// materialised namespace too. + fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64) + where + B: MessageBus + 'static, + { + let Some(frames) = self.take_parked_partition_frames(namespace) else { + return; + }; + tracing::debug!( + shard = self.id, + namespace_raw = namespace.inner(), + count = frames.len(), + epoch, + "re-dispatching parked partition frames after materialisation" + ); + let mut refused_frames: Vec<ParkedFrame> = Vec::new(); + for frame in frames { + // Only a stamp that exists and disagrees is evidence of a prior + // incarnation; see this function's docs on why `None` is not. + if let Some(parked_epoch) = frame.epoch + && parked_epoch != epoch Review Comment: Fixed as suggested, and both directions still reject. Ahead now bumps a separate `partition_frames_rejected_ahead_total` and logs at `debug`; behind keeps `partition_frames_rejected_stale_total` and the `warn`, so the anomaly counter keeps its contract instead of firing on ordinary delete + recreate churn. ########## core/shard/src/lib.rs: ########## @@ -995,9 +996,50 @@ where /// materialised the namespace (post-`CreateTopic` convergence window). /// Parked here instead of dropped -- there is no consensus retransmit /// driver in production yet -- and re-dispatched when the matching - /// `ReconcileOp::InsertOwned` lands. Bounded per namespace; overflow - /// drops the frame (at-least-once: client/primary retries recover). - pending_partition_frames: RefCell<HashMap<IggyNamespace, Vec<Message<GenericHeader>>>>, + /// `ReconcileOp::InsertOwned` lands with the epoch they were stamped + /// against. Bounded per namespace; a full buffer sheds via + /// [`ParkOutcome::Overflow`] so the caller can still answer. + /// + /// An entry only drains when the namespace materialises or leaves committed + /// metadata, so the reconciler reclaims the ones that will do neither -- see + /// `partition_reconciler::reconcile_parked_frames`. Without that sweep a + /// namespace whose build keeps failing would hold its frames for the process + /// lifetime while every client waited out its read timeout. + /// + /// [`BTreeMap`], not `HashMap`: [`Self::parked_namespaces`] feeds the + /// reconciler sweep, which answers frames in the order it walks them. + /// `std::collections::HashMap` seeds its hasher per process, so iteration + /// order would vary run to run for identical committed state, making the + /// simulator's deny ordering unreproducible for a fixed seed -- the same + /// hazard `router.rs` documents as its reason for `select_biased!`. + pending_partition_frames: RefCell<BTreeMap<IggyNamespace, Vec<ParkedFrame>>>, + + /// Running sum of [`parked_footprint`] over every frame in + /// [`Self::pending_partition_frames`], maintained at each mutation site. + /// + /// Recomputing it per arriving frame is O(all parked frames): the budget + /// admits ~262k entries at 256 bytes each, so filling the buffer would cost Review Comment: Fixed. The doc now says 4096 entries and ~8.4M visits, keyed off `MAX_PARKED_BYTES` and the `MESSAGE_ALIGN` floor rather than `MAX_MESSAGE_SIZE`. ########## core/server-ng/src/partition_reconciler.rs: ########## @@ -280,6 +344,9 @@ struct PassCounters { /// blocked by a consumer barrier or by a rejoin whose offsets land via /// journal repair, and neither unblocking bumps `Streams::revision`. trims_pending: usize, + /// Namespaces whose parked frames were answered because this shard is not Review Comment: Fixed. The doc now covers all bump sites and separates the two outcomes: aging answers client requests, discarding also destroys prepares. It also says it counts namespaces, not frames. -- 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]
