hubcio commented on code in PR #3786:
URL: https://github.com/apache/iggy/pull/3786#discussion_r3703809464
##########
core/shard/src/lib.rs:
##########
@@ -1609,8 +1685,106 @@ 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.
+///
+/// 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,
+/// 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.
+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;
Review Comment:
`MAX_PARKED_PASSES` also ages out replicated prepares, and that is a new
loss path: on master a parked frame was held until materialisation, so a
lagging backup's prepare was eventually served. a prepare has no client
(`deny_parked_client_request` returns false) and nothing recovers the op
afterwards - `retransmit_targets` skips anything with `ok_quorum_received`, and
the partition plane opens a repair session only in `on_start_view`. the module
TODO covers shed/refused prepares but not aged ones.
passes are commit-driven (`MetadataCommitTick` fires per partition-shaping
commit, and a non-empty park buffer defeats the fast-skip), so 4 passes elapse
in milliseconds during a create burst - and the sweep ages every parked
namespace on each pass while `staged_this_pass` exempts only the one built that
pass, so unrelated commits burn a lagging namespace's budget. worst shape is
replica_count=2: the primary self-acks to quorum alone, so every prepare the
backup ages out is permanently lost and the backup stays behind `commit_max`
until an unrelated view change.
prepares should not be aged at all - the byte budget already bounds
residency.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -510,6 +599,113 @@ async fn reconcile_additions(
}
}
}
+
+ staged
+}
+
+/// 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
Review Comment:
this justification does not hold for the case that actually fires:
`next_backoff(1)` is 1s (`BACKOFF_BASE`, shift 0) - 60s needs 7 consecutive
failures. `record_failure` stamps attempts 0 -> 1 and the sweep's `now`
postdates the stamp, so the reclaim below fires on the very first failed build.
one transient ENOSPC destroys every parked prepare for the namespace while the
rebuild succeeds a second later, and a reclaimed prepare is unrecoverable (no
client to answer, retransmit skips quorum-acked ops).
reclaim only when the backoff actually exceeds the remaining age budget - or
better, never reclaim prepares on local-convergence grounds.
##########
core/shard/src/lib.rs:
##########
@@ -1841,23 +2331,74 @@ 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()
+ });
+ // 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
Review Comment:
the shard-wide check has no empty-buffer waiver, while the per-namespace
check one line up does (`parked_len > 0`). so a partition frame with footprint
over 16 MiB can never park, on any attempt - `message_bus.max_message_size`
ships at 64 MiB and nothing narrows partition frames. for a client request that
is a transient deny plus a retry; for a prepare it is destroyed at the
`Overflow(_) => {}` arm, whose comment says retransmit redelivers -
contradicted by this PR's own TODO in partition_reconciler.rs (retransmit skips
quorum-acked ops; repair only opens in `on_start_view`).
the const's own TODO admits the >16 MiB gap, but shipping it turns a bound
change into committed-data loss on a backup: on master the only cap was 128
frames, so an oversize prepare parked and was served. and the loss door does
not even need an oversize frame - `MAX_PARKED_BYTES_PER_NAMESPACE` sheds the
4th ordinary 1 MiB frame (footprint 1052672 x 4 > 4 MiB) where master allowed
128.
cleanest fix: do not shed prepares on the byte budgets at all - allow a
one-frame overshoot (worst case = budget + one max-size frame, which is what
the TODO proposes anyway). shedding only requests makes the loss class
structurally impossible instead of raising a ceiling.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
+ let mut answered = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- count = frames.len(),
+ answered,
+ dropped = total - answered,
"discarding parked partition frames for removed namespace"
);
- for frame in frames {
- if frame.header().command == Command2::Request
- && let Ok(request) =
frame.try_into_typed::<RequestHeader>()
+ }
+ }
+
+ /// Remove a namespace's park entry, debiting its bytes from
+ /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
+ /// so the running total and the pending-retry set cannot drift out of step
+ /// with it.
+ fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .remove(&namespace);
+ let frames = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)?;
+ let freed: usize = frames
+ .iter()
+ .map(|frame| parked_footprint(frame.message.as_slice().len()))
+ .sum();
+ self.parked_partition_bytes
+ .set(self.parked_partition_bytes.get().saturating_sub(freed));
+ Some(frames)
+ }
+
+ /// Whether any frame is parked. Cheap enough for the reconciler's per-tick
+ /// fast-skip guard: a non-empty buffer means the shard is by definition
not
+ /// converged, so the skip must not fire.
+ #[must_use]
+ pub const fn has_parked_partition_frames(&self) -> bool {
+ self.parked_partition_bytes.get() > 0
+ }
+
+ /// Namespaces currently holding parked frames. The reconciler pairs this
+ /// against committed metadata to find the ones that will never
materialise,
+ /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that
was
+ /// never built is in neither `IggyPartitions` nor the routing table, so
+ /// nothing else names it.
+ #[must_use]
+ pub fn parked_namespaces(&self) -> Vec<IggyNamespace> {
+ self.pending_partition_frames
+ .borrow()
+ .keys()
+ .copied()
+ .collect()
+ }
+
+ /// Re-queue the frames parked for `namespace` now that its partition
exists
+ /// at `epoch`, onto this shard's own inbox so the pump serves them after
the
+ /// current drain.
+ ///
+ /// A frame stamped with a DIFFERENT incarnation never makes it back: the
+ /// namespace is byte-identical across incarnations, so serving it would
land
+ /// a dead topic's write inside the topic that recycled its keys, and the
+ /// downstream fence cannot see it -- that compares the committed revision
+ /// against the routing row, both of which now describe THIS incarnation.
+ ///
+ /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's
+ /// metadata held no committed partition for the namespace when the frame
+ /// arrived, which on a metadata-lagging backup is the ordinary case the
park
+ /// buffer exists to absorb -- the partition primary materialises and
+ /// replicates as soon as its own metadata commits, well before a lagging
+ /// backup applies the same commit. Treating that as "prior incarnation"
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped with no recovery until an unrelated view change.
+ /// The residual is unchanged from before the stamp existed -- a frame
parked
+ /// while the namespace was absent, then recreated under a new incarnation,
+ /// is served against the replacement -- and closing it needs a wire-level
+ /// discriminator (see the `TODO(krishna)` in
+ /// `partition_reconciler`'s module docs), not a `None`-means-stale rule.
+ ///
+ /// A frame the inbox refuses is re-parked, not answered. Re-queuing
appends,
+ /// so a pass that materialises many namespaces at once can overrun the
inbox;
+ /// staging a deny there is futile because the deny rides that same sender
+ /// with no await in between, so nothing can have drained a slot.
+ ///
+ /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
+ /// sweep ages a namespace only while it is un-materialised, and by here it
+ /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
+ /// retry instead, and the sweep's backstop for an inbox that never drains
is
+ /// `partition_reconciler::reconcile_parked_frames`, which now ages a
+ /// materialised namespace too.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
+ where
+ B: MessageBus + 'static,
+ {
+ let Some(frames) = self.take_parked_partition_frames(namespace) else {
+ return;
+ };
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ epoch,
+ "re-dispatching parked partition frames after materialisation"
+ );
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ for frame in frames {
+ // Only a stamp that exists and disagrees is evidence of a prior
+ // incarnation; see this function's docs on why `None` is not.
+ if let Some(parked_epoch) = frame.epoch
+ && parked_epoch != epoch
+ {
+ 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),
Review Comment:
two compounding issues in this loop. the drop is recorded before the `Full`
vs `Disconnected` match, so a frame that gets re-parked (retained, not lost)
still bumps `frame_drops_total{variant=partition,reason=full}` - the counter
metrics.rs documents as the only stranded-backup signal, and the one the
simulator's `assert_no_frame_drops` pins at zero. and the loop keeps calling
`try_send` after the first `Full`, which is provably futile: no await in the
loop, and the only consumer of `senders[self.id]` is the pump task executing
this very call, so nothing can free a slot mid-loop - up to 127 wasted sends
per call, re-driven by `retry_reparked_frames` on every pump iteration.
one edit fixes both: on `Full`, push the frame, `extend` with the rest of
the iterator, `break`; record a drop only on the `Disconnected` arm. (the
unreachable `let ShardFrame::Consensus .. else` destructure and the inline
duplicate of `deny_parked_client_request` on the disconnected path can fold
into the same edit.)
##########
core/shard/src/lib.rs:
##########
@@ -1383,6 +1429,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) {
Review Comment:
`senders = vec![sender]` puts the sender at index 0, but every consumer
indexes `senders[self.id]` - so for any shard id > 0 the re-dispatch and
`stage_transient_deny` silently no-op, which is the exact failure the doc says
this prevents. it also breaks the `senders[i].shard_id() == i` invariant both
ctors validate, `forward_metadata_submit` indexes `senders[0]` directly
(metadata submits would land in shard id's inbox instead of shard 0's), and
`without_inbox` pins `shard_count: 1` which this never updates. latent today
since every caller passes 0, but a future multi-shard test would silently prove
nothing.
sturdier shape: `attach_senders(&mut self, Vec<TaggedSender>)` run through
the existing `validate_sender_ordering`, update `shard_count` to match, and
gate it `#[cfg(any(test, feature = "simulator"))]` like `parked_frame_count`.
##########
core/shard/src/lib.rs:
##########
@@ -1841,23 +2331,74 @@ 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);
Review Comment:
the stamp is not stable across re-entry: a re-dispatched frame still sitting
in the inbox when a delete + recreate completes (`ConfirmRemove` removes and
untombstones in the same arm, then the rebuild lands) re-parks here stamped
with the new incarnation's revision - read fresh from committed metadata - and
with `passes: 0`, so it is then served against the replacement. that is the
exact write the stamp exists to block, and a third door beyond the two the
module docs admit (`epoch: None`, and the wire-level prepare gap). narrow
window (the full delete + recreate has to finish while the frame waits in the
inbox), but worth carrying the original stamp across re-entry and making
`passes` monotonic - or at least softening the module doc's absolute about
stale drains.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -2104,4 +2423,918 @@ mod tests {
"survivor must take over the disconnected member's partitions"
);
}
+
+ /// A namespace deleted before its build finished is named by nothing: it
is
+ /// absent from `IggyPartitions`, so the removals pass sees no owned ghost,
+ /// and absent from `shards_table`, since the owner seeds a row only via
+ /// `InsertOwned`. Neither `ConfirmRemove` nor `RemoveRouted` can therefore
+ /// reach its parked frames, and without the sweep they are held for the
+ /// process lifetime while every waiting client burns its read timeout.
+ ///
+ /// Reclaim is via the age bound, not on sight of the namespace leaving the
+ /// target set: "absent from committed metadata" reads identically for a
+ /// deleted namespace and for one a metadata-lagging replica has not
applied
+ /// yet, so reclaiming on that would destroy live in-flight traffic. The
first
+ /// pass must therefore hold the frames, and a few passes later they are
gone.
+ #[compio::test]
+ async fn parked_frames_are_reclaimed_when_the_namespace_leaves_metadata() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-reclaim");
+ seed_topic(&mux, 2, 0, "topic-reclaim", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // No pass yet, so the namespace is committed but unmaterialised.
+ park_one_request(&shard, ns).await;
+ assert_eq!(
+ shard.parked_namespaces(),
+ vec![ns],
+ "a request for an unmaterialised namespace must park"
+ );
+
+ // Delete before any pass builds it: the reconciler never materialises
+ // it, so nothing drains the entry the normal way.
+ seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0);
+ reconcile_pass(&ctx).await;
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "the first pass must not destroy the frame: absence from the
target set \
+ is also what a not-yet-applied commit looks like"
+ );
+
+ // Every subsequent pass ages it, and the park buffer keeps defeating
the
+ // revision fast-skip until it drains.
+ for _ in 0..=PARK_MAX_PASSES {
+ reconcile_pass(&ctx).await;
+ }
+
+ assert!(
+ shard.parked_namespaces().is_empty(),
+ "frames for a namespace that left metadata must be answered and
reclaimed"
+ );
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 1,
+ "and the waiting client must get a retriable answer"
+ );
+ }
+
+ /// A frame parked before this node's metadata knew the namespace carries
no
+ /// epoch stamp, and `None` must NOT read as "prior incarnation": on a
+ /// metadata-lagging replica it is the ordinary case, since the partition
+ /// primary materialises and replicates as soon as its own metadata
commits.
+ /// Rejecting it destroys live traffic -- silently for a replicated
prepare,
+ /// which has no client to answer -- and the pre-stamp code served it.
+ #[compio::test]
+ async fn unstamped_parked_frame_is_served_not_rejected_as_stale() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-unstamped");
+ seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Topic slab 1 does not exist yet, so there is no committed
+ // `created_revision` to stamp: the frame parks with `epoch: None`.
+ let unknown = IggyNamespace::new(0, 1, 0);
+ park_one_request(&shard, unknown).await;
+ assert_eq!(
+ shard.parked_frame_count(unknown),
+ 1,
+ "a request for a namespace this node has not applied must park"
+ );
+
+ // The commit this node was lagging behind now lands, and the pass
+ // materialises the namespace.
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ assert_eq!(
+ shard.parked_frame_count(unknown),
+ 0,
+ "materialisation must drain the park entry"
+ );
+ let (served, answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 1,
+ "the unstamped frame must be re-dispatched onto the pump, not
rejected"
+ );
+ assert_eq!(
+ answered, 0,
+ "and it must not be answered with a deny instead of served"
+ );
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 0,
+ "an absent stamp is not evidence of a prior incarnation"
+ );
+ }
+
+ /// The shard-wide byte budget is a running total maintained at each
mutation
+ /// site rather than rescanned per arriving frame. A total that fails to
debit
+ /// on drain silently wedges the budget: the shard would shed every
namespace's
+ /// frames while nothing is actually parked. Exercise each way frames
leave --
+ /// re-dispatch on materialisation, the age bound, and reclaim -- and
assert the
+ /// total returns to empty.
+ #[compio::test]
+ async fn park_byte_total_returns_to_zero_on_every_drain_path() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-bytes");
+ seed_topic(&mux, 2, 0, "topic-bytes", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 16);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Drain path 1: materialisation re-dispatches.
+ let late = IggyNamespace::new(0, 1, 0);
+ park_one_request(&shard, late).await;
+ assert!(shard.has_parked_partition_frames());
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-bytes-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "re-dispatch must debit the parked-byte total"
+ );
+
+ // Drain path 2: the age bound answers the frame.
+ let never = IggyNamespace::new(0, 9, 0);
+ park_one_request(&shard, never).await;
+ assert!(shard.has_parked_partition_frames());
+ for _ in 0..=PARK_MAX_PASSES {
+ shard.age_parked_partition_frames(never);
+ }
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "aging out must debit the parked-byte total"
+ );
+
+ // Drain path 3: an explicit reclaim.
+ park_one_request(&shard, never).await;
+ assert!(shard.has_parked_partition_frames());
+ shard.reclaim_parked_partition_frames(never);
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "reclaim must debit the parked-byte total"
+ );
+
+ drop(inbox);
+ }
+
+ /// The replicated-prepare shape, which no other test covers and where both
+ /// park critical are worst: a prepare has no client, so
`deny_parked_frame`
+ /// no-ops on it and anything that discards it loses committed data
silently,
+ /// with no normal-status repair driver to refetch it.
+ ///
+ /// A backup receives the prepare before its own metadata commits (so the
frame
+ /// parks unstamped), then applies the commit and materialises. The
prepare must
+ /// be re-dispatched, not rejected as a prior incarnation.
+ #[compio::test]
+ async fn unstamped_parked_prepare_is_served_after_materialisation() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-prepare");
+ seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // The primary replicates ahead of this node's metadata: topic slab 1
is
+ // not committed here yet, so the prepare parks with `epoch: None`.
+ let lagging = IggyNamespace::new(0, 1, 0);
+ park_one_prepare(&shard, lagging, 7).await;
+ assert_eq!(
+ shard.parked_frame_count(lagging),
+ 1,
+ "a prepare for a namespace this backup has not applied must park"
+ );
+
+ // The metadata commit catches up and the pass materialises the
namespace.
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ let (served, answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 1,
+ "the parked prepare must be re-dispatched; discarding it is silent
\
+ committed-data loss, since a prepare has no client to answer"
+ );
+ assert_eq!(answered, 0, "a prepare has no client deny to send");
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 0,
+ "an unstamped prepare is not a prior incarnation"
+ );
+ }
+
+ /// A parked prepare whose stamp names a DIFFERENT incarnation must still
be
+ /// dropped: applying a dead topic's op into the topic that recycled its
slab
+ /// keys diverges this replica. This is the half of the fence that stays.
+ #[compio::test]
+ async fn stamped_parked_prepare_from_a_prior_incarnation_is_rejected() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-stale");
+ seed_topic(&mux, 2, 0, "topic-first", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // Parked against the FIRST incarnation, so it carries that revision.
+ park_one_prepare(&shard, ns, 7).await;
+ assert_eq!(shard.parked_frame_count(ns), 1);
+
+ // Delete and recreate: same namespace keys, new committed revision.
+ seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0);
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 4,
+ 0,
+ "topic-recreated",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ let (served, _answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 0,
+ "a prepare stamped with the dead incarnation must not be served
against \
+ its replacement"
+ );
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 1,
+ "and the reject must be counted"
+ );
+ }
+
+ /// Parking does not bump `Streams::revision` and does not wake the
reconciler,
+ /// so a frame that parks in a converged steady state would be held for the
+ /// process lifetime if the revision fast-skip could still fire. A
non-empty
+ /// park buffer must therefore defeat the skip.
+ #[compio::test]
+ async fn park_buffer_defeats_the_revision_fast_skip() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-skip");
+ seed_topic(&mux, 2, 0, "topic-skip", vec![assignment(0, 1)]);
+
+ let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Converge: the first pass materialises, the verify pass after it arms
+ // the skip (same sequence as
`reconcile_fast_skips_when_revision_unchanged`).
+ assert!(reconcile_once(&ctx).await, "first pass runs the full diff");
+ ctx.shard.apply_reconcile_ops();
+ assert!(
+ reconcile_once(&ctx).await,
+ "the verify pass after a working pass must still run"
+ );
+ ctx.shard.apply_reconcile_ops();
+ assert!(
+ !reconcile_once(&ctx).await,
+ "a converged pass with an unchanged revision must fast-skip"
+ );
+
+ // Park a frame for a namespace that is NOT materialised, without
touching
+ // the revision, and the skip must stop firing.
+ let unbuilt = IggyNamespace::new(0, 0, 7);
+ park_one_request(&shard, unbuilt).await;
+ assert!(
+ shard.has_parked_partition_frames(),
+ "the frame must be parked for this test to mean anything"
+ );
+ assert!(
+ reconcile_once(&ctx).await,
+ "a non-empty park buffer must defeat the fast-skip so the sweep
can run"
+ );
+ }
+
+ /// Same hole, reached the other way: the namespace stays committed but
+ /// `build_partition_fresh` keeps failing. The `FailureCause::Add` backoff
+ /// clamps at 60s, twice the client's read timeout, so holding the frames
+ /// cannot help - answer them and let the client re-issue.
+ #[compio::test]
+ async fn parked_frames_are_reclaimed_while_the_build_is_backed_off() {
Review Comment:
this test pins the first-failure reclaim as correct (see the comment on the
sweep - `next_backoff(1)` is 1s, not the 60s the doc claims), and it uses the
inbox-less `build_test_shard`, so it cannot observe whether the client was ever
answered: `stage_transient_deny` bails at the missing sender while
`deny_parked_client_request` still returns true. the sibling test
`parked_frames_are_answered_once_they_outlive_their_admission_window` spells
out this exact trap and uses `build_test_shard_with_inbox` +
`drain_staged_client_sends`. also nothing in this branch covers a parked
prepare.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
+ let mut answered = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- count = frames.len(),
+ answered,
+ dropped = total - answered,
"discarding parked partition frames for removed namespace"
);
- for frame in frames {
- if frame.header().command == Command2::Request
- && let Ok(request) =
frame.try_into_typed::<RequestHeader>()
+ }
+ }
+
+ /// Remove a namespace's park entry, debiting its bytes from
+ /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
+ /// so the running total and the pending-retry set cannot drift out of step
+ /// with it.
+ fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .remove(&namespace);
+ let frames = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)?;
+ let freed: usize = frames
+ .iter()
+ .map(|frame| parked_footprint(frame.message.as_slice().len()))
+ .sum();
+ self.parked_partition_bytes
+ .set(self.parked_partition_bytes.get().saturating_sub(freed));
+ Some(frames)
+ }
+
+ /// Whether any frame is parked. Cheap enough for the reconciler's per-tick
+ /// fast-skip guard: a non-empty buffer means the shard is by definition
not
+ /// converged, so the skip must not fire.
+ #[must_use]
+ pub const fn has_parked_partition_frames(&self) -> bool {
+ self.parked_partition_bytes.get() > 0
+ }
+
+ /// Namespaces currently holding parked frames. The reconciler pairs this
+ /// against committed metadata to find the ones that will never
materialise,
+ /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that
was
+ /// never built is in neither `IggyPartitions` nor the routing table, so
+ /// nothing else names it.
+ #[must_use]
+ pub fn parked_namespaces(&self) -> Vec<IggyNamespace> {
+ self.pending_partition_frames
+ .borrow()
+ .keys()
+ .copied()
+ .collect()
+ }
+
+ /// Re-queue the frames parked for `namespace` now that its partition
exists
+ /// at `epoch`, onto this shard's own inbox so the pump serves them after
the
+ /// current drain.
+ ///
+ /// A frame stamped with a DIFFERENT incarnation never makes it back: the
+ /// namespace is byte-identical across incarnations, so serving it would
land
+ /// a dead topic's write inside the topic that recycled its keys, and the
+ /// downstream fence cannot see it -- that compares the committed revision
+ /// against the routing row, both of which now describe THIS incarnation.
+ ///
+ /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's
+ /// metadata held no committed partition for the namespace when the frame
+ /// arrived, which on a metadata-lagging backup is the ordinary case the
park
+ /// buffer exists to absorb -- the partition primary materialises and
+ /// replicates as soon as its own metadata commits, well before a lagging
+ /// backup applies the same commit. Treating that as "prior incarnation"
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped with no recovery until an unrelated view change.
+ /// The residual is unchanged from before the stamp existed -- a frame
parked
+ /// while the namespace was absent, then recreated under a new incarnation,
+ /// is served against the replacement -- and closing it needs a wire-level
+ /// discriminator (see the `TODO(krishna)` in
+ /// `partition_reconciler`'s module docs), not a `None`-means-stale rule.
+ ///
+ /// A frame the inbox refuses is re-parked, not answered. Re-queuing
appends,
+ /// so a pass that materialises many namespaces at once can overrun the
inbox;
+ /// staging a deny there is futile because the deny rides that same sender
+ /// with no await in between, so nothing can have drained a slot.
+ ///
+ /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
+ /// sweep ages a namespace only while it is un-materialised, and by here it
+ /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
+ /// retry instead, and the sweep's backstop for an inbox that never drains
is
+ /// `partition_reconciler::reconcile_parked_frames`, which now ages a
+ /// materialised namespace too.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
+ where
+ B: MessageBus + 'static,
+ {
+ let Some(frames) = self.take_parked_partition_frames(namespace) else {
+ return;
+ };
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ epoch,
+ "re-dispatching parked partition frames after materialisation"
+ );
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ for frame in frames {
+ // Only a stamp that exists and disagrees is evidence of a prior
+ // incarnation; see this function's docs on why `None` is not.
+ if let Some(parked_epoch) = frame.epoch
+ && parked_epoch != epoch
Review Comment:
fence is direction-blind: park stamps the live committed revision, but
`InsertOwned` carries the epoch snapshotted at build time, so a delete +
recreate committing between staging and the pump apply gives the parked frame a
stamp newer than `epoch` - and it gets rejected with the "prior incarnation"
warn and counted on `partition_frames_rejected_stale_total`, whose doc says
non-zero is a caught correctness anomaly, not routine churn. false positive for
anyone alerting on it.
note the tempting one-liner `parked_epoch < epoch` is worse: a newer-stamped
prepare would then be applied into the old incarnation's partition, which the
staleness teardown erases along with its disk data. keep rejecting both
directions - just split the log line and the label by direction so the anomaly
counter keeps its contract.
##########
core/integration/tests/cluster/multi_shard_partition_convergence.rs:
##########
@@ -0,0 +1,201 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Partition convergence across shards on one node.
+//!
+//! `create_topic` returns on metadata commit, before the owning shard has run
+//! `build_partition_fresh`, so a produce issued immediately after races
+//! materialisation. Nothing about that race is resolved on the shard the
client
+//! is connected to: `shards_table` is a cache of the deterministic hash
+//! assignment and may hold a row before the partition exists. The owning shard
+//! resolves it, by parking the frame until its partition lands
+//! (`park_if_unmaterialised`) and by fencing a mismatched incarnation
+//! (`serves_committed_incarnation`), answering anything it cannot yet serve
+//! with a retriable status the SDK replays.
+//!
+//! Forcing two shards is the point. Existing coverage of these fences is
+//! single-shard, where the shard that admits the request is also the one that
+//! owns the partition, so the cross-core path never runs: the hash fallback in
+//! `router::route_typed`, the frame crossing into a peer's inbox, and that
+//! peer's park queue draining on its own pump. Roughly half of the topics
below
+//! hash to a shard other than the one homing the connection.
+//!
+//! That split is a murmur3 outcome and is invisible from here -- this test
would
+//! stay green while silently degrading to single-shard if the hash or the
shard
+//! count changed. It is pinned instead where the assignment is a pure
function,
+//! by
`partition_reconciler::tests::integration_topic_set_straddles_both_shards`,
+//! which asserts over the same namespaces this test creates. Change the topic
+//! count or the stream here and that guard has to move with it.
+//!
+//! Scope, stated plainly: this is a convergence test, not a barrier test. It
+//! cannot tell a request served straight through from one that parked and was
+//! re-dispatched, so it does not pin *which* mechanism carried it. What it
does
+//! catch is that mechanism failing outright -- a frame that never reaches the
+//! owner, a park queue that never drains, or a fence that denies forever --
+//! since every one of those surfaces as a failed send or a short poll.
+
+#![cfg(feature = "vsr")]
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+
+const STREAM: &str = "convergence-stream";
+const PARTITION_ID: u32 = 0;
+/// Enough topics that the murmur3 assignment lands on both shards; every one
is
+/// asserted, so which side each falls on does not matter.
+const TOPICS: u32 = 8;
+
+fn topic_name(index: u32) -> String {
+ format!("convergence-topic-{index}")
+}
+
+async fn create_topic(client: &IggyClient, stream: &Identifier, name: &str) {
+ client
+ .create_topic(
+ stream,
+ name,
+ 1,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap_or_else(|error| panic!("create_topic {name}: {error}"));
+}
+
+async fn produce(client: &IggyClient, stream: &Identifier, topic: &Identifier,
payload: &str) {
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload(payload.to_owned().into())
+ .build()
+ .expect("message build"),
+ ];
+ client
+ .send_messages(
+ stream,
+ topic,
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .unwrap_or_else(|error| panic!("send_messages {payload}: {error}"));
+}
+
+async fn poll_payloads(
+ client: &IggyClient,
+ stream: &Identifier,
+ topic: &Identifier,
+) -> Vec<String> {
+ client
+ .poll_messages(
+ stream,
+ topic,
+ Some(PARTITION_ID),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 16,
+ false,
+ )
+ .await
+ .unwrap_or_else(|error| panic!("poll_messages: {error}"))
+ .messages
+ .iter()
+ .map(|message| String::from_utf8_lossy(&message.payload).into_owned())
+ .collect()
+}
+
+/// Topics are created in a batch first, so several materialisations are in
+/// flight at once when the produces start.
+#[iggy_harness(cluster_nodes = 1, server(system.sharding.cpu_allocation =
"2"))]
+async fn
given_two_shards_when_producing_right_after_create_topic_should_round_trip(
+ harness: &TestHarness,
+) {
+ let client = harness.new_client().await.unwrap();
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .unwrap();
+ let stream = Identifier::named(STREAM).unwrap();
+ client.create_stream(STREAM).await.unwrap();
+
+ for index in 0..TOPICS {
+ create_topic(&client, &stream, &topic_name(index)).await;
+ }
+ for index in 0..TOPICS {
+ let topic = Identifier::named(&topic_name(index)).unwrap();
+ produce(&client, &stream, &topic, &format!("payload-{index}")).await;
+ }
+ for index in 0..TOPICS {
+ let topic = Identifier::named(&topic_name(index)).unwrap();
+ assert_eq!(
+ poll_payloads(&client, &stream, &topic).await,
+ vec![format!("payload-{index}")],
+ "topic {index} must return the message produced right after its
creation"
+ );
+ }
+}
+
+/// Delete + recreate reuses the freed slab keys, so the namespace is
+/// byte-identical across incarnations and only `created_revision` separates
+/// them. This pins the observable outcome across that transition on a
+/// multi-shard node: the recreated topic serves its own data and none of the
+/// dead incarnation's.
+///
+/// It does NOT exercise the incarnation fence. Each step here is a completed
+/// round trip, so the reconciler converges before the next one starts and
+/// `serves_committed_incarnation` never denies (verified: zero "unverified
Review Comment:
this parenthetical cannot have verified anything: the unverified-incarnation
deny logs at `debug!` and the harness runs the server at the default info
level, so the line cannot appear in the captured log regardless of whether
denials happened.
the harness already has `ServerHandle::stdout_occurrences`, and all three
degraded-park modes log at `warn` ("park buffer at capacity", "retiring parked
partition frames", "rejecting parked partition frame from a prior incarnation")
- asserting those are zero would pin the mechanism positively, and is the
cheapest regression net for the park paths given the assertions here cannot
tell park + redispatch apart from deny + SDK replay.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -510,6 +599,113 @@ async fn reconcile_additions(
}
}
}
+
+ staged
+}
+
+/// 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`).
+///
+/// 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.
Review Comment:
understates what the exemption does: `reconcile_additions` has no cross-pass
guard against a queued-but-unapplied `InsertOwned` (line 475 tests
`partitions.contains`, false while the op is queued), so it re-stages the
namespace on every pass until the pump drains - the set protects parked frames
across arbitrary pump lag, not one pass. worth saying outright: the one-pass
framing here is what makes removing this set look safe, and removing it would
age frames on every commit-driven pass while the pump lags.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
Review Comment:
this retire loop (deny -> answered, else `park_dropped`) is duplicated
verbatim in `age_parked_partition_frames`. one helper returning (answered,
dropped) would also single-site the accounting fix needed for
`deny_parked_client_request`'s return value.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -969,6 +1171,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> {
Review Comment:
byte-identical to `build_partition_request_sized(namespace, 0)` - delegate.
##########
core/server-ng/Cargo.toml:
##########
@@ -184,6 +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"] }
Review Comment:
workspace already has a `shard` entry - `{ workspace = true, features =
["simulator"] }` like the sibling dev-deps. also the comment's mechanism is
slightly off: dev-dependencies are always resolved (they are in the lockfile);
what the resolver guarantees is that their feature selections are not unified
into non-test targets.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
+ let mut answered = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- count = frames.len(),
+ answered,
+ dropped = total - answered,
"discarding parked partition frames for removed namespace"
);
- for frame in frames {
- if frame.header().command == Command2::Request
- && let Ok(request) =
frame.try_into_typed::<RequestHeader>()
+ }
+ }
+
+ /// Remove a namespace's park entry, debiting its bytes from
+ /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
+ /// so the running total and the pending-retry set cannot drift out of step
+ /// with it.
+ fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .remove(&namespace);
+ let frames = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)?;
+ let freed: usize = frames
+ .iter()
+ .map(|frame| parked_footprint(frame.message.as_slice().len()))
+ .sum();
+ self.parked_partition_bytes
+ .set(self.parked_partition_bytes.get().saturating_sub(freed));
+ Some(frames)
+ }
+
+ /// Whether any frame is parked. Cheap enough for the reconciler's per-tick
+ /// fast-skip guard: a non-empty buffer means the shard is by definition
not
+ /// converged, so the skip must not fire.
+ #[must_use]
+ pub const fn has_parked_partition_frames(&self) -> bool {
+ self.parked_partition_bytes.get() > 0
+ }
+
+ /// Namespaces currently holding parked frames. The reconciler pairs this
+ /// against committed metadata to find the ones that will never
materialise,
+ /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that
was
+ /// never built is in neither `IggyPartitions` nor the routing table, so
+ /// nothing else names it.
+ #[must_use]
+ pub fn parked_namespaces(&self) -> Vec<IggyNamespace> {
+ self.pending_partition_frames
+ .borrow()
+ .keys()
+ .copied()
+ .collect()
+ }
+
+ /// Re-queue the frames parked for `namespace` now that its partition
exists
+ /// at `epoch`, onto this shard's own inbox so the pump serves them after
the
+ /// current drain.
+ ///
+ /// A frame stamped with a DIFFERENT incarnation never makes it back: the
+ /// namespace is byte-identical across incarnations, so serving it would
land
+ /// a dead topic's write inside the topic that recycled its keys, and the
+ /// downstream fence cannot see it -- that compares the committed revision
+ /// against the routing row, both of which now describe THIS incarnation.
+ ///
+ /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's
+ /// metadata held no committed partition for the namespace when the frame
+ /// arrived, which on a metadata-lagging backup is the ordinary case the
park
+ /// buffer exists to absorb -- the partition primary materialises and
+ /// replicates as soon as its own metadata commits, well before a lagging
+ /// backup applies the same commit. Treating that as "prior incarnation"
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped with no recovery until an unrelated view change.
+ /// The residual is unchanged from before the stamp existed -- a frame
parked
+ /// while the namespace was absent, then recreated under a new incarnation,
+ /// is served against the replacement -- and closing it needs a wire-level
+ /// discriminator (see the `TODO(krishna)` in
+ /// `partition_reconciler`'s module docs), not a `None`-means-stale rule.
+ ///
+ /// A frame the inbox refuses is re-parked, not answered. Re-queuing
appends,
+ /// so a pass that materialises many namespaces at once can overrun the
inbox;
+ /// staging a deny there is futile because the deny rides that same sender
+ /// with no await in between, so nothing can have drained a slot.
+ ///
+ /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
+ /// sweep ages a namespace only while it is un-materialised, and by here it
+ /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
+ /// retry instead, and the sweep's backstop for an inbox that never drains
is
+ /// `partition_reconciler::reconcile_parked_frames`, which now ages a
+ /// materialised namespace too.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
+ where
+ B: MessageBus + 'static,
+ {
+ let Some(frames) = self.take_parked_partition_frames(namespace) else {
+ return;
+ };
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ epoch,
+ "re-dispatching parked partition frames after materialisation"
+ );
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ for frame in frames {
+ // Only a stamp that exists and disagrees is evidence of a prior
+ // incarnation; see this function's docs on why `None` is not.
+ if let Some(parked_epoch) = frame.epoch
+ && parked_epoch != epoch
+ {
+ 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 and arming the pump-side
retry.
+ ///
+ /// 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.
+ ///
+ /// 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`.
+ 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));
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .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}`.
+ ///
+ /// 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 {
+ 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,
+ );
+ }
}
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ answered,
+ dropped = count - answered,
+ "retiring parked partition frames that outlived their
admission window"
+ );
}
+ 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) {
Review Comment:
pure alias of `discard_parked_partition_frames` - make that one pub and
delete this; the park vocabulary is already at park / shed / repark /
redispatch / age / retire / discard / reclaim.
##########
core/shard/src/lib.rs:
##########
@@ -995,9 +996,50 @@ where
/// materialised the namespace (post-`CreateTopic` convergence window).
/// Parked here instead of dropped -- there is no consensus retransmit
/// driver in production yet -- and re-dispatched when the matching
- /// `ReconcileOp::InsertOwned` lands. Bounded per namespace; overflow
- /// drops the frame (at-least-once: client/primary retries recover).
- pending_partition_frames: RefCell<HashMap<IggyNamespace,
Vec<Message<GenericHeader>>>>,
+ /// `ReconcileOp::InsertOwned` lands with the epoch they were stamped
+ /// against. Bounded per namespace; a full buffer sheds via
+ /// [`ParkOutcome::Overflow`] so the caller can still answer.
+ ///
+ /// An entry only drains when the namespace materialises or leaves
committed
+ /// metadata, so the reconciler reclaims the ones that will do neither --
see
+ /// `partition_reconciler::reconcile_parked_frames`. Without that sweep a
+ /// namespace whose build keeps failing would hold its frames for the
process
+ /// lifetime while every client waited out its read timeout.
+ ///
+ /// [`BTreeMap`], not `HashMap`: [`Self::parked_namespaces`] feeds the
+ /// reconciler sweep, which answers frames in the order it walks them.
+ /// `std::collections::HashMap` seeds its hasher per process, so iteration
+ /// order would vary run to run for identical committed state, making the
+ /// simulator's deny ordering unreproducible for a fixed seed -- the same
+ /// hazard `router.rs` documents as its reason for `select_biased!`.
+ pending_partition_frames: RefCell<BTreeMap<IggyNamespace,
Vec<ParkedFrame>>>,
+
+ /// Running sum of [`parked_footprint`] over every frame in
+ /// [`Self::pending_partition_frames`], maintained at each mutation site.
+ ///
+ /// Recomputing it per arriving frame is O(all parked frames): the budget
+ /// admits ~262k entries at 256 bytes each, so filling the buffer would
cost
Review Comment:
math is off here: the budget is charged on `parked_footprint`, which floors
at 4096, so the ceiling is 16 MiB / 4 KiB = 4096 entries and the worst-case
fill is around 8.4M visits - not 262k entries / 10^10. (262144 is 64 MiB / 256,
looks keyed off `MAX_MESSAGE_SIZE` instead of `MAX_PARKED_BYTES`; even on
logical bytes 16 MiB / 256 is 65536.) the conclusion stands, the running total
is the right call.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -280,6 +344,9 @@ struct PassCounters {
/// blocked by a consumer barrier or by a rejoin whose offsets land via
/// journal repair, and neither unblocking bumps `Streams::revision`.
trims_pending: usize,
+ /// Namespaces whose parked frames were answered because this shard is not
Review Comment:
doc matches 2 of the 4 bump sites - the aged branches below also count
namespaces that are materialised or still building, not just ones this shard
"is not going to materialise". and "answered" is wrong for prepares, which are
destroyed with no reply; that answered/destroyed conflation makes the loss
paths easy to misread.
##########
core/shard/src/metrics.rs:
##########
@@ -94,16 +113,26 @@ 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. `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.
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";
+ pub const PARK_DROPPED: &str = "park_dropped";
}
const VARIANT_COUNT: usize = 7;
-const REASON_COUNT: usize = 5;
+const REASON_COUNT: usize = 7;
Review Comment:
5 -> 7 mints the full 7x7 counter cross product per shard; `park_overflow` /
`park_dropped` are only ever bumped with `variant=partition`, so 12 pairs are
unreachable. free today with no registry - noting for when scraping lands.
##########
core/shard/src/lib.rs:
##########
@@ -1841,23 +2331,74 @@ 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()
+ });
+ // 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
+ {
+ 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
Review Comment:
"rate-limited by the buffer being full" is not rate limiting - the full
buffer is the trigger, and this emits one formatted warn per arriving frame
while the map borrow is held. with the per-namespace cap biting at 4 ordinary 1
MiB frames the storm is easy to reach, and the non-blocking appender is lossy
under pressure, so it can drop unrelated lines. log the empty -> full
transition once and let the counter carry the rest.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
+ let mut answered = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- count = frames.len(),
+ answered,
+ dropped = total - answered,
"discarding parked partition frames for removed namespace"
);
- for frame in frames {
- if frame.header().command == Command2::Request
- && let Ok(request) =
frame.try_into_typed::<RequestHeader>()
+ }
+ }
+
+ /// Remove a namespace's park entry, debiting its bytes from
+ /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
+ /// so the running total and the pending-retry set cannot drift out of step
+ /// with it.
+ fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .remove(&namespace);
+ let frames = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)?;
+ let freed: usize = frames
+ .iter()
+ .map(|frame| parked_footprint(frame.message.as_slice().len()))
+ .sum();
+ self.parked_partition_bytes
+ .set(self.parked_partition_bytes.get().saturating_sub(freed));
+ Some(frames)
+ }
+
+ /// Whether any frame is parked. Cheap enough for the reconciler's per-tick
+ /// fast-skip guard: a non-empty buffer means the shard is by definition
not
+ /// converged, so the skip must not fire.
+ #[must_use]
+ pub const fn has_parked_partition_frames(&self) -> bool {
+ self.parked_partition_bytes.get() > 0
+ }
+
+ /// Namespaces currently holding parked frames. The reconciler pairs this
+ /// against committed metadata to find the ones that will never
materialise,
+ /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that
was
+ /// never built is in neither `IggyPartitions` nor the routing table, so
+ /// nothing else names it.
+ #[must_use]
+ pub fn parked_namespaces(&self) -> Vec<IggyNamespace> {
+ self.pending_partition_frames
+ .borrow()
+ .keys()
+ .copied()
+ .collect()
+ }
+
+ /// Re-queue the frames parked for `namespace` now that its partition
exists
+ /// at `epoch`, onto this shard's own inbox so the pump serves them after
the
+ /// current drain.
+ ///
+ /// A frame stamped with a DIFFERENT incarnation never makes it back: the
+ /// namespace is byte-identical across incarnations, so serving it would
land
+ /// a dead topic's write inside the topic that recycled its keys, and the
+ /// downstream fence cannot see it -- that compares the committed revision
+ /// against the routing row, both of which now describe THIS incarnation.
+ ///
+ /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's
+ /// metadata held no committed partition for the namespace when the frame
+ /// arrived, which on a metadata-lagging backup is the ordinary case the
park
+ /// buffer exists to absorb -- the partition primary materialises and
+ /// replicates as soon as its own metadata commits, well before a lagging
+ /// backup applies the same commit. Treating that as "prior incarnation"
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped with no recovery until an unrelated view change.
+ /// The residual is unchanged from before the stamp existed -- a frame
parked
+ /// while the namespace was absent, then recreated under a new incarnation,
+ /// is served against the replacement -- and closing it needs a wire-level
+ /// discriminator (see the `TODO(krishna)` in
+ /// `partition_reconciler`'s module docs), not a `None`-means-stale rule.
+ ///
+ /// A frame the inbox refuses is re-parked, not answered. Re-queuing
appends,
+ /// so a pass that materialises many namespaces at once can overrun the
inbox;
+ /// staging a deny there is futile because the deny rides that same sender
+ /// with no await in between, so nothing can have drained a slot.
+ ///
+ /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
+ /// sweep ages a namespace only while it is un-materialised, and by here it
+ /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
+ /// retry instead, and the sweep's backstop for an inbox that never drains
is
+ /// `partition_reconciler::reconcile_parked_frames`, which now ages a
+ /// materialised namespace too.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
+ where
+ B: MessageBus + 'static,
+ {
+ let Some(frames) = self.take_parked_partition_frames(namespace) else {
+ return;
+ };
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ epoch,
+ "re-dispatching parked partition frames after materialisation"
+ );
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ for frame in frames {
+ // Only a stamp that exists and disagrees is evidence of a prior
+ // incarnation; see this function's docs on why `None` is not.
+ if let Some(parked_epoch) = frame.epoch
+ && parked_epoch != epoch
+ {
+ 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 and arming the pump-side
retry.
+ ///
+ /// 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.
+ ///
+ /// 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`.
+ 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));
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .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}`.
+ ///
+ /// 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 {
+ 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,
+ );
+ }
}
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ answered,
+ dropped = count - answered,
+ "retiring parked partition frames that outlived their
admission window"
+ );
}
+ 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);
+ }
+
+ /// 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.
+ ///
+ /// 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}`.
+ fn deny_parked_client_request(&self, frame: ParkedFrame) -> bool {
Review Comment:
returns true for any request even when nothing was staged:
`stage_transient_deny` returns `()` and silently bails when `senders[self.id]`
is missing, so callers log the frame as answered and skip the `park_dropped`
bump for a request that was destroyed unanswered - while the doc above says the
false return is exactly what makes callers count it. same count-before-delivery
pattern this PR carefully fixes in `deny_partition_request_transient` and in
`stage_transient_deny`'s own counter. make `stage_transient_deny` return bool
and propagate it.
##########
core/server-ng/src/partition_reconciler.rs:
##########
@@ -2104,4 +2423,918 @@ mod tests {
"survivor must take over the disconnected member's partitions"
);
}
+
+ /// A namespace deleted before its build finished is named by nothing: it
is
+ /// absent from `IggyPartitions`, so the removals pass sees no owned ghost,
+ /// and absent from `shards_table`, since the owner seeds a row only via
+ /// `InsertOwned`. Neither `ConfirmRemove` nor `RemoveRouted` can therefore
+ /// reach its parked frames, and without the sweep they are held for the
+ /// process lifetime while every waiting client burns its read timeout.
+ ///
+ /// Reclaim is via the age bound, not on sight of the namespace leaving the
+ /// target set: "absent from committed metadata" reads identically for a
+ /// deleted namespace and for one a metadata-lagging replica has not
applied
+ /// yet, so reclaiming on that would destroy live in-flight traffic. The
first
+ /// pass must therefore hold the frames, and a few passes later they are
gone.
+ #[compio::test]
+ async fn parked_frames_are_reclaimed_when_the_namespace_leaves_metadata() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-reclaim");
+ seed_topic(&mux, 2, 0, "topic-reclaim", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // No pass yet, so the namespace is committed but unmaterialised.
+ park_one_request(&shard, ns).await;
+ assert_eq!(
+ shard.parked_namespaces(),
+ vec![ns],
+ "a request for an unmaterialised namespace must park"
+ );
+
+ // Delete before any pass builds it: the reconciler never materialises
+ // it, so nothing drains the entry the normal way.
+ seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0);
+ reconcile_pass(&ctx).await;
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "the first pass must not destroy the frame: absence from the
target set \
+ is also what a not-yet-applied commit looks like"
+ );
+
+ // Every subsequent pass ages it, and the park buffer keeps defeating
the
+ // revision fast-skip until it drains.
+ for _ in 0..=PARK_MAX_PASSES {
+ reconcile_pass(&ctx).await;
+ }
+
+ assert!(
+ shard.parked_namespaces().is_empty(),
+ "frames for a namespace that left metadata must be answered and
reclaimed"
+ );
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 1,
+ "and the waiting client must get a retriable answer"
+ );
+ }
+
+ /// A frame parked before this node's metadata knew the namespace carries
no
+ /// epoch stamp, and `None` must NOT read as "prior incarnation": on a
+ /// metadata-lagging replica it is the ordinary case, since the partition
+ /// primary materialises and replicates as soon as its own metadata
commits.
+ /// Rejecting it destroys live traffic -- silently for a replicated
prepare,
+ /// which has no client to answer -- and the pre-stamp code served it.
+ #[compio::test]
+ async fn unstamped_parked_frame_is_served_not_rejected_as_stale() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-unstamped");
+ seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Topic slab 1 does not exist yet, so there is no committed
+ // `created_revision` to stamp: the frame parks with `epoch: None`.
+ let unknown = IggyNamespace::new(0, 1, 0);
+ park_one_request(&shard, unknown).await;
+ assert_eq!(
+ shard.parked_frame_count(unknown),
+ 1,
+ "a request for a namespace this node has not applied must park"
+ );
+
+ // The commit this node was lagging behind now lands, and the pass
+ // materialises the namespace.
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ assert_eq!(
+ shard.parked_frame_count(unknown),
+ 0,
+ "materialisation must drain the park entry"
+ );
+ let (served, answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 1,
+ "the unstamped frame must be re-dispatched onto the pump, not
rejected"
+ );
+ assert_eq!(
+ answered, 0,
+ "and it must not be answered with a deny instead of served"
+ );
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 0,
+ "an absent stamp is not evidence of a prior incarnation"
+ );
+ }
+
+ /// The shard-wide byte budget is a running total maintained at each
mutation
+ /// site rather than rescanned per arriving frame. A total that fails to
debit
+ /// on drain silently wedges the budget: the shard would shed every
namespace's
+ /// frames while nothing is actually parked. Exercise each way frames
leave --
+ /// re-dispatch on materialisation, the age bound, and reclaim -- and
assert the
+ /// total returns to empty.
+ #[compio::test]
+ async fn park_byte_total_returns_to_zero_on_every_drain_path() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-bytes");
+ seed_topic(&mux, 2, 0, "topic-bytes", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 16);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Drain path 1: materialisation re-dispatches.
+ let late = IggyNamespace::new(0, 1, 0);
+ park_one_request(&shard, late).await;
+ assert!(shard.has_parked_partition_frames());
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-bytes-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "re-dispatch must debit the parked-byte total"
+ );
+
+ // Drain path 2: the age bound answers the frame.
+ let never = IggyNamespace::new(0, 9, 0);
+ park_one_request(&shard, never).await;
+ assert!(shard.has_parked_partition_frames());
+ for _ in 0..=PARK_MAX_PASSES {
+ shard.age_parked_partition_frames(never);
+ }
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "aging out must debit the parked-byte total"
+ );
+
+ // Drain path 3: an explicit reclaim.
+ park_one_request(&shard, never).await;
+ assert!(shard.has_parked_partition_frames());
+ shard.reclaim_parked_partition_frames(never);
+ assert!(
+ !shard.has_parked_partition_frames(),
+ "reclaim must debit the parked-byte total"
+ );
+
+ drop(inbox);
+ }
+
+ /// The replicated-prepare shape, which no other test covers and where both
+ /// park critical are worst: a prepare has no client, so
`deny_parked_frame`
+ /// no-ops on it and anything that discards it loses committed data
silently,
+ /// with no normal-status repair driver to refetch it.
+ ///
+ /// A backup receives the prepare before its own metadata commits (so the
frame
+ /// parks unstamped), then applies the commit and materialises. The
prepare must
+ /// be re-dispatched, not rejected as a prior incarnation.
+ #[compio::test]
+ async fn unstamped_parked_prepare_is_served_after_materialisation() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-prepare");
+ seed_topic(&mux, 2, 0, "topic-known", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // The primary replicates ahead of this node's metadata: topic slab 1
is
+ // not committed here yet, so the prepare parks with `epoch: None`.
+ let lagging = IggyNamespace::new(0, 1, 0);
+ park_one_prepare(&shard, lagging, 7).await;
+ assert_eq!(
+ shard.parked_frame_count(lagging),
+ 1,
+ "a prepare for a namespace this backup has not applied must park"
+ );
+
+ // The metadata commit catches up and the pass materialises the
namespace.
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 3,
+ 0,
+ "topic-late",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ let (served, answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 1,
+ "the parked prepare must be re-dispatched; discarding it is silent
\
+ committed-data loss, since a prepare has no client to answer"
+ );
+ assert_eq!(answered, 0, "a prepare has no client deny to send");
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 0,
+ "an unstamped prepare is not a prior incarnation"
+ );
+ }
+
+ /// A parked prepare whose stamp names a DIFFERENT incarnation must still
be
+ /// dropped: applying a dead topic's op into the topic that recycled its
slab
+ /// keys diverges this replica. This is the half of the fence that stays.
+ #[compio::test]
+ async fn stamped_parked_prepare_from_a_prior_incarnation_is_rejected() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-stale");
+ seed_topic(&mux, 2, 0, "topic-first", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // Parked against the FIRST incarnation, so it carries that revision.
+ park_one_prepare(&shard, ns, 7).await;
+ assert_eq!(shard.parked_frame_count(ns), 1);
+
+ // Delete and recreate: same namespace keys, new committed revision.
+ seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0);
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 4,
+ 0,
+ "topic-recreated",
+ vec![assignment(0, 2)],
+ );
+ reconcile_pass(&ctx).await;
+
+ let (served, _answered) = drain_inbox(&inbox);
+ assert_eq!(
+ served, 0,
+ "a prepare stamped with the dead incarnation must not be served
against \
+ its replacement"
+ );
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 1,
+ "and the reject must be counted"
+ );
+ }
+
+ /// Parking does not bump `Streams::revision` and does not wake the
reconciler,
+ /// so a frame that parks in a converged steady state would be held for the
+ /// process lifetime if the revision fast-skip could still fire. A
non-empty
+ /// park buffer must therefore defeat the skip.
+ #[compio::test]
+ async fn park_buffer_defeats_the_revision_fast_skip() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-skip");
+ seed_topic(&mux, 2, 0, "topic-skip", vec![assignment(0, 1)]);
+
+ let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+
+ // Converge: the first pass materialises, the verify pass after it arms
+ // the skip (same sequence as
`reconcile_fast_skips_when_revision_unchanged`).
+ assert!(reconcile_once(&ctx).await, "first pass runs the full diff");
+ ctx.shard.apply_reconcile_ops();
+ assert!(
+ reconcile_once(&ctx).await,
+ "the verify pass after a working pass must still run"
+ );
+ ctx.shard.apply_reconcile_ops();
+ assert!(
+ !reconcile_once(&ctx).await,
+ "a converged pass with an unchanged revision must fast-skip"
+ );
+
+ // Park a frame for a namespace that is NOT materialised, without
touching
+ // the revision, and the skip must stop firing.
+ let unbuilt = IggyNamespace::new(0, 0, 7);
+ park_one_request(&shard, unbuilt).await;
+ assert!(
+ shard.has_parked_partition_frames(),
+ "the frame must be parked for this test to mean anything"
+ );
+ assert!(
+ reconcile_once(&ctx).await,
+ "a non-empty park buffer must defeat the fast-skip so the sweep
can run"
+ );
+ }
+
+ /// Same hole, reached the other way: the namespace stays committed but
+ /// `build_partition_fresh` keeps failing. The `FailureCause::Add` backoff
+ /// clamps at 60s, twice the client's read timeout, so holding the frames
+ /// cannot help - answer them and let the client re-issue.
+ #[compio::test]
+ async fn parked_frames_are_reclaimed_while_the_build_is_backed_off() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-backoff");
+ seed_topic(&mux, 2, 0, "topic-backoff", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ park_one_request(&shard, ns).await;
+ assert_eq!(shard.parked_namespaces(), vec![ns]);
+
+ // Stand in for a failed build (ENOSPC / EPERM): the additions pass
skips
+ // a backed-off namespace, so it stays committed and unmaterialised.
+ ctx.record_failure(ns, FailureCause::Add, Instant::now());
+ reconcile_pass(&ctx).await;
+
+ assert!(
+ !ctx.shard.plane.partitions().contains(&ns),
+ "a backed-off namespace must not have been built"
+ );
+ assert!(
+ shard.parked_namespaces().is_empty(),
+ "frames waiting on a backed-off build must be answered, not held"
+ );
+ }
+
+ /// Delete + recreate recycles the slab keys, so a frame parked against the
+ /// dead incarnation is byte-identical to one for its replacement. Draining
+ /// it into the new partition would land a dead topic's write inside the
live
+ /// one, and the incarnation fence cannot catch it: that compares the
+ /// committed revision against the routing row, both of which describe the
+ /// NEW incarnation. Only the epoch stamped at park time separates them.
+ #[compio::test]
+ async fn
parked_frames_from_a_prior_incarnation_are_not_served_by_its_replacement() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-epoch");
+ seed_topic(&mux, 2, 0, "topic-epoch", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // Parked against the first incarnation, before any pass builds it.
+ park_one_request(&shard, ns).await;
+ assert_eq!(shard.parked_namespaces(), vec![ns]);
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 0,
+ "nothing rejected yet"
+ );
+
+ // Recreate the same tuple. The namespace is unchanged; only
+ // `created_revision` moves.
+ seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0);
+ seed_topic(
+ &shard.plane.metadata().mux_stm,
+ 4,
+ 0,
+ "topic-epoch",
+ vec![assignment(0, 1)],
+ );
+
+ // The pass builds the SECOND incarnation and drains the park entry.
+ reconcile_pass(&ctx).await;
+
+ assert!(
+ shard.plane.partitions().contains(&ns),
+ "the recreated incarnation must materialise"
+ );
+ assert!(
+ shard.parked_namespaces().is_empty(),
+ "the park entry must be drained by the materialisation"
+ );
+ assert_eq!(
+ shard.metrics().partition_frames_rejected_stale_value(),
+ 1,
+ "the frame stamped with the dead incarnation must be rejected, not
\
+ re-dispatched into its replacement"
+ );
+ }
+
+ /// Past the per-namespace cap the frame is gone either way, but a client
+ /// request must still be answered: the transports decode replies in
+ /// lockstep, so a silent shed leaves the connection waiting out its full
+ /// response read-timeout.
+ ///
+ /// Driven against a registered in-process client, so the assertion is
that a
+ /// reply reached a waiter -- not that a counter moved. With no client on
the
+ /// bus every send fails as `ClientNotFound` and a counter bumped before
the
+ /// send reports an answer nobody received, which is the failure this test
+ /// exists to catch.
+ #[compio::test]
+ async fn park_overflow_answers_the_client_instead_of_shedding_silently() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-overflow");
+ seed_topic(&mux, 2, 0, "topic-overflow", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ns = IggyNamespace::new(0, 0, 0);
+ let (_slot, reply_rx) = register_waiting_client(&shard);
+
+ // Fill the buffer to its cap, then one more.
+ for _ in 0..PARK_CAP {
+ park_one_request(&shard, ns).await;
+ }
+ assert_eq!(
+ park_overflow_count(&shard),
+ 0,
+ "everything up to the cap parks without shedding"
+ );
+
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 0,
+ "nothing has been answered yet; the parked frames are still
waiting"
+ );
+
+ park_one_request(&shard, ns).await;
+ assert_eq!(
+ park_overflow_count(&shard),
+ 1,
+ "the frame past the cap must be shed and counted, not parked"
+ );
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ PARK_CAP,
+ "the shed frame must not have grown the buffer past its cap"
+ );
+ // The point of the fix: shedding is unavoidable at the cap, silence is
+ // not. Without the deny the connection waits out its whole response
+ // read-timeout on a frame that is already gone.
+ let reply = reply_rx
+ .await
+ .expect("the shed request must reach the waiting client");
+ assert_eq!(
+ reply_status(&reply),
+ iggy_common::IggyError::TransientNotAccepted.as_code(),
+ "the shed request must be answered with a retriable status"
+ );
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 1,
+ "and the counter must credit that delivered answer"
+ );
+ }
+
+ /// The counter must credit only denies the bus delivered. It previously
+ /// incremented before `send_to_client`, so a shard with no client
registered
+ /// still reported the request answered - which blinded
+ /// `park_overflow_answers_the_client_instead_of_shedding_silently`, the
one
+ /// test that asserts the client hears back.
+ #[compio::test]
+ async fn overflow_deny_is_not_counted_when_the_client_is_gone() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-overflow-gone");
+ seed_topic(&mux, 2, 0, "topic-overflow-gone", vec![assignment(0, 1)]);
+
+ // No client registered: every `send_to_client` fails `ClientNotFound`.
+ let shard = build_test_shard(0, &config, mux);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ for _ in 0..=PARK_CAP {
+ park_one_request(&shard, ns).await;
+ }
+
+ assert_eq!(
+ park_overflow_count(&shard),
+ 1,
+ "the frame past the cap is still shed and counted"
+ );
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 0,
+ "a deny the bus could not deliver must not be counted as an answer"
+ );
+ assert_eq!(
+ shard.metrics().frame_drop_count(
+ shard::metrics::frame_drop_variant::PARTITION,
+ shard::metrics::frame_drop_reason::DELIVERY_FAILED,
+ ),
+ 1,
+ "it must be counted as an undelivered reply instead"
+ );
+ }
+
+ /// A parked prepare that ages out has no client to answer, so nothing is
+ /// staged and `partition_requests_denied_transient_total` stays put. The
op
+ /// is destroyed all the same - the primary retransmits only what has not
+ /// reached quorum - so `park_dropped` is the only record it existed.
+ #[compio::test]
+ async fn aged_out_prepare_is_counted_even_though_nobody_can_be_answered() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-prepare-age");
+ seed_topic(&mux, 2, 0, "topic-prepare-age", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ park_one_prepare(&shard, ns, 7).await;
+ for _ in 0..=PARK_MAX_PASSES {
+ shard.age_parked_partition_frames(ns);
+ }
+
+ assert_eq!(shard.parked_frame_count(ns), 0, "the prepare aged out");
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 0,
+ "a prepare has no client to answer"
+ );
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 0,
+ "and must not be reported as an answered request"
+ );
+ assert_eq!(
+ park_dropped_count(&shard),
+ 1,
+ "the destroyed op must leave a record; silence here is invisible
loss"
+ );
+ }
+
+ /// A frame larger than the per-namespace byte cap used to fail the check
+ /// even against an empty entry, so it could never park on any attempt.
For a
+ /// replicated prepare that is unrecoverable: `retransmit_targets` skips
an op
+ /// that already reached quorum, so the backup stays permanently short of
it.
+ #[compio::test]
+ async fn
a_frame_over_the_namespace_byte_cap_still_parks_into_an_empty_entry() {
+ const OVER_NAMESPACE_CAP: usize = 5 * 1024 * 1024;
+
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-oversize");
+ seed_topic(&mux, 2, 0, "topic-oversize", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ shard
+ .on_message(build_partition_request_sized(ns, OVER_NAMESPACE_CAP))
+ .await;
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "the first frame of an empty entry must park regardless of the
per-namespace cap"
+ );
+ assert_eq!(
+ park_overflow_count(&shard),
+ 0,
+ "and must not be shed doing it"
+ );
+
+ // The waiver is for the first frame only; the cap still applies after.
+ shard
+ .on_message(build_partition_request_sized(ns, OVER_NAMESPACE_CAP))
+ .await;
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ 1,
+ "a second oversize frame must be shed, or one namespace eats the
shard budget"
+ );
+ assert_eq!(park_overflow_count(&shard), 1);
+ }
+
+ /// A namespace whose build is still in flight keeps its frames -- but not
+ /// forever, or the park buffer grows with a namespace that never
materialises.
+ /// The bound is in reconciler passes so the simulator's virtual clock
governs
+ /// it.
+ ///
+ /// Driven through `age_parked_partition_frames` directly. The sweep calls
it
+ /// once per pass for a namespace still building, and that branch is the
only
+ /// way a committed, non-backed-off namespace reaches the bound - which a
unit
+ /// test cannot stage, since its build completes on the first pass.
+ ///
+ /// Uses a shard with a live inbox: the deny is staged onto the pump, so a
+ /// shard with no sender would report the frame answered while nothing was
+ /// ever handed anywhere.
+ #[compio::test]
+ async fn
parked_frames_are_answered_once_they_outlive_their_admission_window() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-age");
+ seed_topic(&mux, 2, 0, "topic-age", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ park_one_request(&shard, ns).await;
+ assert_eq!(shard.parked_frame_count(ns), 1);
+
+ // Each pass ages the frame; it survives until it is over the bound.
+ for pass in 0..PARK_MAX_PASSES {
+ assert_eq!(
+ shard.age_parked_partition_frames(ns),
+ 0,
+ "pass {pass} is still inside the admission window"
+ );
+ assert_eq!(shard.parked_frame_count(ns), 1);
+ }
+ assert_eq!(
+ shard.age_parked_partition_frames(ns),
+ 1,
+ "the pass past the bound must answer the frame"
+ );
+ assert_eq!(shard.parked_frame_count(ns), 0);
+ assert_eq!(
+ drain_staged_client_sends(&inbox),
+ 1,
+ "the answer must actually reach the pump, not just the counter"
+ );
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 1,
+ "and it must be answered with a retriable status, not dropped"
+ );
+ }
+
+ /// The counter must credit only denies the pump accepted. It previously
+ /// incremented before the `try_send`, so a shard whose inbox refused the
frame
+ /// (or had no sender at all) still reported the client answered.
+ #[compio::test]
+ async fn transient_deny_is_not_counted_when_the_inbox_cannot_take_it() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-deny-drop");
+ seed_topic(&mux, 2, 0, "topic-deny-drop", vec![assignment(0, 1)]);
+
+ let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8);
+ let ns = IggyNamespace::new(0, 0, 0);
+ park_one_request(&shard, ns).await;
+
+ // Kill the pump side, so every staged frame is refused.
+ drop(inbox);
+
+ for _ in 0..=PARK_MAX_PASSES {
+ shard.age_parked_partition_frames(ns);
+ }
+
+ assert_eq!(shard.parked_frame_count(ns), 0, "the frame still ages
out");
+ assert_eq!(
+ shard.metrics().partition_requests_denied_transient_value(),
+ 0,
+ "a deny the inbox refused must not be counted as an answer"
+ );
+ }
+
+ /// The frame cap bounds count, not residency: `Message::into_generic` is a
+ /// retag, so each parked entry keeps its whole buffer -- up to 64 MiB.
With
+ /// only a frame cap, one namespace could pin 128 x 64 MiB and nothing
capped
+ /// the namespace count. The shard-wide byte budget is what actually bounds
+ /// it, so large frames must shed well before the frame cap.
+ #[compio::test]
+ async fn park_byte_budget_sheds_large_frames_before_the_frame_cap() {
Review Comment:
this trips the per-namespace cap, not the shard-wide budget it is named and
asserted for: all frames share one namespace, footprint per 1 MiB body is
1052672, so frame 4 sheds at 4210688 > `MAX_PARKED_BYTES_PER_NAMESPACE` (4 MiB)
- nowhere near 16 MiB. both asserts stay green for the wrong reason (4 < 128).
net effect: `MAX_PARKED_BYTES` is never the binding constraint in any test in
the repo, and its only single-namespace-reachable behaviour is the
first-frame-waiver path where the >16 MiB gap lives.
to actually bind the shard-wide budget: 5 namespaces x 3 frames all fit
(15790080), then the first frame into a 6th namespace crosses 16 MiB while
per-namespace-waived - unambiguously the shard-wide shed. and rename this to
the cap it currently proves.
##########
core/shard/src/lib.rs:
##########
@@ -1781,29 +1958,340 @@ where
/// decode replies in lockstep, so silence wedges the connection until the
/// SDK's response read-timeout.
fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
- if let Some(frames) = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)
+ // Bound the borrow to this statement: the guard in an `if let`
+ // scrutinee otherwise lives to the end of the then-block, holding a
+ // shard-global map locked across the outbound sends below.
+ let parked = self.take_parked_partition_frames(namespace);
+ if let Some(frames) = parked
&& !frames.is_empty()
{
+ let total = frames.len();
+ let mut answered = 0;
+ for frame in frames {
+ if self.deny_parked_client_request(frame) {
+ answered += 1;
+ } else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ }
+ }
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- count = frames.len(),
+ answered,
+ dropped = total - answered,
"discarding parked partition frames for removed namespace"
);
- for frame in frames {
- if frame.header().command == Command2::Request
- && let Ok(request) =
frame.try_into_typed::<RequestHeader>()
+ }
+ }
+
+ /// Remove a namespace's park entry, debiting its bytes from
+ /// [`Self::parked_partition_bytes`]. The single place entries leave the
map,
+ /// so the running total and the pending-retry set cannot drift out of step
+ /// with it.
+ fn take_parked_partition_frames(&self, namespace: IggyNamespace) ->
Option<Vec<ParkedFrame>> {
+ self.reparked_partition_namespaces
+ .borrow_mut()
+ .remove(&namespace);
+ let frames = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)?;
+ let freed: usize = frames
+ .iter()
+ .map(|frame| parked_footprint(frame.message.as_slice().len()))
+ .sum();
+ self.parked_partition_bytes
+ .set(self.parked_partition_bytes.get().saturating_sub(freed));
+ Some(frames)
+ }
+
+ /// Whether any frame is parked. Cheap enough for the reconciler's per-tick
+ /// fast-skip guard: a non-empty buffer means the shard is by definition
not
+ /// converged, so the skip must not fire.
+ #[must_use]
+ pub const fn has_parked_partition_frames(&self) -> bool {
+ self.parked_partition_bytes.get() > 0
+ }
+
+ /// Namespaces currently holding parked frames. The reconciler pairs this
+ /// against committed metadata to find the ones that will never
materialise,
+ /// which no `ConfirmRemove` / `RemoveRouted` can reach: a namespace that
was
+ /// never built is in neither `IggyPartitions` nor the routing table, so
+ /// nothing else names it.
+ #[must_use]
+ pub fn parked_namespaces(&self) -> Vec<IggyNamespace> {
+ self.pending_partition_frames
+ .borrow()
+ .keys()
+ .copied()
+ .collect()
+ }
+
+ /// Re-queue the frames parked for `namespace` now that its partition
exists
+ /// at `epoch`, onto this shard's own inbox so the pump serves them after
the
+ /// current drain.
+ ///
+ /// A frame stamped with a DIFFERENT incarnation never makes it back: the
+ /// namespace is byte-identical across incarnations, so serving it would
land
+ /// a dead topic's write inside the topic that recycled its keys, and the
+ /// downstream fence cannot see it -- that compares the committed revision
+ /// against the routing row, both of which now describe THIS incarnation.
+ ///
+ /// An UNSTAMPED frame (`epoch: None`) is served. `None` means this node's
+ /// metadata held no committed partition for the namespace when the frame
+ /// arrived, which on a metadata-lagging backup is the ordinary case the
park
+ /// buffer exists to absorb -- the partition primary materialises and
+ /// replicates as soon as its own metadata commits, well before a lagging
+ /// backup applies the same commit. Treating that as "prior incarnation"
+ /// destroys live traffic: a replicated prepare has no client to answer, so
+ /// it would be dropped with no recovery until an unrelated view change.
+ /// The residual is unchanged from before the stamp existed -- a frame
parked
+ /// while the namespace was absent, then recreated under a new incarnation,
+ /// is served against the replacement -- and closing it needs a wire-level
+ /// discriminator (see the `TODO(krishna)` in
+ /// `partition_reconciler`'s module docs), not a `None`-means-stale rule.
+ ///
+ /// A frame the inbox refuses is re-parked, not answered. Re-queuing
appends,
+ /// so a pass that materialises many namespaces at once can overrun the
inbox;
+ /// staging a deny there is futile because the deny rides that same sender
+ /// with no await in between, so nothing can have drained a slot.
+ ///
+ /// [`MAX_PARKED_PASSES`] does NOT bound a re-parked frame -- the
reconciler
+ /// sweep ages a namespace only while it is un-materialised, and by here it
+ /// is materialised. [`Self::repark_partition_frames`] arms the pump-side
+ /// retry instead, and the sweep's backstop for an inbox that never drains
is
+ /// `partition_reconciler::reconcile_parked_frames`, which now ages a
+ /// materialised namespace too.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64)
+ where
+ B: MessageBus + 'static,
+ {
+ let Some(frames) = self.take_parked_partition_frames(namespace) else {
+ return;
+ };
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ epoch,
+ "re-dispatching parked partition frames after materialisation"
+ );
+ let mut refused_frames: Vec<ParkedFrame> = Vec::new();
+ for frame in frames {
+ // Only a stamp that exists and disagrees is evidence of a prior
+ // incarnation; see this function's docs on why `None` is not.
+ if let Some(parked_epoch) = frame.epoch
+ && parked_epoch != epoch
+ {
+ 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 and arming the pump-side
retry.
+ ///
+ /// 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.
+ ///
+ /// 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`.
+ fn repark_partition_frames(&self, namespace: IggyNamespace, frames:
Vec<ParkedFrame>) {
Review Comment:
the per-namespace byte total is re-summed over the whole entry at four sites
(park admission, take, repark, age) while the shard-wide total is an
incremental cell. a map value of `ParkEntry { frames: Vec<ParkedFrame>, bytes:
usize }` kills the rescans, and reading emptiness from the map
(`!pending.borrow().is_empty()`) instead of the byte cell removes the
bytes-as-emptiness-oracle invariant - which would also make the "single place
entries leave the map" claim on `take_parked_partition_frames` true (aging
currently removes entries directly and never clears the retry set).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
