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
The following commit(s) were added to refs/heads/part-mat-barrier by this push:
new 2e78511bb fix: address review comments
2e78511bb is described below
commit 2e78511bb68ed2769dc528515f77609c7054da74
Author: Krishna Vishal <[email protected]>
AuthorDate: Mon Aug 3 23:15:07 2026 +0530
fix: address review comments
---
.../cluster/multi_shard_partition_convergence.rs | 38 +-
core/server-ng/Cargo.toml | 14 +-
core/server-ng/src/partition_reconciler.rs | 504 ++++++++++-------
core/shard/src/lib.rs | 602 ++++++++++++---------
core/shard/src/metrics.rs | 34 +-
5 files changed, 727 insertions(+), 465 deletions(-)
diff --git
a/core/integration/tests/cluster/multi_shard_partition_convergence.rs
b/core/integration/tests/cluster/multi_shard_partition_convergence.rs
index 82376abf2..a26fbd293 100644
--- a/core/integration/tests/cluster/multi_shard_partition_convergence.rs
+++ b/core/integration/tests/cluster/multi_shard_partition_convergence.rs
@@ -51,6 +51,7 @@
#![cfg(feature = "vsr")]
use iggy::prelude::*;
+use integration::harness::TestHarness;
use integration::iggy_harness;
const STREAM: &str = "convergence-stream";
@@ -119,6 +120,25 @@ async fn poll_payloads(
.collect()
}
+/// Assert no park path degraded. All three modes log at `warn`, which the
+/// harness captures by default, so counting them pins the mechanism
positively:
+/// the round trips stay green whether a produce parked and re-dispatched or
was
+/// denied and replayed, and only these markers tell the two apart.
+fn assert_no_degraded_park_paths(harness: &TestHarness) {
+ for marker in [
+ "park buffer at capacity",
+ "outlived their admission window",
+ "rejecting parked partition frame",
+ ] {
+ assert_eq!(
+ harness.server().stdout_occurrences(marker),
+ 0,
+ "server log must not contain {marker:?}: the park buffer degraded
instead of \
+ converging"
+ );
+ }
+}
+
/// 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"))]
@@ -148,6 +168,7 @@ async fn
given_two_shards_when_producing_right_after_create_topic_should_round_t
"topic {index} must return the message produced right after its
creation"
);
}
+ assert_no_degraded_park_paths(harness);
}
/// Delete + recreate reuses the freed slab keys, so the namespace is
@@ -156,14 +177,14 @@ async fn
given_two_shards_when_producing_right_after_create_topic_should_round_t
/// 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.
+/// Does NOT exercise the incarnation fence. Every step is a completed round
+/// trip, so the reconciler converges before the next starts and
+/// `serves_committed_incarnation` has nothing to deny. That is an argument
from
+/// sequencing, not an observation: the deny logs at `debug` and the harness
runs
+/// the server at `info`, so its absence from the log proves nothing. Driving
the
+/// fence needs a produce concurrent with the delete from a second connection,
+/// which is a different test. This one catches the steady-state failures: a
+/// rebuild that wedges, or stale segments 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,
@@ -198,4 +219,5 @@ async fn
given_two_shards_when_recreating_a_topic_should_serve_only_the_new_inca
"topic {index} must serve only the recreated incarnation"
);
}
+ assert_no_degraded_park_paths(harness);
}
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index 695f1ecae..ce6a86d68 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -184,13 +184,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"] }
+# Reconciler unit tests assert on `ShardMetrics` snapshots and
+# `IggyShard::parked_frame_count`, gated to test/simulator so they cannot grow
+# production callers. `shard`'s own `cfg(test)` is false when compiled as our
+# dependency, so the feature is how those accessors become visible. The
resolver
+# keeps a dev-dependency's features out of non-test targets, so a production
+# build still links `shard` without `simulator`.
+shard = { workspace = true, features = ["simulator"] }
tokio = { workspace = true, features = ["full", "test-util"] }
[lints.clippy]
diff --git a/core/server-ng/src/partition_reconciler.rs
b/core/server-ng/src/partition_reconciler.rs
index dafd190cf..b150e9ecb 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -46,14 +46,14 @@
//! 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 rather
-//! than answered, since the deny would ride the same full sender, and the
pump
-//! re-drives it on its next iteration (`retry_reparked_frames`) once a slot
-//! has freed.
+//! replaced it. One gap, recorded below: the stamp is re-derived if the
frame
+//! re-enters the park path from the inbox. A frame parked with NO stamp is
+//! served; see `redispatch_parked_frames` for why a missing 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. A frame the
inbox
+//! refuses is re-parked rather than answered, since the deny would ride the
+//! same full sender, and the pump re-drives it (`retry_reparked_frames`)
once
+//! a slot frees.
//! - `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
@@ -82,17 +82,28 @@
//! 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. The per-namespace byte cap is
waived
-//! for the first frame of an empty entry, or a frame larger than it could
never
-//! park at all and a replicated prepare would be lost outright. Every frame
that
-//! leaves the buffer unserved is counted under
-//! `frame_drops_total{variant=partition}` -- `park_overflow` when it was shed
on
-//! arrival, `park_dropped` when it parked and then aged out or lost its
-//! namespace. A client request is additionally answered with a retriable
status;
-//! a prepare has nobody to answer, which is why the counter is the record.
+//! Park residency is bounded on three axes, since the frame count alone bounds
+//! nothing useful (`Message::into_generic` is a retag, so each entry keeps its
+//! whole buffer, up to 64 MiB): a per-namespace frame cap, byte budgets per
+//! namespace and per shard, and an age in reconciler passes.
+//!
+//! They apply asymmetrically, because the two frame classes fail differently.
A
+//! shed request costs a retry: answered with a retriable status, re-issued by
+//! the SDK. A shed prepare is permanent loss on this replica, with no client
to
+//! answer and `consensus::retransmit_targets` skipping any op that already
+//! reached quorum. So a request is refused the moment admitting it would
cross a
+//! budget, and answered once it outlives `MAX_PARKED_PASSES`; a prepare never
+//! expires by age and is refused only once a budget is already spent. That
caps
+//! prepare residency at one frame of overshoot per budget instead of at the
+//! budget, and is what makes an oversize frame parkable at all: measured
against
+//! an empty entry, a 5 MiB append would fail the per-namespace check on every
+//! attempt. A parked prepare is destroyed only for a namespace this shard
cannot
+//! serve at all, tombstoned or not hashing here.
+//!
+//! Everything leaving the buffer unserved is counted under
+//! `frame_drops_total{variant=partition}`: `park_overflow` when shed on
arrival,
+//! `park_dropped` when it parked and then lost its namespace. A prepare has
+//! nobody to answer, so the counter is the only record it existed.
//!
//! # Known gaps
//!
@@ -100,12 +111,24 @@
//! 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
+//! TODO(krishna): a shed or discarded *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.
+//! view change. It needs a normal-status repair driver. The park policy above
+//! shrinks the exposure to two cases, a genuinely exhausted budget and a
+//! namespace this shard cannot serve, but only the repair driver removes it.
+//!
+//! TODO(krishna): the park stamp is not stable across re-entry. A
re-dispatched
+//! frame still in the inbox when a delete + recreate completes
(`ConfirmRemove`
+//! removes and untombstones in one arm, then the rebuild lands) re-enters
+//! `park_if_unmaterialised` and is re-stamped with the NEW revision and
+//! `passes: 0`, then served against the replacement: the write the stamp
exists
+//! to block. Narrow (a full delete + recreate has to finish while one frame
+//! waits), but the guarantee is not absolute the way the bullet above reads.
+//! Closing it needs the frame to carry provenance through the inbox instead of
+//! re-deriving it on arrival.
//!
//! 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
@@ -344,8 +367,11 @@ 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`]).
+ /// Namespaces the sweep acted on (see [`reconcile_parked_frames`]):
+ /// discarded because this shard cannot serve them, or aged past
+ /// `MAX_PARKED_PASSES` while it still might. Namespaces, not frames, and
+ /// acted on is not answered: aging answers requests, discarding also
+ /// destroys prepares.
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
@@ -457,10 +483,10 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
true
}
-/// Returns the namespaces whose `ReconcileOp::InsertOwned` this pass staged.
-/// They are not in `IggyPartitions` yet -- the pump applies the op on its own
-/// task -- so [`reconcile_parked_frames`] would otherwise read them as
-/// un-materialised and age their frames on the very pass that built them.
+/// Returns the namespaces whose `ReconcileOp::InsertOwned` this pass staged.
The
+/// pump applies the op on its own task, so they are not in `IggyPartitions`
yet
+/// and [`reconcile_parked_frames`] would read them as un-materialised, aging
+/// their frames on the pass that built them.
async fn reconcile_additions(
ctx: &ReconcilerCtx,
target: Vec<(IggyNamespace, u64)>,
@@ -603,46 +629,50 @@ async fn reconcile_additions(
staged
}
-/// Answer parked frames for namespaces this shard is not going to materialise.
+/// Retire parked frames the shard cannot serve, age the ones it might.
+///
+/// `park_if_unmaterialised` holds a frame until `ReconcileOp::InsertOwned`
lands;
+/// the only other drains are `ConfirmRemove` and `RemoveRouted`. Neither can
name
+/// a namespace that was never built, since it is in neither `IggyPartitions`
(no
+/// owned ghost for `reconcile_removals`) nor `shards_table` (the owner seeds a
+/// row only via `InsertOwned`, and emits `InsertRouted` only for namespaces it
+/// does NOT own). Without this sweep the frames are held for the process
lifetime
+/// and every waiting client burns its full read-timeout.
///
-/// `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 discard needs positive evidence this shard can never serve the
+/// namespace, because it destroys prepares outright. Two signals carry it: the
+/// namespace is tombstoned (a wedged disk delete can postpone the
+/// `ConfirmRemove` that would otherwise answer them indefinitely), or it does
not
+/// hash here, so no `InsertOwned` for it ever lands.
///
-/// 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.
+/// A failed `build_partition_fresh` is not that evidence, which is why the
+/// backoff no longer discards. `next_backoff(1)` is one second, so the first
+/// transient ENOSPC destroyed every parked frame for a namespace that
rebuilds a
+/// second later; the 60s clamp the old reasoning leaned on takes seven
+/// consecutive failures.
///
-/// 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.
+/// Absence from the target set is not evidence either. It covers both a
namespace
+/// that left committed metadata and one this replica has not applied yet, and
+/// local state cannot tell them apart: `snapshot_target_namespaces` reads this
+/// node's committed metadata, so a lagging backup reports a namespace it is
+/// milliseconds from committing exactly as it reports a deleted one. It is
also
+/// snapshotted before `reconcile_additions` awaits `build_partition_fresh`,
so a
+/// topic committing during those awaits is judged against a stale set.
///
-/// 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`).
+/// Everything else is aged: building, backed off, still committing, genuinely
+/// deleted, or materialised with frames the inbox refused.
+/// [`shard::IggyShard::age_parked_partition_frames`] answers CLIENT REQUESTS
past
+/// `MAX_PARKED_PASSES` and leaves prepares alone, so no client waits out its
read
+/// timeout and no committed op dies on a local-convergence signal. Residency
+/// only; see `ParkedFrame::passes`.
///
-/// A namespace in `staged_this_pass` is exempt: its `ReconcileOp::InsertOwned`
-/// is queued for the pump but not applied, so it reads as un-materialised here
-/// and would burn one of its three passes on the pass that built it.
+/// `staged_this_pass` is exempt: its `InsertOwned` is queued but not applied,
so
+/// it reads as un-materialised here. Not a one-pass concession.
+/// `reconcile_additions` has no cross-pass guard against a
queued-but-unapplied
+/// op (it tests `partitions.contains`, false the whole time it sits in the
+/// queue), so it re-stages every pass until the pump drains. The exemption
+/// therefore covers arbitrary pump lag; dropping it ages frames on every
+/// commit-driven pass the pump falls behind.
fn reconcile_parked_frames(
ctx: &ReconcilerCtx,
staged_this_pass: &AHashSet<IggyNamespace>,
@@ -654,57 +684,34 @@ fn reconcile_parked_frames(
}
let partitions = ctx.shard.plane.partitions();
let total_shards = u32::from(ctx.total_shards);
- let now = Instant::now();
for ns in parked {
if staged_this_pass.contains(&ns) {
continue;
}
- // Tombstoned but still in the map, so `contains` is true: the fence
- // forbids serving these frames and only `ConfirmRemove` would
otherwise
- // reach them, which a wedged disk delete postpones indefinitely (it
- // retries under `FailureCause::Delete` backoff, clamped at 60s, past
the
- // client's read timeout). Nothing will ever serve them, so reclaim now
- // rather than aging towards the same outcome.
- if partitions.is_tombstoned(&ns) {
- ctx.shard.reclaim_parked_partition_frames(ns);
+ // Tombstoned namespaces are still in the map, so `contains` below
reads
+ // them as materialised.
+ let not_ours = calculate_shard_assignment(&ns, total_shards) !=
ctx.shard.id;
+ let tombstoned = partitions.is_tombstoned(&ns);
+ if tombstoned || not_ours {
+ debug!(
+ shard = ctx.shard.id,
+ ns_raw = ns.inner(),
+ tombstoned,
+ not_ours,
+ "discarding parked frames for a namespace this shard cannot
serve"
+ );
+ ctx.shard.discard_parked_partition_frames(ns);
counters.parked_reclaimed += 1;
continue;
}
- if partitions.contains(&ns) {
- // Materialised, yet frames remain: the re-dispatch found the inbox
- // full and re-parked them. The pump retries on every iteration, so
- // this is only the residency backstop for an inbox that never
- // drains. Aging, not reclaiming: a refusal is transient, and
- // `MAX_PARKED_PASSES` is the same bound the un-materialised case
- // gets. Without it the frames have no exit at all -- the branch
- // below never runs for a namespace already in `IggyPartitions`,
and
- // `reconcile_additions` stages no second `InsertOwned` for one.
- if ctx.shard.age_parked_partition_frames(ns) > 0 {
- counters.parked_reclaimed += 1;
- }
- 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;
+ // Materialised with frames still parked means the re-dispatch hit a
full
+ // inbox and re-parked them. The pump retries every iteration, so
aging is
+ // only the backstop for an inbox that never drains. Without it they
have
+ // no exit: `reconcile_additions` stages no second `InsertOwned` for a
+ // namespace already in `IggyPartitions`.
+ if ctx.shard.age_parked_partition_frames(ns) > 0 {
+ counters.parked_reclaimed += 1;
}
- 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;
}
}
@@ -1199,22 +1206,7 @@ mod tests {
/// 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 = TEST_REQUEST_ID;
- header.client = TEST_CLIENT_ID;
- msg.into_generic()
+ build_partition_request_sized(namespace, 0)
}
/// Park one client request for `namespace` through the real pump entry
@@ -1242,6 +1234,8 @@ mod tests {
header.size = u32::try_from(total_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 = TEST_REQUEST_ID;
header.client = TEST_CLIENT_ID;
@@ -1417,21 +1411,36 @@ 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`.
+ /// [`build_test_shard`] with a sender mesh, for tests asserting on work
+ /// handed back to the pump (transient denies, parked-frame re-dispatch).
+ /// Caller must keep the returned receiver alive; dropping it turns every
+ /// `try_send` into `Disconnected`.
+ ///
+ /// Mesh covers `0..=shard_id` since consumers index `senders[shard_id]`.
+ /// Peer receivers are dropped, so a misroute fails loudly instead of
landing
+ /// in this shard's inbox and reading as success.
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 senders = Vec::with_capacity(usize::from(shard_id) + 1);
+ let mut own_rx = None;
+ for peer in 0..=shard_id {
+ let (tx, rx) = shard::shard_channel(peer, capacity);
+ senders.push(tx);
+ if peer == shard_id {
+ own_rx = Some(rx);
+ }
+ }
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)
+ shard.attach_senders(senders);
+ (
+ Rc::new(shard),
+ own_rx.expect("the loop covers shard_id itself"),
+ )
}
/// Drain a test shard's inbox into `(re-dispatched frames, staged client
@@ -2592,7 +2601,7 @@ mod tests {
// Drain path 3: an explicit reclaim.
park_one_request(&shard, never).await;
assert!(shard.has_parked_partition_frames());
- shard.reclaim_parked_partition_frames(never);
+ shard.discard_parked_partition_frames(never);
assert!(
!shard.has_parked_partition_frames(),
"reclaim must debit the parked-byte total"
@@ -2740,23 +2749,24 @@ mod tests {
);
}
- /// 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.
+ /// Namespace stays committed, `build_partition_fresh` keeps failing. One
+ /// failure must destroy nothing: `next_backoff(1)` is a second, the
rebuild
+ /// usually lands on the next pass, and the sweep cannot tell a transient
+ /// ENOSPC from a permanent one. Request gets the age bound, prepare is
kept.
#[compio::test]
- async fn parked_frames_are_reclaimed_while_the_build_is_backed_off() {
+ async fn a_backed_off_build_ages_requests_and_retains_prepares() {
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 (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 16);
let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
let ns = IggyNamespace::new(0, 0, 0);
park_one_request(&shard, ns).await;
+ park_one_prepare(&shard, ns, 7).await;
assert_eq!(shard.parked_namespaces(), vec![ns]);
// Stand in for a failed build (ENOSPC / EPERM): the additions pass
skips
@@ -2768,9 +2778,40 @@ mod tests {
!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"
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 2,
+ "one failed build must not destroy anything; the backoff is a
second"
+ );
+
+ // Held only until the request outlives the admission window.
+ for _ in 0..PARK_MAX_PASSES {
+ reconcile_pass(&ctx).await;
+ }
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "the request must age out, so no client waits out its read timeout"
+ );
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 1,
+ "and it must be answered, not dropped"
+ );
+ assert_eq!(
+ park_dropped_count(&shard),
+ 0,
+ "the prepare must be retained: destroying it is unrecoverable"
+ );
+
+ // Retained across an unbounded number of further passes.
+ for _ in 0..(PARK_MAX_PASSES * 4) {
+ reconcile_pass(&ctx).await;
+ }
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "prepares are bounded by the byte budget, never by the age bound"
);
}
@@ -2836,11 +2877,10 @@ mod tests {
/// lockstep, so a silent shed leaves the connection waiting out its full
/// response read-timeout.
///
- /// Driven against a registered in-process client, so the assertion is
that a
- /// reply reached a waiter -- not that a counter moved. With no client on
the
- /// bus every send fails as `ClientNotFound` and a counter bumped before
the
- /// send reports an answer nobody received, which is the failure this test
- /// exists to catch.
+ /// Registers an in-process client, so the assertion is that a reply
reached
+ /// a waiter, not that a counter moved. With no client every send fails
+ /// `ClientNotFound`, and a counter bumped before the send reports an
answer
+ /// nobody received.
#[compio::test]
async fn park_overflow_answers_the_client_instead_of_shedding_silently() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -2898,11 +2938,9 @@ mod tests {
);
}
- /// The counter must credit only denies the bus delivered. It previously
- /// incremented before `send_to_client`, so a shard with no client
registered
- /// still reported the request answered - which blinded
- /// `park_overflow_answers_the_client_instead_of_shedding_silently`, the
one
- /// test that asserts the client hears back.
+ /// Credit only denies the bus delivered. Incrementing before
+ /// `send_to_client` reported an answer with no client registered, blinding
+ /// `park_overflow_answers_the_client_instead_of_shedding_silently`.
#[compio::test]
async fn overflow_deny_is_not_counted_when_the_client_is_gone() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -2939,12 +2977,12 @@ mod tests {
);
}
- /// A parked prepare that ages out has no client to answer, so nothing is
- /// staged and `partition_requests_denied_transient_total` stays put. The
op
- /// is destroyed all the same - the primary retransmits only what has not
- /// reached quorum - so `park_dropped` is the only record it existed.
+ /// The age bound answers requests and steps over prepares. Expiring a
+ /// prepare is permanent loss (no client, and `retransmit_targets` skips
an op
+ /// already at quorum), and passes are commit-driven, so a create burst
+ /// elapses four in milliseconds across every parked namespace at once.
#[compio::test]
- async fn aged_out_prepare_is_counted_even_though_nobody_can_be_answered() {
+ async fn aging_answers_requests_and_never_expires_a_prepare() {
let tmp = TempDir::new().expect("tempdir for system path");
let config = test_config(&tmp);
let mux = TestMux::default();
@@ -2955,11 +2993,50 @@ mod tests {
let ns = IggyNamespace::new(0, 0, 0);
park_one_prepare(&shard, ns, 7).await;
+ park_one_request(&shard, ns).await;
for _ in 0..=PARK_MAX_PASSES {
shard.age_parked_partition_frames(ns);
}
- assert_eq!(shard.parked_frame_count(ns), 0, "the prepare aged out");
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "the request ages out; the prepare stays"
+ );
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 1,
+ "the request must be answered rather than dropped"
+ );
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 1
+ );
+ assert_eq!(
+ park_dropped_count(&shard),
+ 0,
+ "nothing may be destroyed on an age bound"
+ );
+ }
+
+ /// The one path that still destroys a prepare: namespace gone from this
+ /// shard, so holding it buys nothing. No client, so `park_dropped` is the
+ /// only record the op existed.
+ #[compio::test]
+ async fn
a_discarded_prepare_is_counted_even_though_nobody_can_be_answered() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-prepare-discard");
+ seed_topic(&mux, 2, 0, "topic-prepare-discard", vec![assignment(0,
1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ park_one_prepare(&shard, ns, 7).await;
+ shard.discard_parked_partition_frames(ns);
+
+ assert_eq!(shard.parked_frame_count(ns), 0, "the prepare is gone");
assert_eq!(
drain_staged_client_sends(&inbox),
0,
@@ -2977,10 +3054,9 @@ mod tests {
);
}
- /// A frame larger than the per-namespace byte cap used to fail the check
- /// even against an empty entry, so it could never park on any attempt.
For a
- /// replicated prepare that is unrecoverable: `retransmit_targets` skips
an op
- /// that already reached quorum, so the backup stays permanently short of
it.
+ /// A frame larger than the per-namespace cap failed the check even against
+ /// an empty entry, so it could never park. Unrecoverable for a prepare:
+ /// `retransmit_targets` skips an op already at quorum.
#[compio::test]
async fn
a_frame_over_the_namespace_byte_cap_still_parks_into_an_empty_entry() {
const OVER_NAMESPACE_CAP: usize = 5 * 1024 * 1024;
@@ -3105,14 +3181,16 @@ mod tests {
}
/// 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.
+ /// retag, so each entry keeps its whole buffer, up to 64 MiB. With only a
+ /// frame cap one namespace could pin 128 x 64 MiB, so bytes must bite
first.
+ ///
+ /// Single namespace, so this proves the PER-NAMESPACE cap: 1 MiB bodies
+ /// charge 1052672 each, so the 4th crosses 4 MiB, nowhere near the
shard-wide
+ /// 16 MiB that
+ /// [`park_shard_wide_byte_budget_sheds_a_namespace_that_would_cross_it`]
+ /// covers.
#[compio::test]
- async fn park_byte_budget_sheds_large_frames_before_the_frame_cap() {
- const BODY: usize = 1024 * 1024;
-
+ async fn
park_namespace_byte_budget_sheds_large_frames_before_the_frame_cap() {
let tmp = TempDir::new().expect("tempdir for system path");
let config = test_config(&tmp);
let mux = TestMux::default();
@@ -3124,30 +3202,70 @@ mod tests {
for _ in 0..PARK_CAP {
shard
- .on_message(build_partition_request_sized(ns, BODY))
+ .on_message(build_partition_request_sized(ns, MIB_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_eq!(
+ shard.parked_frame_count(ns),
+ PER_NAMESPACE_MIB_FRAMES,
+ "the per-namespace budget admits {PER_NAMESPACE_MIB_FRAMES} x
{MIB_FOOTPRINT} \
+ and sheds the next"
);
- 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)
+ assert_eq!(park_overflow_count(&shard), 1);
+ }
+
+ /// Shard-wide budget on its own terms: fill several namespaces to just
under
+ /// it, then one frame into a fresh namespace. The per-namespace check
waives
+ /// the first frame of an empty entry, so only the shard-wide check can
shed.
+ ///
+ /// Needs its own test because nothing else reaches it: a single namespace
+ /// hits its quarter-sized cap first, leaving `MAX_PARKED_BYTES`
unexercised.
+ #[compio::test]
+ async fn
park_shard_wide_byte_budget_sheds_a_namespace_that_would_cross_it() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-shard-bytes");
+ seed_topic(&mux, 2, 0, "topic-shard-bytes", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let filled = SHARD_BUDGET / (PER_NAMESPACE_MIB_FRAMES * MIB_FOOTPRINT);
+ for topic_id in 0..filled {
+ let ns = IggyNamespace::new(0, topic_id, 0);
+ for _ in 0..PER_NAMESPACE_MIB_FRAMES {
+ shard
+ .on_message(build_partition_request_sized(ns, MIB_BODY))
+ .await;
+ }
+ }
+ assert_eq!(
+ park_overflow_count(&shard),
+ 0,
+ "{filled} namespaces x {PER_NAMESPACE_MIB_FRAMES} frames must all
fit"
);
+
+ // One frame into an empty entry: per-namespace waived, shard-wide not.
+ let crossing = IggyNamespace::new(0, filled, 0);
+ shard
+ .on_message(build_partition_request_sized(crossing, MIB_BODY))
+ .await;
+ assert_eq!(
+ shard.parked_frame_count(crossing),
+ 0,
+ "the frame that would cross the shard-wide budget must be shed"
+ );
+ assert_eq!(park_overflow_count(&shard), 1);
}
- /// A re-dispatch the inbox refuses re-parks the frame, and by then the
- /// namespace is materialised - which closes every other exit. The sweep
- /// skips a namespace in `IggyPartitions`, and `reconcile_additions` stages
- /// no second `InsertOwned` for one, so before the pump-side retry the
frame
- /// sat there until a topic delete: the client never answered, the
shard-wide
- /// byte budget never returned, and the revision fast-skip never re-armed.
+ /// A refused re-dispatch re-parks the frame, and by then the namespace is
+ /// materialised, closing every other exit: the sweep skips a namespace in
+ /// `IggyPartitions` and `reconcile_additions` stages no second
+ /// `InsertOwned`. Before the pump retry the frame sat until a topic
delete,
+ /// unanswered, with its bytes charged and the fast-skip never re-arming.
#[compio::test]
async fn a_re_parked_frame_is_re_dispatched_once_the_inbox_drains() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -3196,9 +3314,9 @@ mod tests {
}
/// A namespace mid-teardown is still in `IggyPartitions`, so it reads as
- /// materialised while the fence forbids serving anything. `ConfirmRemove`
- /// would answer its frames, but a disk delete that keeps failing never
- /// enqueues one, so the sweep must reclaim them itself.
+ /// materialised while the fence forbids serving it. `ConfirmRemove` would
+ /// answer its frames, but a disk delete that keeps failing never enqueues
+ /// one, so the sweep has to.
#[compio::test]
async fn
parked_frames_of_a_tombstoned_namespace_are_reclaimed_without_confirm_remove() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -3236,9 +3354,8 @@ mod tests {
);
}
- /// Residency backstop for the same path: an inbox that never drains must
not
- /// hold a re-parked frame forever. `MAX_PARKED_PASSES` applies to a
- /// materialised namespace too, so the sweep eventually answers it.
+ /// Residency backstop: an inbox that never drains must not hold a
re-parked
+ /// frame forever, so `MAX_PARKED_PASSES` covers a materialised namespace
too.
#[compio::test]
async fn a_re_parked_frame_ages_out_when_the_inbox_never_drains() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -3268,6 +3385,17 @@ mod tests {
/// Mirrors `MAX_PARKED_PER_NAMESPACE` in `shard::park_if_unmaterialised`.
const PARK_CAP: usize = 128;
+ /// Mirrors `MAX_PARKED_BYTES`.
+ const SHARD_BUDGET: usize = 16 * 1024 * 1024;
+ /// Mirrors `MAX_PARKED_BYTES_PER_NAMESPACE`.
+ const NAMESPACE_BUDGET: usize = SHARD_BUDGET / 4;
+ /// Body the byte-budget tests park, and its charged footprint: a parked
+ /// frame keeps its whole `MESSAGE_ALIGN`-granular buffer, so the header
+ /// pushes a 1 MiB body into the next page.
+ const MIB_BODY: usize = 1024 * 1024;
+ const MIB_FOOTPRINT: usize = MIB_BODY + 4096;
+ /// [`MIB_BODY`] frames one namespace admits before its budget refuses
more.
+ const PER_NAMESPACE_MIB_FRAMES: usize = NAMESPACE_BUDGET / MIB_FOOTPRINT;
/// Mirrors `MAX_PARKED_PASSES`.
const PARK_MAX_PASSES: u32 = 3;
/// Top 16 bits are the owning shard, so this resolves to shard 0's
registry
@@ -3309,9 +3437,9 @@ mod tests {
)
}
- /// Register the client id [`build_partition_request`] stamps, with a
waiter
- /// for its request id, so a deny that reaches the bus resolves a oneshot
the
- /// test can await. The guard keeps the slot installed.
+ /// Register the client id [`build_partition_request`] stamps plus a waiter
+ /// for its request id, so a deny reaching the bus resolves a oneshot. The
+ /// guard keeps the slot installed.
fn register_waiting_client(
shard: &TestShard,
) -> (
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 5445f6395..6bca33a19 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1012,33 +1012,26 @@ where
/// 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>>>,
+ pending_partition_frames: RefCell<BTreeMap<IggyNamespace, ParkEntry>>,
- /// Running sum of [`parked_footprint`] over every frame in
- /// [`Self::pending_partition_frames`], maintained at each mutation site.
+ /// Running sum of [`ParkEntry::bytes`], 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.
+ /// Recomputing per arriving frame is O(all parked frames). Footprints
floor
+ /// at [`MESSAGE_ALIGN`], so the budget admits 4096 entries: ~8.4M visits
per
+ /// admission, on the reactor thread inside the map's `borrow_mut`.
parked_partition_bytes: Cell<usize>,
- /// Namespaces whose park entry still holds frames the inbox refused during
- /// [`Self::redispatch_parked_frames`], so the pump re-drives them on its
- /// next iteration.
+ /// Namespaces holding frames [`Self::redispatch_parked_frames`] could not
+ /// re-queue, for the pump to retry.
///
- /// Without this set a re-parked frame has no exit at all. Its namespace is
- /// materialised by then, so the reconciler sweep skips it, and
- /// `reconcile_additions` skips a namespace already in `IggyPartitions`, so
- /// no second `InsertOwned` ever stages another re-dispatch. Only a topic
- /// delete would reach it. The retry runs on the pump, which is the task
- /// that drains the inbox, so a refusal normally clears on the very next
- /// iteration.
+ /// Without it a re-parked frame has no exit: its namespace is materialised
+ /// by then, so the sweep skips it and `reconcile_additions` stages no
second
+ /// `InsertOwned`. Only a topic delete would reach it. The pump drains the
+ /// inbox, so a refusal usually clears on its next iteration.
///
- /// [`BTreeSet`] for the same reason [`Self::pending_partition_frames`] is
a
- /// [`BTreeMap`]: the simulator replays a fixed seed, so iteration order
has
- /// to be a function of the namespaces alone.
+ /// [`BTreeSet`] for the reason [`Self::pending_partition_frames`] is a
+ /// [`BTreeMap`]: fixed-seed simulator replay needs iteration order to be a
+ /// function of the namespaces alone.
reparked_partition_namespaces: RefCell<BTreeSet<IggyNamespace>>,
/// Live ceiling on prepares served per `RequestPrepares` round. Defaults
@@ -1429,27 +1422,34 @@ where
&self.metrics
}
- /// Attach this shard's own inbox sender to a shard built by
- /// [`Self::without_inbox`], which leaves the mesh empty.
+ /// Attach the sender mesh to a shard built by [`Self::without_inbox`],
which
+ /// leaves it 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`.
+ /// `senders[self.id]`, so without a mesh they silently no-op and a test
+ /// asserting on them proves nothing. The caller must keep the paired
+ /// receivers alive; dropping one turns every `try_send` into
`Disconnected`.
+ ///
+ /// Whole mesh, not one sender: consumers index by shard id and
+ /// `forward_metadata_submit` indexes `senders[0]` unconditionally, so a
+ /// one-element vec is correct only for shard 0. `shard_count` tracks it,
as
+ /// in both constructors.
///
/// # 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(),
+ /// If the mesh is not ordered `senders[i].shard_id() == i` or does not
cover
+ /// this shard. Either routes frames to the wrong pump.
+ #[cfg(any(test, feature = "simulator"))]
+ pub fn attach_senders(&mut self, senders: Vec<TaggedSender>) {
+ assert!(
+ (self.id as usize) < senders.len(),
+ "attach_senders: mesh of {} does not cover shard {}",
+ senders.len(),
self.id
);
- self.senders = vec![sender];
+ validate_sender_ordering(&senders).expect("attach_senders: mesh must
be ordered by shard");
+ self.shard_count = u32::try_from(senders.len()).expect("shard count
fits u32");
+ self.senders = senders;
}
/// `None` removes the handler; subsequent ticks drop with a metric bump.
@@ -1543,10 +1543,9 @@ where
/// fires right after a frame was consumed and a slot freed. Only another
/// refusal puts a namespace back, so the set empties itself.
///
- /// The epoch comes from the routing row: `InsertOwned` writes it alongside
- /// the partition, so a materialised namespace always has one. Skipped when
- /// the row is gone or the namespace is fenced -- teardown did both, and
the
- /// reconciler sweep answers the frames.
+ /// Epoch comes from the routing row, which `InsertOwned` writes alongside
+ /// the partition. Skipped when the row is gone or the namespace is fenced;
+ /// teardown does both, and the reconciler sweep retires the frames.
fn retry_reparked_frames(&self) {
let pending: Vec<IggyNamespace> = {
let mut reparked = self.reparked_partition_namespaces.borrow_mut();
@@ -1575,9 +1574,8 @@ where
where
B: MessageBus + 'static,
{
- // Before the staged ops, and outside the empty-queue early return
- // below: a re-parked frame is waiting on inbox capacity, not on
another
- // reconcile op, and a quiet shard stages none.
+ // Ahead of the staged ops and outside their empty-queue early return:
a
+ // re-parked frame waits on inbox capacity, not on a reconcile op.
self.retry_reparked_frames();
let staged: Vec<ReconcileOp<B>> = {
let mut q = self.reconcile_queue.borrow_mut();
@@ -1704,22 +1702,66 @@ enum ParkOutcome<H> {
/// 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.
+ /// Reconciler passes survived. `reconcile_parked_frames` increments it and
+ /// answers CLIENT REQUESTS past [`MAX_PARKED_PASSES`], in units the
+ /// simulator's virtual clock 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.
+ /// Never expires a replicated prepare: no client to answer, and
+ /// `consensus::retransmit_targets` skips an op that already reached
quorum,
+ /// so expiry is silent permanent loss. Byte budgets bound those instead.
+ ///
+ /// Bounds RESIDENCY, not staleness. The SDK replays the identical request
+ /// for the rest of its response timeout, so an absolute-offset
+ /// `StoreConsumerOffset` rewinds the group on the replay anyway. What it
+ /// buys: a buffer that cannot grow without limit, and a client that learns
+ /// the outcome from a reply rather than a timeout.
passes: u32,
message: Message<GenericHeader>,
}
+impl ParkedFrame {
+ fn footprint(&self) -> usize {
+ parked_footprint(self.message.as_slice().len())
+ }
+
+ /// No client on this node: nothing to answer, nothing recovers it.
+ fn is_replicated(&self) -> bool {
+ self.message.header().command != Command2::Request
+ }
+}
+
+/// One namespace's parked frames plus their running footprint.
+///
+/// Carried, not re-summed: `park_if_unmaterialised` reads it per arriving
frame
+/// over an entry up to [`MAX_PARKED_PER_NAMESPACE`] deep, so a rescan makes
+/// admission quadratic in the depth it exists to bound.
+#[derive(Default)]
+struct ParkEntry {
+ frames: Vec<ParkedFrame>,
+ bytes: usize,
+ /// Frames shed since the entry was created. Only the first warns.
+ shed: u64,
+}
+
+impl ParkEntry {
+ fn push(&mut self, frame: ParkedFrame) {
+ self.bytes = self.bytes.saturating_add(frame.footprint());
+ self.frames.push(frame);
+ }
+
+ /// Remove the selected frames, returning them and the footprint freed so
the
+ /// caller can debit the shard-wide total.
+ fn extract(
+ &mut self,
+ predicate: impl FnMut(&mut ParkedFrame) -> bool,
+ ) -> (Vec<ParkedFrame>, usize) {
+ let taken: Vec<ParkedFrame> = self.frames.extract_if(..,
predicate).collect();
+ let freed: usize = taken.iter().map(ParkedFrame::footprint).sum();
+ self.bytes = self.bytes.saturating_sub(freed);
+ (taken, freed)
+ }
+}
+
/// Per-namespace ceiling on parked frames.
const MAX_PARKED_PER_NAMESPACE: usize = 128;
@@ -1742,20 +1784,12 @@ 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.
///
-/// Applied only to a namespace that already holds something. A frame larger
-/// than this on its own would otherwise never park at all -- the check fails
-/// even against an empty entry -- and for a replicated prepare that is silent,
+/// Applied only to an entry that already holds something. Sized against an
+/// empty entry a larger frame could never park at all, and for a prepare that
is
/// unrecoverable loss: `consensus::retransmit_targets` skips an op that
already
-/// reached quorum, so the backup stays permanently short of it. Shipped
-/// `message_bus.max_message_size` is 64 MiB, well past this, so an ordinary
-/// batched append hits it. Admitting the first frame regardless costs the
other
-/// namespaces up to the whole shard budget for one convergence window; losing
a
-/// committed op costs them the replica.
-///
-/// TODO(krishna): [`MAX_PARKED_BYTES`] is still a hard ceiling, so a frame
above
-/// 16 MiB remains unparkable. Closing that needs the budget derived from the
-/// configured `message_bus.max_message_size` rather than a const here, which
-/// puts worst-case park residency at one max-size frame per shard.
+/// reached quorum. Shipped `message_bus.max_message_size` is 64 MiB, so an
+/// ordinary batched append exceeds this. Cost of the waiver is one convergence
+/// window of shard budget; cost of the loss is the replica.
const MAX_PARKED_BYTES_PER_NAMESPACE: usize = MAX_PARKED_BYTES / 4;
/// Resident cost of parking a frame of `len` bytes.
@@ -1951,13 +1985,15 @@ where
false
}
- /// Drop parked frames for a namespace that will never materialise (it was
- /// removed before its `ReconcileOp::InsertOwned`), so the pending entry is
- /// reclaimed instead of leaking until process exit. Parked client requests
- /// are denied with a transient status rather than dropped: the transports
- /// decode replies in lockstep, so silence wedges the connection until the
- /// SDK's response read-timeout.
- fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
+ /// Discard every frame parked under a namespace this shard can never
serve:
+ /// gone from committed metadata, mid-teardown, or not hashing here. Client
+ /// requests get a transient deny, not silence; transports decode replies
in
+ /// lockstep, so silence wedges the connection until the SDK read-timeout.
+ ///
+ /// The one retirement path a prepare still travels. It is retained
+ /// everywhere else (see [`ParkedFrame::passes`]); here the namespace
itself
+ /// is unreachable, so holding it buys nothing.
+ pub fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
// 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.
@@ -1965,55 +2001,66 @@ where
if let Some(frames) = parked
&& !frames.is_empty()
{
- let total = frames.len();
- let mut answered = 0;
- for frame in frames {
- if self.deny_parked_client_request(frame) {
- answered += 1;
- } else {
- self.metrics.record_frame_drop(
- crate::metrics::frame_drop_variant::PARTITION,
- crate::metrics::frame_drop_reason::PARK_DROPPED,
- );
- }
- }
+ let (answered, dropped) = self.retire_parked_frames(frames);
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
answered,
- dropped = total - answered,
- "discarding parked partition frames for removed namespace"
+ dropped,
+ "discarding parked partition frames for an unreachable
namespace"
);
}
}
- /// Remove a namespace's park entry, debiting its bytes from
- /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
- /// so the running total and the pending-retry set cannot drift out of step
- /// with it.
+ /// Remove a namespace's entry, debiting [`Self::parked_partition_bytes`]
and
+ /// disarming the pump retry. Single place an entry leaves the map, so
+ /// neither can drift out of step with it.
fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
self.reparked_partition_namespaces
.borrow_mut()
.remove(&namespace);
- let frames = self
+ let entry = 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)
+ self.parked_partition_bytes.set(
+ self.parked_partition_bytes
+ .get()
+ .saturating_sub(entry.bytes),
+ );
+ Some(entry.frames)
+ }
+
+ /// Answer client requests, destroy the rest, report `(answered, dropped)`.
+ /// A destroyed frame has nobody to reply to, so
+ /// `frame_drops_total{variant=partition,reason=park_dropped}` is the only
+ /// record it existed.
+ fn retire_parked_frames(&self, frames: Vec<ParkedFrame>) -> (usize, usize)
{
+ let mut answered = 0;
+ let mut dropped = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ dropped += 1;
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
+ (answered, dropped)
}
/// 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.
+ ///
+ /// Read from the map, not the byte cell: an empty entry is never left
+ /// behind, but keying convergence off bytes makes that a silent invariant.
#[must_use]
- pub const fn has_parked_partition_frames(&self) -> bool {
- self.parked_partition_bytes.get() > 0
+ pub fn has_parked_partition_frames(&self) -> bool {
+ !self.pending_partition_frames.borrow().is_empty()
}
/// Namespaces currently holding parked frames. The reconciler pairs this
@@ -2054,17 +2101,18 @@ where
/// 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.
+ /// A frame the inbox refuses is re-parked: retained, so not counted as a
+ /// drop. Re-queuing appends, so a pass materialising many namespaces can
+ /// overrun the inbox; staging a deny is futile because it rides the same
+ /// sender with no await in between. The first `Full` ends the loop, since
+ /// the sole consumer of `senders[self.id]` is the pump task running this
+ /// call and no later frame can find a slot the first one could not.
///
- /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
- /// sweep ages a namespace only while it is un-materialised, and by here it
- /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
- /// retry instead, and the sweep's backstop for an inbox that never drains
is
- /// `partition_reconciler::reconcile_parked_frames`, which now ages a
- /// materialised namespace too.
+ /// [`MAX_PARKED_PASSES`] does not bound a re-parked frame: the sweep ages
a
+ /// namespace only while un-materialised, and by here it is materialised.
+ /// [`Self::repark_partition_frames`] arms the pump retry instead;
+ /// `partition_reconciler::reconcile_parked_frames` is the backstop for an
+ /// inbox that never drains.
fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
where
B: MessageBus + 'static,
@@ -2079,7 +2127,9 @@ where
epoch,
"re-dispatching parked partition frames after materialisation"
);
- let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ // Incarnation filter first, independent of the sender: a prior
+ // incarnation is rejected whether or not this shard can re-queue.
+ let mut servable: Vec<ParkedFrame> = Vec::with_capacity(frames.len());
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.
@@ -2087,54 +2137,61 @@ where
&& parked_epoch != epoch
{
self.reject_stale_parked_frame(namespace, epoch, frame);
- continue;
+ } else {
+ servable.push(frame);
}
- let Some(sender) = self.senders.get(self.id as usize) else {
- continue;
- };
+ }
+ let Some(sender) = self.senders.get(self.id as usize) else {
+ self.retire_parked_frames(servable);
+ return;
+ };
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ let mut remaining = servable.into_iter();
+ while let Some(frame) = remaining.next() {
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;
+ unreachable!("try_send returns the frame it was handed");
+ };
+ let refused_frame = ParkedFrame {
+ epoch: parked_epoch,
+ passes,
+ message,
};
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.
+ // Pump gone: re-parking holds the frame until process exit,
and
+ // every later send hits the same dead channel.
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::DISCONNECTED,
+ );
tracing::warn!(
shard = self.id,
namespace_raw = namespace.inner(),
- "re-dispatch of parked partition frame refused: inbox
disconnected"
+ "re-dispatch of parked partition frames refused: inbox
disconnected"
);
- if message.header().command == Command2::Request
- && let Ok(request) =
message.try_into_typed::<RequestHeader>()
- {
- self.stage_transient_deny(request.header());
- }
- continue;
+ let mut stranded = vec![refused_frame];
+ stranded.extend(remaining);
+ self.retire_parked_frames(stranded);
+ return;
}
+ refused_frames.push(refused_frame);
+ refused_frames.extend(remaining);
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
+ count = refused_frames.len(),
passes,
- "re-parking parked partition frame: inbox full"
+ "re-parking parked partition frames: inbox full"
);
- refused_frames.push(ParkedFrame {
- epoch: parked_epoch,
- passes,
- message,
- });
+ break;
}
if !refused_frames.is_empty() {
self.repark_partition_frames(namespace, refused_frames);
@@ -2149,21 +2206,18 @@ where
/// moment ago, and shedding here would answer a frame the inbox merely
/// deferred.
///
- /// Arming [`Self::reparked_partition_namespaces`] is what makes the
deferral
- /// a deferral. Every other exit from the park map is closed for a
- /// materialised namespace: the reconciler sweep only ages one it has not
- /// built, and `reconcile_additions` never stages a second `InsertOwned`
for
- /// one already in `IggyPartitions`.
+ /// Arming [`Self::reparked_partition_namespaces`] is what makes it a
+ /// deferral. Every other exit is closed once materialised: the sweep only
+ /// ages a namespace it has not built, and `reconcile_additions` stages no
+ /// second `InsertOwned` for one already in `IggyPartitions`.
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);
+ let restored: usize = frames.iter().map(ParkedFrame::footprint).sum();
+ let mut pending = self.pending_partition_frames.borrow_mut();
+ let entry = pending.entry(namespace).or_default();
+ for frame in frames {
+ entry.push(frame);
+ }
+ drop(pending);
self.parked_partition_bytes
.set(self.parked_partition_bytes.get().saturating_add(restored));
self.reparked_partition_namespaces
@@ -2171,60 +2225,50 @@ where
.insert(namespace);
}
- /// Age every frame parked under `namespace` by one reconciler pass and
- /// retire the ones that have outlived [`MAX_PARKED_PASSES`]. Returns the
- /// number retired -- client requests answered plus prepares destroyed, the
- /// latter counted under
- /// `frame_drops_total{variant=partition,reason=park_dropped}`.
+ /// Age every frame under `namespace` by one pass, answering CLIENT
REQUESTS
+ /// past [`MAX_PARKED_PASSES`]. Returns the number answered.
+ ///
+ /// Prepares age but never expire. Expiry destroys a committed op with
+ /// nothing to recover it (see [`ParkedFrame::passes`]), and passes are
+ /// commit-driven: a non-empty buffer defeats the reconciler fast-skip, so
a
+ /// create burst elapses four in milliseconds, across every parked
namespace
+ /// rather than the one it concerns. Byte budgets bound them instead. Only
+ /// [`Self::discard_parked_partition_frames`] still destroys a prepare.
///
- /// 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.
+ /// Passes, not wall-clock, so the simulator's virtual clock governs it.
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 {
+ let Some(entry) = pending.get_mut(&namespace) else {
return 0;
};
- for frame in frames.iter_mut() {
- frame.passes += 1;
+ for frame in &mut entry.frames {
+ frame.passes = frame.passes.saturating_add(1);
}
- let expired: Vec<ParkedFrame> = frames
- .extract_if(.., |frame| frame.passes > MAX_PARKED_PASSES)
- .collect();
- if frames.is_empty() {
- pending.remove(&namespace);
+ let (expired, freed) =
+ entry.extract(|frame| !frame.is_replicated() && frame.passes >
MAX_PARKED_PASSES);
+ let emptied = entry.frames.is_empty();
+ drop(pending);
+ if emptied {
+ // Through the shared remover so the pump-retry set is disarmed
+ // with it; the entry is already empty, so this only unhooks
it.
+ self.take_parked_partition_frames(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 {
- let mut answered = 0;
- for frame in expired {
- if self.deny_parked_client_request(frame) {
- answered += 1;
- } else {
- self.metrics.record_frame_drop(
- crate::metrics::frame_drop_variant::PARTITION,
- crate::metrics::frame_drop_reason::PARK_DROPPED,
- );
- }
- }
+ // Never replicated traffic (the predicate excludes it), so this is
+ // a request whose deny the pump refused: destroyed, and counted.
+ let (answered, unanswered) = self.retire_parked_frames(expired);
tracing::warn!(
shard = self.id,
namespace_raw = namespace.inner(),
answered,
- dropped = count - answered,
- "retiring parked partition frames that outlived their
admission window"
+ unanswered,
+ "answering parked partition requests that outlived their
admission window"
);
}
count
@@ -2241,33 +2285,27 @@ where
self.pending_partition_frames
.borrow()
.get(&namespace)
- .map_or(0, Vec::len)
+ .map_or(0, |entry| entry.frames.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);
- }
-
- /// Retire a parked frame that will never be served: a client request gets
a
- /// transient deny so it can re-issue, replicated traffic is destroyed.
- /// Returns `true` when a reply was staged. 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.
+ /// Retire a frame that will never be served: a client request gets a
+ /// transient deny, replicated traffic is destroyed. Returns `true` only
when
+ /// a reply reached the pump.
///
- /// A prepare gets no reply because there is no client on this node to send
- /// one to, but "no reply" must not mean "no record": the primary only
- /// retransmits an op that has not reached quorum, so a destroyed prepare
on
- /// a lagging backup is invisible loss. The `false` return makes callers
- /// count it under
`frame_drops_total{variant=partition,reason=park_dropped}`.
+ /// Callers are synchronous (`apply_reconcile_ops`, the reconciler sweep),
so
+ /// the deny rides the pump's lifecycle path, not an inline bus send. A
shard
+ /// with no sender stages nothing, hence forwarding
+ /// [`Self::stage_transient_deny`]'s verdict rather than assuming success.
+ ///
+ /// No reply must not mean no record: the primary retransmits only what has
+ /// not reached quorum, so a destroyed prepare is invisible loss. The
`false`
+ /// return is what makes callers bump
+ /// `frame_drops_total{variant=partition,reason=park_dropped}`.
fn deny_parked_client_request(&self, frame: ParkedFrame) -> bool {
if frame.message.header().command == Command2::Request
&& let Ok(request) =
frame.message.try_into_typed::<RequestHeader>()
{
- self.stage_transient_deny(request.header());
- return true;
+ return self.stage_transient_deny(request.header());
}
false
}
@@ -2282,15 +2320,37 @@ where
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();
+ // Both directions reject: a frame stamped AHEAD must not be applied
into
+ // the incarnation the staleness teardown is about to erase either.
Only
+ // BEHIND is the anomaly `partition_frames_rejected_stale_total` is
+ // alerted on. Ahead means the recreate committed between the
reconciler
+ // snapshotting `epoch` and the pump applying `InsertOwned`: expected
+ // churn, and counting it there fires the alert on a race by design.
+ let ahead = frame
+ .epoch
+ .is_some_and(|parked_epoch| parked_epoch > materialised_epoch);
+ let replicated = frame.is_replicated();
+ if ahead {
+ self.metrics.record_partition_frame_rejected_ahead();
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ parked_epoch = ?frame.epoch,
+ materialised_epoch,
+ replicated,
+ "rejecting parked partition frame stamped ahead of the
materialised incarnation"
+ );
+ } else {
+ self.metrics.record_partition_frame_rejected_stale();
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ parked_epoch = ?frame.epoch,
+ materialised_epoch,
+ replicated,
+ "rejecting parked partition frame from a prior incarnation"
+ );
+ }
self.deny_parked_client_request(frame);
}
@@ -2341,48 +2401,71 @@ where
.streams()
.created_revision_for_namespace(namespace);
let frame_cost = parked_footprint(message.as_slice().len());
+ let replicated = message.header().command() != Command2::Request;
let mut pending = self.pending_partition_frames.borrow_mut();
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()
- });
- // The per-namespace byte cap is waived for the first frame, or a frame
- // bigger than the cap could never park at all; see
- // [`MAX_PARKED_BYTES_PER_NAMESPACE`]. The shard-wide budget is not
- // waived, so residency is unchanged.
- let over_namespace_budget = parked_len > 0
- && namespace_bytes.saturating_add(frame_cost) >
MAX_PARKED_BYTES_PER_NAMESPACE;
- if parked_len >= MAX_PARKED_PER_NAMESPACE
- || over_namespace_budget
- || parked_bytes.saturating_add(frame_cost) > MAX_PARKED_BYTES
- {
+ // leave an empty entry 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_mut(&namespace);
+ let parked_len = existing.as_ref().map_or(0, |entry|
entry.frames.len());
+ let namespace_bytes = existing.as_ref().map_or(0, |entry| entry.bytes);
+ // A prepare is never shed on a byte budget. No client to answer, and
no
+ // recovery: `consensus::retransmit_targets` skips an op that already
+ // reached quorum and the plane opens a repair session only in
+ // `on_start_view`, so shedding one is permanent loss where shedding a
+ // request costs a retry. A request is refused the moment admitting it
+ // would cross a budget; a prepare only once one is already spent. Caps
+ // prepare residency at one frame of overshoot per budget (worst case
+ // `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead
of
+ // at the budget, and is what makes an oversize frame parkable at all.
+ let namespace_budget_spent = parked_len > 0
+ && if replicated {
+ namespace_bytes >= MAX_PARKED_BYTES_PER_NAMESPACE
+ } else {
+ namespace_bytes.saturating_add(frame_cost) >
MAX_PARKED_BYTES_PER_NAMESPACE
+ };
+ let shard_budget_spent = if replicated {
+ parked_bytes >= MAX_PARKED_BYTES
+ } else {
+ parked_bytes.saturating_add(frame_cost) > MAX_PARKED_BYTES
+ };
+ if parked_len >= MAX_PARKED_PER_NAMESPACE || namespace_budget_spent ||
shard_budget_spent {
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_frames = parked_len,
- namespace_bytes,
- parked_bytes,
- frame_cost,
- "park buffer at capacity; shedding partition frame"
- );
+ // Warn once per namespace on the transition into shedding, then
+ // `debug`. A full buffer is this branch's trigger, not a rate
limit:
+ // every later frame lands here too, and one formatted `warn`
apiece
+ // is enough for the non-blocking appender to shed unrelated lines.
+ // The counter carries the volume.
+ let first_shed = existing.is_none_or(|entry| {
+ let first = entry.shed == 0;
+ entry.shed = entry.shed.saturating_add(1);
+ first
+ });
+ if first_shed {
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ parked_frames = parked_len,
+ namespace_bytes,
+ parked_bytes,
+ frame_cost,
+ replicated,
+ "park buffer at capacity; shedding partition frames"
+ );
+ } else {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ parked_bytes,
+ frame_cost,
+ replicated,
+ "park buffer still at capacity; shedding partition frame"
+ );
+ }
return ParkOutcome::Overflow(message);
}
tracing::debug!(
@@ -2397,6 +2480,7 @@ where
passes: 0,
message: message.into_generic(),
});
+ drop(pending);
self.parked_partition_bytes
.set(parked_bytes.saturating_add(frame_cost));
ParkOutcome::Parked
@@ -2414,9 +2498,9 @@ where
IggyError::TransientNotAccepted.as_code(),
);
// Count only what the bus accepted, matching `stage_transient_deny`
and
- // the contract on `record_partition_request_denied_transient`: a deny
- // the bus refused is a shed frame, not an answered one, and crediting
it
- // here hides exactly the silent-shed case this counter exists to
expose.
+ // `record_partition_request_denied_transient`'s contract: a refused
deny
+ // is a shed frame, and crediting it hides the silent shed this counter
+ // exists to expose.
if let Err(error) = self
.bus
.send_to_client(request_header.client,
reply.into_generic().into_frozen())
@@ -2442,7 +2526,10 @@ where
/// hand the deny to this shard's own pump as a
/// [`LifecycleFrame::ForwardClientSend`], whose handler performs the bus
/// send (same funnel the parked-frame re-dispatch uses).
- fn stage_transient_deny(&self, request_header: &RequestHeader) {
+ ///
+ /// Returns whether the pump took it. A shard with no sender stages
nothing,
+ /// so assuming success logs an answer for a request destroyed unanswered.
+ fn stage_transient_deny(&self, request_header: &RequestHeader) -> bool {
let reply = build_deny_reply_from_request_header(
request_header,
IggyError::TransientNotAccepted.as_code(),
@@ -2452,7 +2539,7 @@ where
msg: reply.into_generic().into_frozen(),
});
let Some(sender) = self.senders.get(self.id as usize) else {
- return;
+ return false;
};
// 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
@@ -2468,9 +2555,10 @@ where
operation = ?request_header.operation,
"dropping transient deny for discarded partition frame: inbox
rejected: {error:?}"
);
- return;
+ return false;
}
self.metrics.record_partition_request_denied_transient();
+ true
}
#[allow(clippy::future_not_send)]
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index b73a99a54..5b8fa3438 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -116,11 +116,11 @@ pub mod frame_drop_variant {
/// `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. `PARK_DROPPED` ticks when a
-/// frame that did park leaves the buffer without being served: it outlived
-/// `MAX_PARKED_PASSES`, or its namespace was torn down. A client request also
-/// bumps `partition_requests_denied_transient_total` there, since it gets a
-/// reply; replicated traffic has nobody to answer, so this counter is the only
-/// record that the op was destroyed.
+/// frame that did park leaves unserved: its namespace became unreachable, or a
+/// request outlived `MAX_PARKED_PASSES` with no pump to take the deny. A
request
+/// also bumps `partition_requests_denied_transient_total` when answered;
+/// replicated traffic has nobody to answer, so this is the only record the op
+/// was destroyed.
pub mod frame_drop_reason {
pub const FULL: &str = "full";
pub const DISCONNECTED: &str = "disconnected";
@@ -131,6 +131,10 @@ pub mod frame_drop_reason {
pub const PARK_DROPPED: &str = "park_dropped";
}
+// Minted in full, so 7 x 7 includes pairs no drop site produces:
+// `park_overflow` and `park_dropped` pair only with `PARTITION`, leaving 12
+// unreachable. Free while nothing scrapes these (module `TODO(hubcio)`); mint
+// per drop site once a registry lands, so the scrape carries no permanent
zeroes.
const VARIANT_COUNT: usize = 7;
const REASON_COUNT: usize = 7;
@@ -183,6 +187,7 @@ pub struct ShardMetrics {
partitions_removed_total: Counter,
partitions_reconcile_failures_total: Counter,
partition_frames_rejected_stale_total: Counter,
+ partition_frames_rejected_ahead_total: Counter,
partition_requests_denied_transient_total: Counter,
}
@@ -214,6 +219,7 @@ impl ShardMetrics {
partitions_removed_total: Counter::default(),
partitions_reconcile_failures_total: Counter::default(),
partition_frames_rejected_stale_total: Counter::default(),
+ partition_frames_rejected_ahead_total: Counter::default(),
partition_requests_denied_transient_total: Counter::default(),
}
}
@@ -270,6 +276,16 @@ impl ShardMetrics {
self.partition_frames_rejected_stale_total.inc();
}
+ /// Bumped when a parked frame carries an epoch AHEAD of the one its
+ /// partition materialised at: the recreate committed between the
reconciler
+ /// snapshotting the epoch for `InsertOwned` and the pump applying it.
Split
+ /// from `partition_frames_rejected_stale_total` so that counter keeps
meaning
+ /// caught correctness anomaly; this direction is an expected race and
would
+ /// fire the alert on ordinary delete + recreate churn. Both still reject.
+ pub fn record_partition_frame_rejected_ahead(&self) {
+ self.partition_frames_rejected_ahead_total.inc();
+ }
+
/// Total frame drops across every `{variant, reason}` pair.
///
/// Simulator assertion hook: a run without injected loss must keep
@@ -337,6 +353,14 @@ impl ShardMetrics {
self.partition_frames_rejected_stale_total.get()
}
+ /// Snapshot of `partition_frames_rejected_ahead_total`. Test/simulator
+ /// accessor.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn partition_frames_rejected_ahead_value(&self) -> u64 {
+ self.partition_frames_rejected_ahead_total.get()
+ }
+
/// Snapshot of one `frame_drops_total{variant, reason}` pair, or 0 when
the
/// pair is not a known label combination.
///
