This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch part-mat-barrier in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 397f1a11e757f53b8ec11059d5b90a6cde32caab Author: Krishna Vishal <[email protected]> AuthorDate: Sat Aug 1 03:05:54 2026 +0530 feat(server-ng): close the partition materialisation race at the owner --- core/integration/tests/cluster/mod.rs | 1 + .../cluster/multi_shard_partition_convergence.rs | 201 +++++ core/metadata/src/stm/stream.rs | 16 +- core/server-ng/Cargo.toml | 7 + core/server-ng/src/dispatch.rs | 68 +- core/server-ng/src/partition_reconciler.rs | 980 ++++++++++++++++++++- core/shard/src/lib.rs | 549 ++++++++++-- core/shard/src/metrics.rs | 108 ++- core/shard/src/router.rs | 40 +- core/simulator/src/lib.rs | 17 + 10 files changed, 1815 insertions(+), 172 deletions(-) diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs index 93c631bd0..555f37dc6 100644 --- a/core/integration/tests/cluster/mod.rs +++ b/core/integration/tests/cluster/mod.rs @@ -16,3 +16,4 @@ // under the License. mod client_table_restart; +mod multi_shard_partition_convergence; diff --git a/core/integration/tests/cluster/multi_shard_partition_convergence.rs b/core/integration/tests/cluster/multi_shard_partition_convergence.rs new file mode 100644 index 000000000..82376abf2 --- /dev/null +++ b/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 +/// incarnation" denials in the server log across a full run). Driving the fence +/// needs a produce concurrent with the delete, from a second connection -- +/// worth adding, but it is a different test. What this one catches is the +/// steady-state failure modes: a rebuild that wedges, or stale segments +/// surviving the delete and being served under the recycled identity. +#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation = "2"))] +async fn given_two_shards_when_recreating_a_topic_should_serve_only_the_new_incarnation( + 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 { + let name = topic_name(index); + let topic = Identifier::named(&name).unwrap(); + create_topic(&client, &stream, &name).await; + produce(&client, &stream, &topic, &format!("first-{index}")).await; + + client + .delete_topic(&stream, &topic) + .await + .unwrap_or_else(|error| panic!("delete_topic {name}: {error}")); + create_topic(&client, &stream, &name).await; + produce(&client, &stream, &topic, &format!("second-{index}")).await; + + // Only the second incarnation's message may be readable: the first + // partition's segments went with the delete, and a write admitted + // against that incarnation would have been erased with it. + assert_eq!( + poll_payloads(&client, &stream, &topic).await, + vec![format!("second-{index}")], + "topic {index} must serve only the recreated incarnation" + ); + } +} diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index 5996bcff9..e4c533b97 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -1175,15 +1175,29 @@ impl Streams { /// byte-identical namespace. That difference is the only thing separating /// the two incarnations, so callers can use it to tell a materialised /// partition apart from the committed one it is impersonating. + /// This sits on the per-request incarnation fence + /// (`IggyShard::serves_committed_incarnation`) and on the park stamp, so it + /// runs once per partition request rather than once per reconciler pass. A + /// plain scan of `partitions` would therefore cost ~N element visits per + /// request on an N-partition topic. Partitions are pushed in dense id order by + /// `CreateTopicWithAssignments` / `CreatePartitionsWithAssignments`, so the + /// direct index hits in one step; the scan stays as the fallback because + /// nothing in the type enforces that density. #[must_use] pub fn created_revision_for_namespace(&self, namespace: IggyNamespace) -> Option<u64> { self.inner.read(|inner| { let stream = inner.items.get(namespace.stream_id())?; let topic = stream.topics.get(namespace.topic_id())?; + let partition_id = namespace.partition_id(); + if let Some(partition) = topic.partitions.get(partition_id) + && partition.id == partition_id + { + return Some(partition.created_revision); + } topic .partitions .iter() - .find(|partition| partition.id == namespace.partition_id()) + .find(|partition| partition.id == partition_id) .map(|partition| partition.created_revision) }) } diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml index d710066d5..35ac350b3 100644 --- a/core/server-ng/Cargo.toml +++ b/core/server-ng/Cargo.toml @@ -185,6 +185,13 @@ vergen-git2 = { workspace = true } assert_cmd = { workspace = true } bytemuck = { workspace = true } iggy = { workspace = true } +# The reconciler's unit tests assert on `ShardMetrics` snapshots and +# `IggyShard::parked_frame_count`, which are gated to test/simulator builds so +# they cannot grow production callers. `shard`'s own `cfg(test)` is false when it +# is compiled as our dependency, so the feature is how those accessors become +# visible here. Dev-only: a production `cargo build -p iggy-server-ng` does not +# resolve dev-dependencies, so nothing extra is compiled in. +shard = { path = "../shard", features = ["simulator"] } tokio = { workspace = true, features = ["full", "test-util"] } [lints.clippy] diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index b6bea8cd8..d7f31bdd4 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -1099,13 +1099,14 @@ pub(crate) async fn dispatch_partition_request<B, MJ, S>( send_partition_deny_reply(shard, transport_client_id, &header, status).await; return; } - // Convergence wait: a CreateTopic commit returns to the client - // before the per-shard reconcilers seed routing rows and - // materialise the partition (next wake/periodic tick). The SDK - // does not replay sends, so an immediately-following partition op - // would be dropped as unroutable. Absorb that window here with a - // bounded wait; steady-state sends (row present, partition probed - // once) skip it entirely. + // Convergence wait: a CreateTopic commit returns to the client before the + // per-shard reconcilers seed routing rows and materialise the partition + // (next wake/periodic tick). An op arriving inside that window is not lost + // if it skips this wait -- `router::route_typed` falls back to the hash + // assignment, and the owning shard parks it -- so this is an admission + // courtesy that keeps the steady state off that park buffer, not a + // correctness gate. See `wait_for_partition_routable`, which spells out why + // there is no owner-readiness probe here any more. if !wait_for_partition_routable(shard, IggyNamespace::from_raw(namespace)).await { // The op never reached the partition plane, so it is safe to re-issue // anywhere -- the same contract the plane itself answers for a @@ -1886,13 +1887,26 @@ async fn send_empty_partition_reply<B, MJ, S>( } } -/// Wait (bounded) until `namespace` is routable: this shard's routing row -/// exists and the owning shard answers a probe read (partition -/// materialised). Fast path: row already present -> no probe, no wait. +/// Wait (bounded) until this shard holds a routing row for `namespace`. Fast +/// path: row already present -> no wait. /// -/// Covers the post-`CreateTopic` convergence window where the metadata -/// commit has returned to the client but the per-shard reconcilers have -/// not yet seeded routing rows / materialised partitions. +/// Covers the post-`CreateTopic` convergence window where the metadata commit +/// has returned to the client but the per-shard reconcilers have not yet seeded +/// routing rows. This is an admission courtesy, not a correctness gate: the row +/// is a cache of the deterministic hash assignment and may exist before the +/// owner has materialised anything, so its presence proves only where the +/// partition belongs. What makes an early arrival safe is the owning shard +/// itself - `park_if_unmaterialised` holds the frame until its partition lands, +/// and `serves_committed_incarnation` refuses to serve a mismatched +/// incarnation. Waiting here simply keeps the steady state off that park +/// buffer, whose overflow is the one path that still sheds a request without +/// replying (`frame_drops_total{variant=partition,reason=park_overflow}`). +/// +/// Deliberately no owner-readiness probe. One used to run here, on the theory +/// that the table could not be trusted; it could not close the window either, +/// because the fast path above skipped it in exactly the case it was meant to +/// cover - a row seeded from the hash by a shard that owns nothing. Readiness +/// belongs to the owner, which is where it is now enforced. #[allow(clippy::future_not_send)] async fn wait_for_partition_routable<B, MJ, S>( shard: &Rc<ShellShard<B, MJ, S>>, @@ -1910,9 +1924,6 @@ where // bus sleep advances virtual time, whereas `Instant::now` would not. const MAX_ATTEMPTS: u32 = 60; - if shard.shards_table().shard_for(namespace).is_some() { - return true; - } let mut attempts = 0u32; while shard.shards_table().shard_for(namespace).is_none() { if attempts >= MAX_ATTEMPTS { @@ -1921,30 +1932,7 @@ where attempts += 1; shard.bus.sleep(ATTEMPT_DELAY).await; } - // The local row is seeded by THIS shard's reconciler; the owner - // materialises the partition on its own pass. Probe with a cheap read - // until the owner answers, so the write below normally clears the - // owner's "partition not initialized" guard. Not a hard guarantee: the - // partition can de-materialise between this probe and the dispatch, but - // the park/tombstone path re-checks and the client retries. - while attempts < MAX_ATTEMPTS { - match shard - .partition_read( - namespace, - PartitionRead::ConsumerOffset { - consumer: PollingConsumer::Consumer(0, 0), - }, - ) - .await - { - Some(PartitionReadReply::NotFound) | None => { - attempts += 1; - shard.bus.sleep(ATTEMPT_DELAY).await; - } - Some(_) => return true, - } - } - false + true } /// The 16-byte `PolledMessages` body with zero messages diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs index da549bb1b..228f826c9 100644 --- a/core/server-ng/src/partition_reconciler.rs +++ b/core/server-ng/src/partition_reconciler.rs @@ -24,59 +24,117 @@ //! `ReconcileOp::InsertOwned` for pump-side apply. //! - ghosts: two-phase tombstone, disk delete, `ConfirmRemove`. //! -//! # Dormant race: reply ships before partition materialises +//! # Materialisation race: the reply ships before the partition exists //! //! `metadata::on_ack` fires the commit notifier and emits the wire reply //! immediately after STM apply, but the owning shard's reconciler wakes //! asynchronously and only enqueues `ReconcileOp::InsertOwned` once //! `build_partition_fresh` finishes (mkdir + segment open + fallocate, -//! multi-millisecond). Until the pump drains the queue, -//! `shards_table.shard_for(ns)` returns `None` and `router::route_typed` -//! drops any partition op silently with -//! `frame_drops_total{variant=partition,reason=unroutable}` (see -//! `shard/src/router.rs:147-162`). +//! multi-millisecond). A client that produces the instant `create_topic` +//! returns therefore races the partition into existence, on every shard at +//! once. //! -//! Today this race is **unreachable** from any SDK: the `vsr` feature -//! gate only wires VSR framing for `users` + `personal_access_tokens`; -//! `topics.rs` and `partitions.rs` `binary_impls` still emit pre-VSR -//! encoding server-ng can't dispatch. The first SDK trait impl that adds -//! a `#[cfg(feature = "vsr")]` branch for `create_topic` or -//! `send_messages` surfaces the race as a silent first-produce drop after -//! every `create_topic`. +//! The race is closed at the **owning shard**, not by the routing table: //! -//! TODO: block VSR-ification of `topics.rs` / `partitions.rs` -//! `binary_impls` on a materialization barrier. Two changes, both -//! required together (one without the other does not close the race): +//! - `router::route_typed` treats a missing row as "not seeded yet", not +//! "unroutable", and falls back to `calculate_shard_assignment`. The frame +//! always reaches the shard that will own the partition. +//! - `IggyShard::park_if_unmaterialised` holds it there until the matching +//! `InsertOwned` lands, then re-queues it onto this shard's inbox -- but not to +//! a DIFFERENT incarnation than the one it was addressed to. Each parked frame +//! carries the committed `created_revision` observed when it was parked, and a +//! drain whose epoch disagrees with that stamp answers the client instead of +//! serving it: recycled slab keys make the namespace byte-identical, so such a +//! frame would otherwise land a dead topic's write inside the topic that +//! replaced it. A frame parked with NO stamp is served -- see +//! `redispatch_parked_frames` for why absence of a committed revision is not +//! evidence of a prior incarnation. Re-queuing appends, so a parked frame is +//! ordered behind whatever is already in the inbox rather than restored to its +//! original arrival position; a frame the inbox refuses is re-parked for the +//! next pass rather than answered, since the deny would ride the same full +//! sender. +//! - `IggyShard::serves_committed_incarnation` refuses a namespace whose +//! committed `created_revision` disagrees with the epoch on the local row, so +//! a request arriving mid-teardown cannot be acked against the incarnation +//! teardown is about to erase. It discriminates the shard's own state, not the +//! frame's provenance, which is why the park stamp above is separate. +//! - Nothing is left unanswered: a tombstoned namespace, an overflowing park +//! buffer, and a namespace this shard has given up materialising +//! ([`reconcile_parked_frames`]) all reply with a retriable status, so a +//! lockstep transport never waits out its read timeout on silence. //! -//! 1. **Owner becomes the sole writer of its own `shards_table` row.** -//! Today any non-owning shard's reconciler independently seeds an -//! `InsertRouted` row the moment it observes committed metadata (see -//! the bullet above) -- a pure hash computation, no coordination with -//! the owner. That is fine for routing (`calculate_shard_assignment` -//! is a static function of the namespace, identical on every shard, -//! no placement decision to propagate) but it means a row can exist -//! before the owner's own `build_partition_fresh` has finished. -//! Non-owning shards must stop writing this row ahead of the owner. -//! 2. **Owner pushes the row to every other shard once materialised**, -//! instead of each shard independently guessing it. Cheap: this is a -//! same-node, cross-shard-core message (the existing `ReconcileOp` -//! inter-shard channel already carries `InsertOwned`/`ConfirmRemove`; -//! extend it with a push variant), not a network round trip to -//! another replica. +//! `shards_table` is therefore a **cache of a deterministic hash**, never a +//! readiness proof: every shard derives the same rows from the same committed +//! metadata, and a row may exist before its partition does. Nothing may treat +//! presence as "the owner is ready" - `dispatch::wait_for_partition_routable` +//! documents why the owner-readiness probe that used to live there was both +//! unnecessary and ineffective. //! -//! With both in place, `shards_table.shard_for(ns).is_some()` on ANY -//! shard implies the owner has already materialised the partition, so -//! `dispatch::wait_for_partition_routable`'s second-phase `partition_read` -//! probe (the owner-readiness check a router-side reader currently has to -//! do by hand, since the table alone can't be trusted) becomes -//! unnecessary; a single `shards_table` poll is a sufficient barrier for -//! both server-ng-shard routing AND the pump/client reply. The heavier -//! "shard 0 holds the client reply until every assigned shard acks" -//! design was the original idea here; this is a smaller, cheaper -//! alternative scoped to the local (same-node) table-visibility problem -//! only, not the reply-timing one -- the create-topic reply can still -//! ship on metadata commit as it does today, since the retry loop that -//! consumes `shards_table` is what actually needs the invariant. +//! Keeping the table a hint is what makes it repairable: a pass that runs +//! re-derives the full row set from committed metadata, so a lost row is +//! rewritten. Note the qualifier -- the revision fast-skip below returns before +//! reading `shards_table` at all, so repair is driven by the signals that defeat +//! that skip (a partition-shaping commit, a pending retry, unfinished work, a +//! non-empty park buffer), not by every tick. An earlier design made the owner the sole writer and pushed +//! rows to peers to promote presence into a materialisation proof; it bought +//! nothing the owner-side fences above do not already guarantee, and it traded +//! that level-triggered repair for cross-core delta propagation that has to be +//! ordered, retried, and repaired to stay correct. +//! +//! Park residency is bounded on three axes, because the frame count alone bounds +//! nothing useful (`Message::into_generic` is a retag, so each entry retains its +//! whole buffer, up to 64 MiB): a per-namespace frame cap, a shard-wide byte +//! budget, and an age in reconciler passes. Anything shed or aged out is +//! answered with a retriable status and counted under +//! `frame_drops_total{variant=partition}`. +//! +//! # Known gaps +//! +//! Recorded here because both were previously carried as a TODO on the +//! materialization barrier this module used to promise, and the barrier is gone +//! (see above) while these are not: +//! +//! TODO(krishna): a shed or refused *prepare* has no recovery once its op has +//! reached quorum. `consensus::retransmit_targets` skips entries with +//! `ok_quorum_received`, and the partition plane creates a repair session only +//! in `on_start_view` -- `tick_partitions` re-drives an existing session but +//! cannot open one -- so the backup stays behind `commit_max` until an unrelated +//! view change. It needs a normal-status repair driver. +//! +//! TODO(krishna): re-dispatch APPENDS to the inbox, so a parked prepare loses its +//! arrival position. `router.rs`'s `select_biased!` puts the consensus tick (which +//! runs `apply_reconcile_ops`, and with it the re-dispatch) above the inbox arm, +//! so a parked op N is re-queued *behind* an op N+1 that was already sitting in +//! the inbox. The partition plane then sees N+1 first, rejects it against its +//! backup gap check, and N+1 is gone -- with no normal-status repair driver to +//! refetch it (see the TODO below). Ordering has to be restored at the plane, by +//! buffering out-of-order prepares rather than dropping them, or by re-dispatching +//! through a priority path that preserves op order. +//! +//! TODO(krishna): `serves_committed_incarnation` and the park stamp both call +//! `Streams::created_revision_for_namespace`, now on the per-request fence path. +//! It indexes directly and falls back to a scan only if partition ids are not +//! dense, so the common case is O(1) -- but nothing in the type enforces that +//! density, and a future sparse layout silently reverts every fenced request to a +//! full scan. It wants a partition-id-keyed map in the STM. +//! +//! TODO(krishna): the transient deny answers with `IggyError::TransientNotAccepted`, +//! which the SDK treats as a leader-liveness signal. It replays same-session for +//! its `transient_deadline` first -- which is the right response and usually long +//! enough for the namespace to materialise -- but past that deadline `tcp_client` +//! runs `handle_leader_redirection` and reconnects, re-registering and losing the +//! session. Every cause of a park deny is node-local convergence, so that failover +//! cannot help; it needs a distinct "retry here shortly" code that does not move +//! the client. +//! +//! TODO(krishna): replicated traffic is deliberately exempt from the incarnation +//! fence, since a backup must apply whatever the primary admitted. +//! `PrepareHeader` carries no incarnation, so a backup still holding a prior one +//! cannot tell that an arriving prepare belongs to its replacement. Parked +//! prepares are covered by the epoch stamp above; one arriving against an +//! already-materialised stale incarnation is not. Closing it needs a wire-level +//! discriminator, like `checkpoint_id` on every prepare +//! -- `PrepareHeader.reserved` has room, but it is a `#[repr(C)]` wire change. use crate::bootstrap::ServerNgShard; use crate::partition_helpers::{build_partition_fresh, delete_partitions_from_disk}; @@ -280,6 +338,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 + /// going to materialise them (see [`reconcile_parked_frames`]). + parked_reclaimed: usize, /// Rebuilds deferred until an in-flight `ConfirmRemove` drains. Counted /// so the pass does not arm the fast-skip: the pump's drop clears the /// tombstone and re-wakes us without bumping `Streams::revision`, so an @@ -298,6 +359,7 @@ impl PassCounters { + self.cg_offsets_purged + self.trims_pending + self.deferred + + self.parked_reclaimed } } @@ -322,9 +384,18 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { // `failure_state` non-empty, and a pass that found work it could not // finish (a deferred rebuild, an incomplete trim) leaves // `last_pass_noop` false; any of the three forces the next pass. + // + // A non-empty park buffer is the fourth signal. Parking does not bump + // `revision` and does not wake the reconciler, so without this a frame that + // parks in a converged steady state is held for the process lifetime while + // its client burns the full response read-timeout -- exactly what + // `reconcile_parked_frames` exists to prevent. Held frames also occupy the + // shard-wide byte budget, so one stranded namespace would shed every other + // namespace's legitimate convergence window. if ctx.last_revision.get() == Some(revision) && ctx.last_pass_noop.get() && ctx.failure_state.borrow().is_empty() + && !ctx.shard.has_parked_partition_frames() { trace!( shard = shard_id, @@ -339,6 +410,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { reconcile_additions(ctx, target, &mut counters).await; reconcile_removals(ctx, &target_set, &mut counters).await; + reconcile_parked_frames(ctx, &mut counters); reconcile_consumer_group_offsets(ctx, &mut counters).await; reconcile_segment_truncations(ctx, &mut counters); reconcile_partition_purges(ctx); @@ -366,6 +438,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { backoff_skipped = counters.backoff_skipped, stale = counters.stale, deferred = counters.deferred, + parked_reclaimed = counters.parked_reclaimed, "partition reconciler pass complete" ); } else { @@ -451,7 +524,11 @@ async fn reconcile_additions( let owning_shard = calculate_shard_assignment(&ns, total_shards); if owning_shard != shard_id { - if !shards_table_contains(ctx, ns) { + // 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), @@ -512,6 +589,78 @@ async fn reconcile_additions( } } +/// Answer parked frames for namespaces this shard is not going to materialise. +/// +/// `park_if_unmaterialised` holds a frame until `ReconcileOp::InsertOwned` lands +/// for its namespace, and the only other things that drain the entry are +/// `ConfirmRemove` and `RemoveRouted`. Neither can name a namespace that was +/// never built: it is absent from `IggyPartitions` (so `reconcile_removals` +/// sees no owned ghost) and absent from `shards_table` (the owner seeds a row +/// only via `InsertOwned`, and emits `InsertRouted` only for namespaces it does +/// NOT own). So without this sweep the frames are held for the process +/// lifetime and every waiting client burns its full response read-timeout. +/// +/// Immediate reclaim needs positive evidence that the build will not finish. Two +/// signals carry it: `build_partition_fresh` failed (ENOSPC, EPERM) and is backed +/// off -- the backoff clamps at 60s, well past the client's 30s read timeout, so +/// holding the frames cannot help -- or the namespace does not hash to this shard +/// at all, so no `InsertOwned` for it will ever land here. +/// +/// Absence from the target set is NOT that evidence, which is why this no longer +/// consults it. "Not in the target" covers a namespace that left committed +/// metadata AND one this replica has simply not applied yet, and those are +/// indistinguishable from local state: `snapshot_target_namespaces` reads this +/// node's committed metadata, so a metadata-lagging backup reports a namespace it +/// is milliseconds from committing exactly as it reports a deleted one. Reclaiming +/// on that reading destroys the in-flight traffic the park buffer exists to hold +/// (silently, for a replicated prepare, which has no client to answer). The stale +/// reading was doubly wrong: `target_set` is snapshotted before +/// `reconcile_additions` awaits `build_partition_fresh`, so a topic committing +/// during those awaits was judged against a set that predates it. +/// +/// Everything without that evidence -- building, still committing, or genuinely +/// deleted -- is aged instead. [`shard::IggyShard::age_parked_partition_frames`] +/// answers frames past `MAX_PARKED_PASSES`, so residency stays bounded and no +/// client waits out its read timeout; the deleted case simply takes a few passes +/// rather than one. The bound is residency only -- the SDK replays the identical +/// request, so answering a late frame does not stop its operation from being +/// applied late (see `ParkedFrame::passes`). +fn reconcile_parked_frames(ctx: &ReconcilerCtx, counters: &mut PassCounters) { + let parked = ctx.shard.parked_namespaces(); + if parked.is_empty() { + return; + } + let partitions = ctx.shard.plane.partitions(); + let total_shards = u32::from(ctx.total_shards); + let now = Instant::now(); + for ns in parked { + if partitions.contains(&ns) { + continue; + } + // This shard will never materialise a namespace it does not own. The + // frame got here through a stale `shards_table` row (the table is a hash + // cache, never a readiness proof), so no `InsertOwned` will ever drain it + // and aging is otherwise its only exit. + let not_ours = calculate_shard_assignment(&ns, total_shards) != ctx.shard.id; + let backed_off = ctx.is_backed_off(ns, FailureCause::Add, now); + if !not_ours && !backed_off { + if ctx.shard.age_parked_partition_frames(ns) > 0 { + counters.parked_reclaimed += 1; + } + continue; + } + debug!( + shard = ctx.shard.id, + ns_raw = ns.inner(), + not_ours, + backed_off, + "reclaiming parked frames for a namespace this shard will not materialise" + ); + ctx.shard.reclaim_parked_partition_frames(ns); + counters.parked_reclaimed += 1; + } +} + async fn reconcile_removals( ctx: &ReconcilerCtx, target_set: &AHashSet<IggyNamespace>, @@ -811,8 +960,11 @@ fn fetch_partition_stats( }) } -fn shards_table_contains(ctx: &ReconcilerCtx, ns: IggyNamespace) -> bool { - ctx.shard.shards_table().shard_for(ns).is_some() +/// `true` when this shard's routing row for `ns` already records `epoch`. A row +/// carrying any other epoch (or none) is stale and must be rewritten, since the +/// namespace is byte-identical across incarnations. +fn shards_table_has_epoch(ctx: &ReconcilerCtx, ns: IggyNamespace, epoch: u64) -> bool { + ctx.shard.shards_table().epoch_for(ns) == Some(epoch) } /// Enforce committed `TruncatePartition` watermarks: for each owned partition @@ -898,7 +1050,9 @@ mod tests { use iggy_binary_protocol::requests::topics::{ CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest, }; - use iggy_binary_protocol::{Command2, Operation, PrepareHeader, WireIdentifier}; + use iggy_binary_protocol::{ + Command2, GenericHeader, Operation, PrepareHeader, RequestHeader, WireIdentifier, + }; use message_bus::IggyMessageBus; use metadata::IggyMetadata; use metadata::MuxStateMachine; @@ -969,6 +1123,83 @@ mod tests { msg } + /// Build a partition-plane replicated `Prepare` for `namespace`, as a backup + /// receives it from the primary. The frame a client never sees: it has no + /// client to answer, so anything that discards it is silent data loss. + fn build_partition_prepare(namespace: IggyNamespace, op: u64) -> Message<GenericHeader> { + let header_size = size_of::<PrepareHeader>(); + let mut msg = Message::<PrepareHeader>::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::<PrepareHeader>( + &mut msg.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes form a valid PrepareHeader"); + header.command = Command2::Prepare; + header.size = u32::try_from(header_size).expect("prepare size fits u32"); + header.operation = Operation::SendMessages; + header.namespace = namespace.inner(); + header.op = op; + msg.into_generic() + } + + async fn park_one_prepare(shard: &TestShard, namespace: IggyNamespace, op: u64) { + shard + .on_message(build_partition_prepare(namespace, op)) + .await; + } + + /// Build a partition-plane client `Request` for `namespace`, as the pump + /// receives it off the wire. Only the routing fields matter: parking reads + /// `operation` + `namespace` and never touches the body. + fn build_partition_request(namespace: IggyNamespace) -> Message<GenericHeader> { + let header_size = size_of::<RequestHeader>(); + let mut msg = Message::<RequestHeader>::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>( + &mut msg.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes form a valid RequestHeader"); + header.command = Command2::Request; + header.size = u32::try_from(header_size).expect("request size fits u32"); + header.operation = Operation::SendMessages; + header.namespace = namespace.inner(); + // Header validation rejects a zero session / request on a non-register + // op, and the park path runs after that validation. + header.session = 1; + header.request = 1; + header.client = 1; + msg.into_generic() + } + + /// Park one client request for `namespace` through the real pump entry + /// point, so the epoch stamp and the park accounting are the production + /// ones. The namespace must be unmaterialised, or the frame is delivered to + /// the plane instead of parked. + async fn park_one_request(shard: &TestShard, namespace: IggyNamespace) { + shard.on_message(build_partition_request(namespace)).await; + } + + /// [`build_partition_request`] with `body_len` trailing payload bytes, so a + /// test can drive the park buffer's byte budget rather than its frame cap. + fn build_partition_request_sized( + namespace: IggyNamespace, + body_len: usize, + ) -> Message<GenericHeader> { + let header_size = size_of::<RequestHeader>(); + let total_size = header_size + body_len; + let mut msg = Message::<RequestHeader>::new(total_size); + let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>( + &mut msg.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes form a valid RequestHeader"); + header.command = Command2::Request; + header.size = u32::try_from(total_size).expect("request size fits u32"); + header.operation = Operation::SendMessages; + header.namespace = namespace.inner(); + header.session = 1; + header.request = 1; + header.client = 1; + msg.into_generic() + } + fn assignment(partition_id: u32, consensus_group_id: u64) -> CreatedPartitionAssignment { CreatedPartitionAssignment { partition_id, @@ -1138,6 +1369,46 @@ mod tests { Rc::new(shard) } + /// [`build_test_shard`] with this shard's own inbox wired up, for the tests + /// that assert on work handed back to the pump (transient denies, parked-frame + /// re-dispatch). The receiver comes back so the caller keeps it alive and can + /// drain it; without a live receiver every `try_send` reports `Disconnected`. + fn build_test_shard_with_inbox( + shard_id: u16, + config: &ServerNgConfig, + mux: TestMux, + capacity: usize, + ) -> (Rc<TestShard>, shard::Receiver<shard::ShardFrame>) { + let (tx, rx) = shard::shard_channel(shard_id, capacity); + let mut shard = Rc::into_inner(build_test_shard(shard_id, config, mux)) + .expect("freshly built shard is uniquely owned"); + shard.attach_self_sender(tx); + (Rc::new(shard), rx) + } + + /// Drain a test shard's inbox into `(re-dispatched frames, staged client + /// sends)`: served parked frames vs answers headed for a client. + fn drain_inbox(rx: &shard::Receiver<shard::ShardFrame>) -> (usize, usize) { + let mut served = 0; + let mut answered = 0; + while let Ok(frame) = rx.try_recv() { + match frame { + shard::ShardFrame::Consensus { .. } => served += 1, + shard::ShardFrame::Lifecycle(shard::LifecycleFrame::ForwardClientSend { + .. + }) => answered += 1, + _ => {} + } + } + (served, answered) + } + + /// Count the `ForwardClientSend` frames sitting in a test shard's inbox: the + /// staged transient denies, which is what actually reaches a client. + fn drain_staged_client_sends(rx: &shard::Receiver<shard::ShardFrame>) -> usize { + drain_inbox(rx).1 + } + fn make_ctx( shard: Rc<TestShard>, total_shards: u16, @@ -2104,4 +2375,617 @@ 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. + #[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); + + // 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. + assert_eq!( + shard.metrics().partition_requests_denied_transient_value(), + 1, + "the shed request must be answered with a retriable status" + ); + } + + /// 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() { + const BODY: usize = 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-bytes"); + seed_topic(&mux, 2, 0, "topic-bytes", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ns = IggyNamespace::new(0, 0, 0); + + for _ in 0..PARK_CAP { + shard + .on_message(build_partition_request_sized(ns, BODY)) + .await; + if park_overflow_count(&shard) > 0 { + break; + } + } + + assert!( + park_overflow_count(&shard) > 0, + "1 MiB frames must reach the shard-wide byte budget" + ); + assert!( + shard.parked_frame_count(ns) < PARK_CAP, + "the byte budget must bite before the frame cap; parked {} of {PARK_CAP}", + shard.parked_frame_count(ns) + ); + } + + /// Mirrors `MAX_PARKED_PER_NAMESPACE` in `shard::park_if_unmaterialised`. + const PARK_CAP: usize = 128; + /// Mirrors `MAX_PARKED_PASSES`. + const PARK_MAX_PASSES: u32 = 3; + + /// `cluster::multi_shard_partition_convergence` exists to drive the + /// cross-core path, which only happens for namespaces the connection's shard + /// does not own. That property is a murmur3 outcome, invisible from the + /// integration test itself: it would stay green while silently degrading to + /// single-shard if the hash or the shard count changed. Pin it here, where + /// the assignment is a pure function, over the namespaces that test creates + /// (stream 0, topics 0..8, partition 0 - the slab keys the STM hands out). + #[test] + fn integration_topic_set_straddles_both_shards() { + let owners: Vec<u16> = (0..8) + .map(|topic_id| calculate_shard_assignment(&IggyNamespace::new(0, topic_id, 0), 2)) + .collect(); + let on_shard_one = owners.iter().filter(|owner| **owner == 1).count(); + assert!( + on_shard_one > 0 && on_shard_one < owners.len(), + "the integration test's topics must land on both shards, else it \ + silently stops covering the cross-core path; got {owners:?}" + ); + } + + fn park_overflow_count(shard: &TestShard) -> u64 { + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::PARTITION, + shard::metrics::frame_drop_reason::PARK_OVERFLOW, + ) + } } diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 835308e6d..9aff5a682 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -33,6 +33,7 @@ use consensus::{ }; #[cfg(any(test, feature = "simulator"))] use crossfire::AsyncRxTrait; +use crossfire::TrySendError; use futures::FutureExt; use iggy_binary_protocol::{ Command2, CommitHeader, DoViewChangeHeader, GenericHeader, Operation, PrepareHeader, @@ -58,7 +59,7 @@ use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen}; use shards_table::ShardsTable; use std::cell::{Cell, RefCell}; -use std::collections::{HashMap, VecDeque}; +use std::collections::{BTreeMap, VecDeque}; use std::future::Future; use std::rc::Rc; #[cfg(any(test, feature = "simulator"))] @@ -831,9 +832,33 @@ 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 + /// ~10^10 element visits on the reactor thread, inside the map's + /// `borrow_mut`, collapsing a saturated shard to a few hundred frames/sec + /// while starving the reconciler pass that would drain it. + parked_partition_bytes: Cell<usize>, /// Live ceiling on prepares served per `RequestPrepares` round. Defaults /// to [`REPAIR_CHUNK_MAX`]; server-ng overrides it from @@ -934,7 +959,8 @@ where metrics, metadata_tick_handler: RefCell::new(None), reconcile_queue: RefCell::new(VecDeque::new()), - pending_partition_frames: RefCell::new(HashMap::new()), + pending_partition_frames: RefCell::new(BTreeMap::new()), + parked_partition_bytes: Cell::new(0), metadata_repair: RefCell::new(None), repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), @@ -1157,7 +1183,8 @@ where metrics: crate::metrics::ShardMetrics::for_shard(), metadata_tick_handler: RefCell::new(None), reconcile_queue: RefCell::new(VecDeque::new()), - pending_partition_frames: RefCell::new(HashMap::new()), + pending_partition_frames: RefCell::new(BTreeMap::new()), + parked_partition_bytes: Cell::new(0), metadata_repair: RefCell::new(None), repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), @@ -1174,6 +1201,29 @@ where &self.metrics } + /// Attach this shard's own inbox sender to a shard built by + /// [`Self::without_inbox`], which leaves the mesh empty. + /// + /// Exists for out-of-crate tests: the paths that hand work back to the pump + /// (`stage_transient_deny`, the parked-frame re-dispatch) index + /// `senders[self.id]`, so without it they silently no-op and a test asserting + /// on them proves nothing. The caller must keep the paired receiver alive; + /// dropping it turns every `try_send` into `Disconnected`. + /// + /// # Panics + /// If `sender` is not tagged for this shard, which would route this shard's + /// own frames to a peer. + pub fn attach_self_sender(&mut self, sender: TaggedSender) { + assert_eq!( + sender.shard_id(), + self.id, + "attach_self_sender: sender is tagged for shard {} but this is shard {}", + sender.shard_id(), + self.id + ); + self.senders = vec![sender]; + } + /// `None` removes the handler; subsequent ticks drop with a metric bump. pub fn set_metadata_tick_handler(&self, handler: Option<Rc<dyn Fn()>>) { *self.metadata_tick_handler.borrow_mut() = handler; @@ -1306,34 +1356,7 @@ where PartitionLocation::new(ShardId::new(self_shard_id), epoch), ); self.metrics.record_partition_materialised(); - // Re-dispatch frames that arrived before this partition - // materialised (see `park_if_unmaterialised`). `dispatch` - // re-routes them onto our own inbox, so the pump - // processes them after this drain completes. - let parked = self - .pending_partition_frames - .borrow_mut() - .remove(&namespace); - if let Some(frames) = parked { - tracing::debug!( - shard = self_shard_id, - namespace_raw = namespace.inner(), - count = frames.len(), - "re-dispatching parked partition frames after materialisation" - ); - for frame in frames { - if let Some(sender) = self.senders.get(self_shard_id as usize) - && let Err(error) = - sender.try_send(ShardFrame::consensus(self_shard_id, frame)) - { - tracing::warn!( - shard = self_shard_id, - namespace_raw = namespace.inner(), - "dropping parked partition frame: inbox rejected: {error:?}" - ); - } - } - } + self.redispatch_parked_frames(namespace, epoch); } ReconcileOp::InsertRouted { namespace, @@ -1400,8 +1423,97 @@ enum ParkOutcome<H> { /// transient status; replicated traffic still flows to the plane, whose /// own tombstone guards drop it. Tombstoned(Message<H>), + /// Namespace is unmaterialised and its park buffer is at capacity. Client + /// requests must be denied with a transient status: the frame is gone, and + /// silence would leave a lockstep transport waiting out its response + /// read-timeout. Replicated traffic is dropped, recovered by retransmit. + Overflow(Message<H>), +} + +/// A partition frame held until its namespace materialises. +/// +/// `epoch` is the committed `created_revision` observed when the frame was +/// parked, or `None` when the namespace had no committed partition to read one +/// from. Delete + recreate recycles the slab keys, so the namespace alone cannot +/// distinguish incarnations: without this stamp a frame parked against the dead +/// incarnation would be drained into its replacement by `InsertOwned` and +/// served, because `serves_committed_incarnation` compares the committed +/// revision against the routing row - both of which describe the NEW +/// incarnation - and never the frame's provenance. +struct ParkedFrame { + epoch: Option<u64>, + /// Reconciler passes this frame has survived. The sweep in + /// `partition_reconciler::reconcile_parked_frames` increments it and + /// reclaims past [`MAX_PARKED_PASSES`], bounding how long a frame can sit + /// here in units the simulator's virtual clock already controls. + /// + /// This bounds RESIDENCY, not staleness. Answering a late frame does not stop + /// its operation from being applied late: the SDK replays the identical + /// request, same payload, for the rest of its response timeout, so an + /// absolute-offset `StoreConsumerOffset` that would have rewound the group on + /// admission rewinds it on the replay too. What the bound buys is a park + /// buffer that cannot accumulate without limit, and a client that learns the + /// outcome from a reply rather than a timeout. + passes: u32, + message: Message<GenericHeader>, +} + +/// Per-namespace ceiling on parked frames. +const MAX_PARKED_PER_NAMESPACE: usize = 128; + +/// Shard-wide ceiling on parked bytes, measured as resident footprint (see +/// [`parked_footprint`]). +/// +/// The per-namespace cap counts frames, and `Message::into_generic` is a retag +/// rather than a copy, so each entry retains its whole buffer -- up to +/// `message_bus::framing::MAX_MESSAGE_SIZE` (64 MiB). Frames alone therefore +/// bound nothing useful: 128 × 64 MiB is 8 GiB for a single namespace, and +/// nothing capped the namespace count. This is the budget that actually bounds +/// residency, so a burst against many un-materialised namespaces sheds instead +/// of exhausting the host. +/// +/// Deliberately well below `MAX_MESSAGE_SIZE`. Sized equal to it, one legal +/// max-size frame consumes the entire shard-wide budget and head-of-line-blocks +/// every other namespace's convergence window. +const MAX_PARKED_BYTES: usize = 16 * 1024 * 1024; + +/// Per-namespace ceiling on parked bytes, so one un-materialised namespace +/// cannot spend the whole shard's budget and shed everyone else's frames. +/// +/// A frame whose own footprint exceeds this can never park; it is shed (and +/// answered) on every attempt until its namespace materialises. That is the +/// intended trade: admitting it would hand a single namespace a quarter of the +/// shard budget, and a shed frame is answered with a retriable status rather +/// than lost. +const MAX_PARKED_BYTES_PER_NAMESPACE: usize = MAX_PARKED_BYTES / 4; + +/// Resident cost of parking a frame of `len` bytes. +/// +/// A parked frame retains its whole [`server_common::iobuf`] buffer, which is +/// allocated at [`MESSAGE_ALIGN`] granularity, so a 256-byte frame occupies +/// 4 KiB. Charging the logical length instead under-counts RSS by up to 16x for +/// header-only frames, which would let an accounted 16 MiB grow to ~256 MiB +/// resident per shard. +const fn parked_footprint(len: usize) -> usize { + len.next_multiple_of(MESSAGE_ALIGN) } +/// Reconciler passes a frame may survive before it is answered rather than held. +/// +/// Passes, not seconds, and deliberately not described in seconds: a pass fires +/// on the periodic interval OR on a commit-tick wake, so the wall-clock window +/// this maps to spans orders of magnitude. `reconcile_periodic_interval` legally +/// reaches 30s, which would put four passes at 120s -- four times the SDK's +/// response read-timeout, so the client times out first and the bound stops +/// being the thing that answers it. Commit-tick wakes collapse it the other way, +/// to tens of milliseconds. It bounds residency in units the simulator's virtual +/// clock governs; it is not a latency guarantee. +/// +/// TODO(krishna): derive this from `reconcile_periodic_interval` and the SDK +/// response timeout so the bound tracks the configured interval instead of +/// assuming one. +const MAX_PARKED_PASSES: u32 = 3; + /// Local message processing — these methods handle messages that have been /// routed to this shard via the message pump. impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T> @@ -1457,7 +1569,7 @@ where // reply, and the transports decode replies in lockstep, // so silence wedges the connection until the SDK's // response read-timeout. - ParkOutcome::Tombstoned(request) => { + ParkOutcome::Tombstoned(request) | ParkOutcome::Overflow(request) => { self.deny_partition_request_transient(request.header()) .await; } @@ -1490,7 +1602,10 @@ where } } } - ParkOutcome::Parked => {} + // Shed under a full park buffer, or parked. Either way there + // is no client awaiting a reply on this node; the primary's + // retransmit redelivers. + ParkOutcome::Overflow(_) | ParkOutcome::Parked => {} } } Ok(MessageBag::PrepareOk(prepare_ok)) => self.on_ack(prepare_ok).await, @@ -1561,10 +1676,11 @@ 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() { tracing::debug!( @@ -1574,16 +1690,279 @@ where "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>() + self.deny_parked_client_request(frame); + } + } + } + + /// 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 cannot drift out of step with it. + fn take_parked_partition_frames(&self, namespace: IggyNamespace) -> Option<Vec<ParkedFrame>> { + 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. Re-parking + /// keeps the frame for the next pass and preserves `passes`, so + /// [`MAX_PARKED_PASSES`] still bounds its residency. + 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 + { + self.reject_stale_parked_frame(namespace, epoch, frame); + continue; + } + let Some(sender) = self.senders.get(self.id as usize) else { + continue; + }; + let passes = frame.passes; + let parked_epoch = frame.epoch; + let Err(error) = sender.try_send(ShardFrame::consensus(self.id, frame.message)) else { + continue; + }; + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::coordinator::classify_try_send_err(&error), + ); + let (refused, disconnected) = match error { + TrySendError::Full(frame) => (frame, false), + TrySendError::Disconnected(frame) => (frame, true), + }; + let ShardFrame::Consensus { message, .. } = refused else { + continue; + }; + if disconnected { + // The pump is gone, so re-parking would hold the frame until + // process exit. Answer a client request; a prepare has nothing + // left to serve it. + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + "re-dispatch of parked partition frame refused: inbox disconnected" + ); + if message.header().command == Command2::Request + && let Ok(request) = message.try_into_typed::<RequestHeader>() { - // Callers are synchronous (`apply_reconcile_ops`), so the - // deny rides the pump's outbound lifecycle path instead of - // an inline bus send. self.stage_transient_deny(request.header()); } + continue; + } + tracing::debug!( + shard = self.id, + namespace_raw = namespace.inner(), + passes, + "re-parking parked partition frame: inbox full" + ); + refused_frames.push(ParkedFrame { + epoch: parked_epoch, + passes, + message, + }); + } + if !refused_frames.is_empty() { + self.repark_partition_frames(namespace, refused_frames); + } + } + + /// Put frames back under `namespace` after a refused re-dispatch, keeping + /// [`Self::parked_partition_bytes`] in step. + /// + /// Deliberately not budget-checked: these bytes were already counted while + /// parked, so re-admitting them cannot grow the total past what it held a + /// moment ago, and shedding here would answer a frame the inbox merely + /// deferred. + fn repark_partition_frames(&self, namespace: IggyNamespace, frames: Vec<ParkedFrame>) { + let restored: usize = frames + .iter() + .map(|frame| parked_footprint(frame.message.as_slice().len())) + .sum(); + self.pending_partition_frames + .borrow_mut() + .entry(namespace) + .or_default() + .extend(frames); + self.parked_partition_bytes + .set(self.parked_partition_bytes.get().saturating_add(restored)); + } + + /// Age every frame parked under `namespace` by one reconciler pass and + /// answer the ones that have outlived [`MAX_PARKED_PASSES`]. Returns the + /// number answered. + /// + /// The bound is in passes rather than wall-clock so the simulator's virtual + /// clock governs it like everything else. It exists to bound residency: a + /// namespace can stay un-materialised indefinitely, and the buffer must not + /// grow with it. See [`ParkedFrame::passes`] for why this is not also + /// staleness protection -- the SDK replays the same request, so answering a + /// late frame does not prevent its operation from being applied late. + pub fn age_parked_partition_frames(&self, namespace: IggyNamespace) -> usize { + let expired = { + let mut pending = self.pending_partition_frames.borrow_mut(); + let Some(frames) = pending.get_mut(&namespace) else { + return 0; + }; + for frame in frames.iter_mut() { + frame.passes += 1; + } + let expired: Vec<ParkedFrame> = frames + .extract_if(.., |frame| frame.passes > MAX_PARKED_PASSES) + .collect(); + if frames.is_empty() { + pending.remove(&namespace); + } + let freed: usize = expired + .iter() + .map(|frame| parked_footprint(frame.message.as_slice().len())) + .sum(); + self.parked_partition_bytes + .set(self.parked_partition_bytes.get().saturating_sub(freed)); + expired + }; + let count = expired.len(); + if count > 0 { + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + count, + "answering parked partition frames that outlived their admission window" + ); + for frame in expired { + self.deny_parked_client_request(frame); } } + count + } + + /// How many frames are parked under `namespace`. Bounded by + /// `MAX_PARKED_PER_NAMESPACE`; a shed frame must never grow it past that. + /// + /// Test/simulator accessor: nothing in production branches on a per-namespace + /// park depth, and gating keeps it that way. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn parked_frame_count(&self, namespace: IggyNamespace) -> usize { + self.pending_partition_frames + .borrow() + .get(&namespace) + .map_or(0, Vec::len) + } + + /// Answer every frame parked under `namespace` and drop the entry, without + /// touching the routing table. Used by the reconciler for a namespace it has + /// given up on materialising this pass. + pub fn reclaim_parked_partition_frames(&self, namespace: IggyNamespace) { + self.discard_parked_partition_frames(namespace); + } + + /// Answer a parked frame that will never be served: a client request gets a + /// transient deny so it can re-issue, replicated traffic is dropped and + /// recovered by the primary's retransmit. Callers are synchronous + /// (`apply_reconcile_ops`, the reconciler sweep), so the deny rides the + /// pump's outbound lifecycle path instead of an inline bus send. + fn deny_parked_client_request(&self, frame: ParkedFrame) { + if frame.message.header().command == Command2::Request + && let Ok(request) = frame.message.try_into_typed::<RequestHeader>() + { + self.stage_transient_deny(request.header()); + } + } + + /// A parked frame addressed an incarnation this shard no longer holds. + /// Answering the client is what keeps it from waiting out its read timeout; + /// a stale prepare is dropped, since applying it would write a dead + /// incarnation's op into its replacement and diverge this replica. + fn reject_stale_parked_frame( + &self, + namespace: IggyNamespace, + materialised_epoch: u64, + frame: ParkedFrame, + ) { + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + parked_epoch = ?frame.epoch, + materialised_epoch, + replicated = frame.message.header().command != Command2::Request, + "rejecting parked partition frame from a prior incarnation" + ); + self.metrics.record_partition_frame_rejected_stale(); + self.deny_parked_client_request(frame); } /// Park a partition-plane frame whose namespace this shard has not yet @@ -1595,8 +1974,10 @@ where /// client requests instead of feeding them to the plane's silent-drop /// guard, while replicated traffic still flows there. Parked frames are /// re-dispatched by [`Self::apply_reconcile_ops`] once the matching - /// `ReconcileOp::InsertOwned` lands; overflow drops the frame - /// (at-least-once: client/primary retries recover). + /// `ReconcileOp::InsertOwned` lands, and only if the epoch stamped here + /// still matches (see [`ParkedFrame`]); a full buffer reports + /// [`ParkOutcome::Overflow`] so the caller can answer rather than shed + /// silently. fn park_if_unmaterialised<H>( &self, message: Message<H>, @@ -1605,8 +1986,8 @@ where ) -> ParkOutcome<H> where H: iggy_binary_protocol::ConsensusHeader, + M: StreamsFrontend, { - const MAX_PARKED_PER_NAMESPACE: usize = 128; if !operation.is_partition() { return ParkOutcome::Deliver(message); } @@ -1621,23 +2002,68 @@ where if partitions.contains(&namespace) { return ParkOutcome::Deliver(message); } + // Read the committed revision before taking the borrow below: the frame + // is stamped with the incarnation it was addressed to, so a later drain + // can tell it apart from a same-key replacement. + let epoch = self + .plane + .metadata() + .mux_stm + .streams() + .created_revision_for_namespace(namespace); + let frame_cost = parked_footprint(message.as_slice().len()); let mut pending = self.pending_partition_frames.borrow_mut(); - let parked = pending.entry(namespace).or_default(); - if parked.len() >= MAX_PARKED_PER_NAMESPACE { + let parked_bytes = self.parked_partition_bytes.get(); + // Read the entry without `entry().or_default()`: inserting first would + // leave an empty Vec behind on the overflow path below, which reads as a + // parked namespace to the reconciler sweep and its fast-skip guard. + let existing = pending.get(&namespace); + let parked_len = existing.map_or(0, Vec::len); + let namespace_bytes: usize = existing.map_or(0, |frames| { + frames + .iter() + .map(|frame| parked_footprint(frame.message.as_slice().len())) + .sum() + }); + if parked_len >= MAX_PARKED_PER_NAMESPACE + || namespace_bytes.saturating_add(frame_cost) > MAX_PARKED_BYTES_PER_NAMESPACE + || parked_bytes.saturating_add(frame_cost) > MAX_PARKED_BYTES + { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::metrics::frame_drop_reason::PARK_OVERFLOW, + ); + // Shedding drops a frame, so it is logged at `warn` like every other + // drop site: the counter is not registered with a scrape endpoint yet + // (see the TODO in `crate::metrics`), so these logs are the only + // alertable signal. Rate-limited by the buffer being full -- once it + // is, the namespace is already the subject of one warning per + // arriving frame, which is the condition an operator needs to see. tracing::warn!( shard = self.id, namespace_raw = namespace.inner(), - "parked-frame buffer full; dropping partition frame" + parked_frames = parked_len, + namespace_bytes, + parked_bytes, + frame_cost, + "park buffer at capacity; shedding partition frame" ); - return ParkOutcome::Parked; + return ParkOutcome::Overflow(message); } tracing::debug!( shard = self.id, namespace_raw = namespace.inner(), operation = ?operation, + epoch = ?epoch, "parking partition frame until namespace materialises" ); - parked.push(message.into_generic()); + pending.entry(namespace).or_default().push(ParkedFrame { + epoch, + passes: 0, + message: message.into_generic(), + }); + self.parked_partition_bytes + .set(parked_bytes.saturating_add(frame_cost)); ParkOutcome::Parked } @@ -1648,6 +2074,7 @@ where /// this reply (the client recovers via its own read-timeout). #[allow(clippy::future_not_send)] async fn deny_partition_request_transient(&self, request_header: &RequestHeader) { + self.metrics.record_partition_request_denied_transient(); let reply = build_deny_reply_from_request_header( request_header, IggyError::TransientNotAccepted.as_code(), @@ -1680,16 +2107,26 @@ where client_id: request_header.client, msg: reply.into_generic().into_frozen(), }); - if let Some(sender) = self.senders.get(self.id as usize) - && let Err(error) = sender.try_send(frame) - { + let Some(sender) = self.senders.get(self.id as usize) else { + return; + }; + // Count only what was actually handed to the pump: crediting before the + // send reports an answer to a client that never received one, which is + // the opposite of what this counter is read for. + if let Err(error) = sender.try_send(frame) { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::coordinator::classify_try_send_err(&error), + ); tracing::warn!( shard = self.id, client = request_header.client, operation = ?request_header.operation, "dropping transient deny for discarded partition frame: inbox rejected: {error:?}" ); + return; } + self.metrics.record_partition_request_denied_transient(); } #[allow(clippy::future_not_send)] diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 1937a314b..5c69a45bc 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -17,13 +17,18 @@ //! Per-shard frame-drop accounting. //! -//! [`ShardMetrics`] holds `frame_drops_total{variant, reason}`, -//! bumped whenever an inter-shard `try_send` is rejected (`Full` / -//! `Disconnected`) or its target shard id is out of range -//! (`Unroutable`): +//! [`ShardMetrics`] holds `frame_drops_total{variant, reason}`, bumped wherever +//! a frame is shed instead of delivered -- an inter-shard `try_send` rejected +//! (`Full` / `Disconnected`), a target shard id out of range (`Unroutable`), or +//! a buffer at capacity (`ParkOverflow`): //! - [`crate::coordinator::ShardZeroCoordinator`] - fd-transfer delegation. //! - the cross-shard forward closures built in [`crate::builder`]. -//! - `IggyShard::try_send_to_target` - consensus frames. +//! - `IggyShard::try_send_to_target` - consensus and partition frames, labelled +//! by plane. +//! - `IggyShard::park_if_unmaterialised` - partition frames shed because the +//! park buffer is at its frame or byte cap. +//! - `IggyShard::apply_reconcile_ops` - parked frames whose re-dispatch onto +//! this shard's own inbox was refused. //! //! The counter uses atomic interior mutability, safe to bump from `!Send` //! compio reactor contexts. Each shard owns its own instance. It is not @@ -44,8 +49,9 @@ use prometheus_client::metrics::family::Family; /// `variant` describes the dropped frame class; `reason` is `"full"` or /// `"disconnected"` per crossfire `TrySendError`, `"unroutable"` when the /// target shard id has no sender slot, `"delivery_failed"` when the -/// receiver path could not place the frame, or `"misrouted"` when a frame -/// reached a shard that does not own its namespace. +/// receiver path could not place the frame, `"misrouted"` when a frame +/// reached a shard that does not own its namespace, or `"park_overflow"` when +/// an un-materialised namespace's park buffer was already at capacity. /// /// `shard_id` is intentionally NOT a label here: each shard owns its own /// `Family<FrameDropLabel, Counter>`, so the per-shard scope is implicit @@ -69,9 +75,22 @@ pub struct FrameDropLabel { /// counter is not scrape-able yet, see the module doc) and size /// `inbox_capacity` for the worst-case cross-shard reply burst. /// `FORWARD_REPLICA_SEND` is the symmetric variant for replica forwards; -/// VSR retransmit covers its loss so it stays informational. `PARTITION` -/// ticks when a partition-targeted frame cannot be dispatched because the -/// namespace is absent from the local `ShardsTable`. +/// VSR retransmit covers its loss so it stays informational. +/// +/// `PARTITION` covers the partition plane: a frame shed because the namespace +/// had not materialised and its park buffer was at capacity +/// (`reason=park_overflow`), a re-dispatch the shard's own inbox refused, or a +/// routing send the target inbox refused. A shed client request is answered with +/// a retriable status, so the client recovers -- but a shed *prepare* is not +/// covered by retransmit once its op has reached quorum +/// (`consensus::retransmit_targets` skips `ok_quorum_received`, and the +/// partition plane creates a repair session only in `on_start_view`), so it +/// leaves that backup behind until an unrelated view change. +// +// TODO(krishna): give the partition plane a normal-status repair driver so a +// shed or refused prepare is repaired without waiting for a view change. Until +// then `variant=partition` is the only signal that a backup may be stranded +// behind `commit_max`. pub mod frame_drop_variant { pub const CONSENSUS: &str = "consensus"; pub const FD_TRANSFER: &str = "fd_transfer"; @@ -94,16 +113,20 @@ pub mod frame_drop_variant { /// is the receiver-side equivalent: the frame arrived at the owning shard /// but the local registry refused it. `MISROUTED` ticks when the pump /// receives a Consensus frame whose target shard is not `self.id`. +/// `PARK_OVERFLOW` ticks when a partition frame arrives for a namespace this +/// shard has not materialised and the per-namespace park buffer is already at +/// its cap, so the frame is shed with no reply. pub mod frame_drop_reason { pub const FULL: &str = "full"; pub const DISCONNECTED: &str = "disconnected"; pub const UNROUTABLE: &str = "unroutable"; pub const DELIVERY_FAILED: &str = "delivery_failed"; pub const MISROUTED: &str = "misrouted"; + pub const PARK_OVERFLOW: &str = "park_overflow"; } const VARIANT_COUNT: usize = 7; -const REASON_COUNT: usize = 5; +const REASON_COUNT: usize = 6; const VARIANTS: [&str; VARIANT_COUNT] = [ frame_drop_variant::CONSENSUS, @@ -121,6 +144,7 @@ const REASONS: [&str; REASON_COUNT] = [ frame_drop_reason::UNROUTABLE, frame_drop_reason::DELIVERY_FAILED, frame_drop_reason::MISROUTED, + frame_drop_reason::PARK_OVERFLOW, ]; fn variant_index(s: &str) -> Option<usize> { @@ -151,6 +175,8 @@ pub struct ShardMetrics { partitions_materialised_total: Counter, partitions_removed_total: Counter, partitions_reconcile_failures_total: Counter, + partition_frames_rejected_stale_total: Counter, + partition_requests_denied_transient_total: Counter, } impl ShardMetrics { @@ -180,6 +206,8 @@ impl ShardMetrics { partitions_materialised_total: Counter::default(), partitions_removed_total: Counter::default(), partitions_reconcile_failures_total: Counter::default(), + partition_frames_rejected_stale_total: Counter::default(), + partition_requests_denied_transient_total: Counter::default(), } } @@ -222,6 +250,19 @@ impl ShardMetrics { self.partitions_reconcile_failures_total.inc(); } + /// Bumped when a parked partition frame is answered instead of served + /// because it was addressed to an incarnation this shard no longer holds + /// (delete + recreate recycled the namespace's slab keys). Serving it would + /// have written a dead topic's op into the topic that replaced it, so a + /// non-zero value is a caught correctness anomaly, not routine churn. + /// + /// Like every counter in this module it is not scrape-able yet (see the + /// module-level `TODO(hubcio)`); the `warn!` at the reject site is what an + /// operator can actually alert on today. + pub fn record_partition_frame_rejected_stale(&self) { + self.partition_frames_rejected_stale_total.inc(); + } + /// Total frame drops across every `{variant, reason}` pair. /// /// Simulator assertion hook: a run without injected loss must keep @@ -259,6 +300,51 @@ impl ShardMetrics { pub fn partitions_reconcile_failures_value(&self) -> u64 { self.partitions_reconcile_failures_total.get() } + + /// Bumped for every partition request answered with + /// `TransientNotAccepted` rather than served - a namespace mid-teardown, an + /// unverified incarnation, a shed park buffer, or a build this shard gave up + /// on. Counted only once the answer has been handed to a transport or the + /// pump, so it measures answers delivered rather than attempted. The client + /// re-issues, so this is retry pressure rather than error rate, and it is + /// what distinguishes "answered and retried" from the silent sheds it + /// replaced. + pub fn record_partition_request_denied_transient(&self) { + self.partition_requests_denied_transient_total.inc(); + } + + /// Snapshot of `partition_requests_denied_transient_total`. Test/simulator + /// accessor; production scrape goes through the prometheus registry. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partition_requests_denied_transient_value(&self) -> u64 { + self.partition_requests_denied_transient_total.get() + } + + /// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator + /// accessor, readable from any crate under those cfgs so the crates that + /// drive the reconciler can assert a reject did not happen. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partition_frames_rejected_stale_value(&self) -> u64 { + self.partition_frames_rejected_stale_total.get() + } + + /// Snapshot of one `frame_drops_total{variant, reason}` pair, or 0 when the + /// pair is not a known label combination. + /// + /// An unknown pair reports 0 rather than falling through to + /// `Family::get_or_create`, which would materialise a permanent zero-valued + /// series in the registry as a side effect of a read: a typo'd label in a + /// test or assertion would then leak a metric series into production scrapes. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn frame_drop_count(&self, variant: &'static str, reason: &'static str) -> u64 { + match (variant_index(variant), reason_index(reason)) { + (Some(v_idx), Some(r_idx)) => self.cached_counters[v_idx][r_idx].get(), + _ => 0, + } + } } #[cfg(test)] diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 3ebaeb256..74a2676c7 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -188,10 +188,13 @@ where // assignment (every `InsertOwned`/`InsertRouted` row stores // `calculate_shard_assignment`'s result), seeded asynchronously // by each shard's reconciler. A miss therefore means "not seeded - // yet", not "unroutable": fall back to the hash so replication - // frames arriving during the post-commit convergence window are - // not dropped (the partition plane re-checks materialisation on - // the owning shard). + // yet", not "unroutable": fall back to the hash so frames arriving + // during the post-commit convergence window still reach the shard + // that will own the partition. That shard is where earliness is + // resolved -- it parks the frame until its partition materialises + // (`park_if_unmaterialised`) and fences a mismatched incarnation + // (`serves_committed_incarnation`) -- so neither a hit nor a miss + // here carries any claim about readiness. let target = self .shards_table .shard_for(partition_namespace) @@ -222,21 +225,28 @@ where /// Send `message` into `senders[target]`. Honors the `io_uring` reactor /// constraint: never blocks; drops on `Full` / `Disconnected` and - /// records the drop in `frame_drops_total{variant=consensus}`. VSR - /// retransmit recovers consensus drops. A `target` past the end of - /// `senders` (a stored `u16` from `shard_for`, not a trusted index) - /// is dropped with `reason=unroutable` rather than panicking. - /// Metadata frames always pass `target = 0` here, since `is_metadata` - /// operations are owned by shard 0. + /// records the drop in `frame_drops_total`, under `variant=partition` for a + /// partition-plane operation and `variant=consensus` otherwise -- the two + /// have different recovery stories, so folding them into one label hides + /// which one is bleeding. VSR retransmit recovers consensus drops. A + /// `target` past the end of `senders` (a stored `u16` from `shard_for`, not + /// a trusted index) is dropped with `reason=unroutable` rather than + /// panicking. Metadata frames always pass `target = 0` here, since + /// `is_metadata` operations are owned by shard 0. fn try_send_to_target( &self, target: u16, message: Message<GenericHeader>, operation: Operation, ) { + let variant = if operation.is_partition() { + frame_drop_variant::PARTITION + } else { + frame_drop_variant::CONSENSUS + }; let Some(sender) = self.senders.get(target as usize) else { self.metrics - .record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::UNROUTABLE); + .record_frame_drop(variant, frame_drop_reason::UNROUTABLE); tracing::error!( shard = self.id, target, @@ -249,7 +259,7 @@ where Ok(()) => {} Err(TrySendError::Full(_)) => { self.metrics - .record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::FULL); + .record_frame_drop(variant, frame_drop_reason::FULL); tracing::warn!( shard = self.id, target, @@ -258,10 +268,8 @@ where ); } Err(TrySendError::Disconnected(_)) => { - self.metrics.record_frame_drop( - frame_drop_variant::CONSENSUS, - frame_drop_reason::DISCONNECTED, - ); + self.metrics + .record_frame_drop(variant, frame_drop_reason::DISCONNECTED); tracing::warn!( shard = self.id, target, diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 8301a1ecb..a23d9ad71 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1795,6 +1795,18 @@ mod tests { /// without injected loss must keep the counters at zero; a non-zero /// value means an inbox silently shed a frame (undersized capacity or /// a routing bug), which would otherwise hide behind VSR retransmit. + /// + /// `park_overflow` is deliberately NOT excluded. The simulator never wires + /// the partition reconciler (`init_partition` mirrors its outcome directly), + /// so nothing here ever drains the park buffer: a parked frame is never + /// re-dispatched and never swept. A non-zero `park_overflow` in the simulator + /// therefore means frames were shed for a namespace that will never + /// materialise -- which is the very fault class this assert exists to catch, + /// not the back-pressure it would be in production. + /// + /// For the same reason the park buffer must be empty at quiescence: a frame + /// still parked here has no drainer, so it will neither be delivered nor + /// answered. fn assert_no_frame_drops(sim: &Simulator) { for (replica_idx, replica) in sim.replicas.iter().enumerate() { for (shard_idx, shard) in replica.shards.iter().enumerate() { @@ -1803,6 +1815,11 @@ mod tests { 0, "replica {replica_idx} shard {shard_idx} dropped frames without injected loss" ); + assert!( + shard.parked_namespaces().is_empty(), + "replica {replica_idx} shard {shard_idx} left partition frames parked; the \ + simulator wires no reconciler, so nothing will deliver or answer them" + ); } } }
