krishvishal commented on code in PR #3786:
URL: https://github.com/apache/iggy/pull/3786#discussion_r3706452307
##########
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:
Fixed, and stated outright: `reconcile_additions` has no cross-pass guard
against a queued-but-unapplied `InsertOwned`, so it re-stages every pass until
the pump drains. The exemption therefore covers arbitrary pump lag, and
removing it would age frames on every commit-driven pass the pump falls behind.
##########
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:
Fixed. `ParkEntry` carries a `shed` count; the transition into shedding
warns once per namespace and everything after logs at `debug`. The counter
carries the volume.
##########
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:
Noted in a comment beside the constants: 12 pairs are unreachable, and once
a registry lands the counters should be minted per drop site rather than by
cross product so the scrape carries no permanent zeroes.
##########
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:
Fixed both. Now `shard = { workspace = true, features = ["simulator"] }`,
and the comment says the resolver keeps a dev-dependency's feature selection
out of non-test targets, rather than claiming dev-dependencies are not resolved.
##########
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:
Done. `ParkEntry { frames, bytes, shed }` removes the rescans at all four
sites, `has_parked_partition_frames` reads the map instead of using the byte
cell as an emptiness oracle, and aging now routes its removal through
`take_parked_partition_frames`, which clears the retry set with the entry. The
single-removal-site claim is true now.
--
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]
