krishvishal commented on code in PR #3786:
URL: https://github.com/apache/iggy/pull/3786#discussion_r3706449618
##########
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:
Fixed. `age_parked_partition_frames` now expires client requests only; the
extract predicate skips `is_replicated()`. Prepare residency is bounded by the
byte budgets instead. The only path that still destroys a parked prepare is
`discard_parked_partition_frames`, where the namespace is unreachable from this
shard (tombstoned or not hashing here). Covered by
`aging_answers_requests_and_never_expires_a_prepare`.
##########
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:
Fixed the way you suggested: prepares are not shed on a byte budget crossing
at all. A request is refused the moment admitting it would cross a budget; a
prepare only once a budget is already spent, which gives one frame of overshoot
(worst case `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead
of loss. That also removes the >16 MiB gap, so the TODO is gone. Covered by
`a_frame_over_the_namespace_byte_cap_still_parks_into_an_empty_entry`.
##########
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:
You are right, the 60s justification was wrong: `next_backoff(1)` is 1s and
the clamp needs seven consecutive failures. The `is_backed_off` branch is gone
from the sweep, so a backed-off namespace is aged like any other (requests get
`MAX_PARKED_PASSES`, prepares are retained). Immediate discard is now only
tombstoned or not-ours. Covered by
`a_backed_off_build_ages_requests_and_retains_prepares`.
##########
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:
Fixed in one edit as suggested. On `Full`: push the frame, `extend` with the
rest of the iterator, `break`, no counter. The drop is recorded only on the
`Disconnected` arm. The unreachable destructure is now `unreachable!`, and the
inline duplicate of `deny_parked_client_request` folds into the shared
`retire_parked_frames`. Covered by
`a_re_parked_frame_is_re_dispatched_once_the_inbox_drains`.
##########
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:
Fixed with the shape you proposed: `attach_senders(&mut self,
Vec<TaggedSender>)`, run through `validate_sender_ordering`, updating
`shard_count`, gated `#[cfg(any(test, feature = "simulator"))]`. The test
builder now wires `0..=shard_id` and drops the peer receivers, so a frame
misrouted to one fails as `Disconnected` instead of landing in this shard's
inbox and reading as success.
##########
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:
Fixed. `stage_transient_deny` returns `bool` and
`deny_parked_client_request` forwards it, so a request whose deny was never
staged is counted under `park_dropped` rather than logged as answered.
--
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]
