krishvishal commented on code in PR #3786:
URL: https://github.com/apache/iggy/pull/3786#discussion_r3706453467
##########
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:
Deleted. `discard_parked_partition_frames` is `pub` and callers use it
directly.
##########
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:
Extracted `retire_parked_frames(frames) -> (answered, dropped)`, used by
both sites. That also single-sited the accounting fix 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:
Delegated to `build_partition_request_sized(namespace, 0)`.
--
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]
