You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/07/31 21:37:56 UTC
(iggy) branch part-mat-barrier created (now 397f1a11e)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch part-mat-barrier
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 397f1a11e feat(server-ng): close the partition materialisation race at the owner
This branch includes the following new commits:
new 397f1a11e feat(server-ng): close the partition materialisation race at the owner
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(server-ng): close the partition materialisation race at the owner
Posted by kr...@apache.org.
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 <kr...@gmail.com>
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"
+ );
}
}
}