This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch part-mat-barrier
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/part-mat-barrier by this push:
new 4d7960c77 fix: address review comments
4d7960c77 is described below
commit 4d7960c773f16ae78a17dc4e578b4a23bf371e2e
Author: Krishna Vishal <[email protected]>
AuthorDate: Tue Aug 4 00:16:00 2026 +0530
fix: address review comments
---
core/server-ng/src/partition_reconciler.rs | 108 ++++++++++++++++++++++++-----
core/shard/src/lib.rs | 78 ++++++++++++++++-----
2 files changed, 148 insertions(+), 38 deletions(-)
diff --git a/core/server-ng/src/partition_reconciler.rs
b/core/server-ng/src/partition_reconciler.rs
index b150e9ecb..fde1d682a 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -91,14 +91,18 @@
//! shed request costs a retry: answered with a retriable status, re-issued by
//! the SDK. A shed prepare is permanent loss on this replica, with no client
to
//! answer and `consensus::retransmit_targets` skipping any op that already
-//! reached quorum. So a request is refused the moment admitting it would
cross a
-//! budget, and answered once it outlives `MAX_PARKED_PASSES`; a prepare never
-//! expires by age and is refused only once a budget is already spent. That
caps
-//! prepare residency at one frame of overshoot per budget instead of at the
-//! budget, and is what makes an oversize frame parkable at all: measured
against
-//! an empty entry, a 5 MiB append would fail the per-namespace check on every
-//! attempt. A parked prepare is destroyed only for a namespace this shard
cannot
-//! serve at all, tombstoned or not hashing here.
+//! reached quorum.
+//!
+//! All three bind a request: refused when admitting it would cross a byte
budget
+//! or the frame cap, answered past `MAX_PARKED_PASSES`. Only the byte budgets
+//! bind a prepare, and only once one is spent. Excluding the frame cap is
+//! deliberate: a header-only frame charges `MESSAGE_ALIGN`, so 128 is 512 KiB
+//! against a 4 MiB namespace budget, and a shared cap would shed small
prepares
+//! before any byte budget spoke. Prepares admit 1024 header-only frames per
+//! namespace, overshooting each budget by one frame rather than stopping at
it,
+//! which also makes an oversize frame parkable: against an empty entry a 5 MiB
+//! append fails the per-namespace check every attempt. A parked prepare dies
only
+//! for a namespace this shard cannot serve, tombstoned or not hashing here.
//!
//! Everything leaving the buffer unserved is counted under
//! `frame_drops_total{variant=partition}`: `park_overflow` when shed on
arrival,
@@ -117,7 +121,7 @@
//! in `on_start_view` -- `tick_partitions` re-drives an existing session but
//! cannot open one -- so the backup stays behind `commit_max` until an
unrelated
//! view change. It needs a normal-status repair driver. The park policy above
-//! shrinks the exposure to two cases, a genuinely exhausted budget and a
+//! shrinks the exposure to two cases, a genuinely exhausted byte budget and a
//! namespace this shard cannot serve, but only the repair driver removes it.
//!
//! TODO(krishna): the park stamp is not stable across re-entry. A
re-dispatched
@@ -3180,6 +3184,61 @@ mod tests {
);
}
+ /// The frame cap is request-only. Applied to prepares it is the binding
+ /// constraint for any footprint under `NAMESPACE_BUDGET / PARK_CAP` (32
KiB),
+ /// so header-only prepares would shed at 128 frames, 512 KiB into a 4 MiB
+ /// budget, and the byte budgets would never speak. Small-append prepares
+ /// would then be destroyed exactly as before the class split, at every
+ /// replica count: quorum always leaves at least one surplus backup, so a
+ /// lagging one loses shed prepares with nobody noticing.
+ #[compio::test]
+ async fn
the_frame_cap_bounds_requests_only_so_small_prepares_reach_the_byte_budget() {
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-frame-cap");
+ seed_topic(&mux, 2, 0, "topic-frame-cap", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ns = IggyNamespace::new(0, 0, 0);
+
+ // Well past the frame cap, prepares keep parking.
+ let beyond_cap = PARK_CAP + 72;
+ for op in 0..beyond_cap {
+ park_one_prepare(&shard, ns, op as u64).await;
+ }
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ beyond_cap,
+ "the frame cap must not shed prepares"
+ );
+ assert_eq!(park_overflow_count(&shard), 0);
+
+ // A request into that same deep entry is still capped.
+ park_one_request(&shard, ns).await;
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ beyond_cap,
+ "a request past the cap must still be shed"
+ );
+ assert_eq!(park_overflow_count(&shard), 1);
+
+ // The per-namespace byte budget is what finally stops the prepares.
+ for op in beyond_cap..=HEADER_FRAMES_PER_NAMESPACE {
+ park_one_prepare(&shard, ns, op as u64).await;
+ }
+ assert_eq!(
+ shard.parked_frame_count(ns),
+ HEADER_FRAMES_PER_NAMESPACE,
+ "prepares must admit up to the byte budget, not the frame cap"
+ );
+ assert_eq!(
+ park_overflow_count(&shard),
+ 2,
+ "and the frame crossing the byte budget is the only further shed"
+ );
+ }
+
/// The frame cap bounds count, not residency: `Message::into_generic` is a
/// retag, so each entry keeps its whole buffer, up to 64 MiB. With only a
/// frame cap one namespace could pin 128 x 64 MiB, so bytes must bite
first.
@@ -3249,16 +3308,22 @@ mod tests {
);
// One frame into an empty entry: per-namespace waived, shard-wide not.
- let crossing = IggyNamespace::new(0, filled, 0);
- shard
- .on_message(build_partition_request_sized(crossing, MIB_BODY))
- .await;
- assert_eq!(
- shard.parked_frame_count(crossing),
- 0,
- "the frame that would cross the shard-wide budget must be shed"
- );
- assert_eq!(park_overflow_count(&shard), 1);
+ // Two distinct fresh namespaces, so the entryless shed path runs
twice.
+ // It has no `ParkEntry` to warn once from and is gated shard-wide
+ // instead; the gate is log volume only, so the counter still records
+ // every shed.
+ for offset in 0..2 {
+ let crossing = IggyNamespace::new(0, filled + offset, 0);
+ shard
+ .on_message(build_partition_request_sized(crossing, MIB_BODY))
+ .await;
+ assert_eq!(
+ shard.parked_frame_count(crossing),
+ 0,
+ "the frame that would cross the shard-wide budget must be shed"
+ );
+ }
+ assert_eq!(park_overflow_count(&shard), 2);
}
/// A refused re-dispatch re-parks the frame, and by then the namespace is
@@ -3394,6 +3459,11 @@ mod tests {
/// pushes a 1 MiB body into the next page.
const MIB_BODY: usize = 1024 * 1024;
const MIB_FOOTPRINT: usize = MIB_BODY + 4096;
+ /// Footprint of a header-only frame: buffers are `MESSAGE_ALIGN`-granular.
+ const HEADER_FOOTPRINT: usize = 4096;
+ /// Header-only frames one namespace admits before its byte budget refuses
+ /// more. Requests stop at [`PARK_CAP`] long before this; prepares do not.
+ const HEADER_FRAMES_PER_NAMESPACE: usize = NAMESPACE_BUDGET /
HEADER_FOOTPRINT;
/// [`MIB_BODY`] frames one namespace admits before its budget refuses
more.
const PER_NAMESPACE_MIB_FRAMES: usize = NAMESPACE_BUDGET / MIB_FOOTPRINT;
/// Mirrors `MAX_PARKED_PASSES`.
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 6bca33a19..aec89d61b 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1034,6 +1034,11 @@ where
/// function of the namespaces alone.
reparked_partition_namespaces: RefCell<BTreeSet<IggyNamespace>>,
+ /// Set while the shard-wide budget is shedding for namespaces holding no
+ /// park entry of their own, which have no [`ParkEntry::shed`] to warn once
+ /// from. Cleared when the park map empties, so one episode warns once.
+ shard_park_shedding: Cell<bool>,
+
/// Live ceiling on prepares served per `RequestPrepares` round. Defaults
/// to [`REPAIR_CHUNK_MAX`]; server-ng overrides it from
/// `[cluster] repair_chunk_max` at bootstrap.
@@ -1164,6 +1169,7 @@ where
pending_partition_frames: RefCell::new(BTreeMap::new()),
parked_partition_bytes: Cell::new(0),
reparked_partition_namespaces: RefCell::new(BTreeSet::new()),
+ shard_park_shedding: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
metadata_transfer_offers: RefCell::new(HashMap::new()),
@@ -1401,6 +1407,7 @@ where
pending_partition_frames: RefCell::new(BTreeMap::new()),
parked_partition_bytes: Cell::new(0),
reparked_partition_namespaces: RefCell::new(BTreeSet::new()),
+ shard_park_shedding: Cell::new(false),
metadata_repair: RefCell::new(None),
metadata_transfer: RefCell::new(None),
metadata_transfer_offers: RefCell::new(HashMap::new()),
@@ -1762,7 +1769,14 @@ impl ParkEntry {
}
}
-/// Per-namespace ceiling on parked frames.
+/// Per-namespace ceiling on parked CLIENT REQUESTS.
+///
+/// Requests only, like the byte budgets: a header-only frame charges
+/// [`MESSAGE_ALIGN`], so 128 of them is 512 KiB against a 4 MiB per-namespace
+/// budget. Applied to prepares this would be the binding constraint for every
+/// footprint under 32 KiB and would shed them long before any byte budget
could,
+/// which is the loss class the split exists to remove. A prepare is bounded by
+/// [`MAX_PARKED_BYTES_PER_NAMESPACE`] instead: 1024 header-only frames.
const MAX_PARKED_PER_NAMESPACE: usize = 128;
/// Shard-wide ceiling on parked bytes, measured as resident footprint (see
@@ -2019,15 +2033,21 @@ where
self.reparked_partition_namespaces
.borrow_mut()
.remove(&namespace);
- let entry = self
- .pending_partition_frames
- .borrow_mut()
- .remove(&namespace)?;
+ let (entry, converged) = {
+ let mut pending = self.pending_partition_frames.borrow_mut();
+ let entry = pending.remove(&namespace)?;
+ let converged = pending.is_empty();
+ (entry, converged)
+ };
self.parked_partition_bytes.set(
self.parked_partition_bytes
.get()
.saturating_sub(entry.bytes),
);
+ if converged {
+ // Episode over: the next entryless shed is a new one and warns.
+ self.shard_park_shedding.set(false);
+ }
Some(entry.frames)
}
@@ -2104,7 +2124,10 @@ where
/// A frame the inbox refuses is re-parked: retained, so not counted as a
/// drop. Re-queuing appends, so a pass materialising many namespaces can
/// overrun the inbox; staging a deny is futile because it rides the same
- /// sender with no await in between. The first `Full` ends the loop, since
+ /// sender with no await in between. One namespace alone can now do it,
since
+ /// [`MAX_PARKED_BYTES_PER_NAMESPACE`] admits 1024 header-only prepares
+ /// against a default `inbox_capacity` of 1024. That costs other
namespaces a
+ /// later convergence, not a frame. The first `Full` ends the loop, since
/// the sole consumer of `senders[self.id]` is the pump task running this
/// call and no later frame can find a slot the first one could not.
///
@@ -2274,8 +2297,9 @@ where
count
}
- /// How many frames are parked under `namespace`. Bounded by
- /// `MAX_PARKED_PER_NAMESPACE`; a shed frame must never grow it past that.
+ /// How many frames are parked under `namespace`. Client requests are
bounded
+ /// by `MAX_PARKED_PER_NAMESPACE`, prepares by
+ /// `MAX_PARKED_BYTES_PER_NAMESPACE`; a shed frame must never grow either.
///
/// Test/simulator accessor: nothing in production branches on a
per-namespace
/// park depth, and gating keeps it that way.
@@ -2430,21 +2454,37 @@ where
} else {
parked_bytes.saturating_add(frame_cost) > MAX_PARKED_BYTES
};
- if parked_len >= MAX_PARKED_PER_NAMESPACE || namespace_budget_spent ||
shard_budget_spent {
+ // The frame cap is request-only for the same reason. Applied to both
it
+ // would be the binding constraint for any footprint under
+ // `MAX_PARKED_BYTES_PER_NAMESPACE / MAX_PARKED_PER_NAMESPACE` (32
KiB),
+ // so header-only prepares would shed at 128 frames, 512 KiB into a 4
MiB
+ // budget, and the byte budgets above would never get a say.
+ let frame_cap_spent = !replicated && parked_len >=
MAX_PARKED_PER_NAMESPACE;
+ if frame_cap_spent || namespace_budget_spent || shard_budget_spent {
self.metrics.record_frame_drop(
crate::metrics::frame_drop_variant::PARTITION,
crate::metrics::frame_drop_reason::PARK_OVERFLOW,
);
- // Warn once per namespace on the transition into shedding, then
- // `debug`. A full buffer is this branch's trigger, not a rate
limit:
- // every later frame lands here too, and one formatted `warn`
apiece
- // is enough for the non-blocking appender to shed unrelated lines.
- // The counter carries the volume.
- let first_shed = existing.is_none_or(|entry| {
- let first = entry.shed == 0;
- entry.shed = entry.shed.saturating_add(1);
- first
- });
+ // Warn once per namespace on entering the shed, then `debug`: a
+ // full buffer is this branch's trigger, not a rate limit, so every
+ // later frame lands here too, and one formatted `warn` apiece
makes
+ // the non-blocking appender shed unrelated lines. The counter
+ // carries the volume.
+ //
+ // An entryless namespace has no `ParkEntry::shed` to gate on and
is
+ // reachable only via the shard-wide budget (the other two
conditions
+ // need a non-empty entry), which is the many-namespace burst
+ // `MAX_PARKED_BYTES` is sized for. Hence the shard-level gate, and
+ // not `entry().or_default()`, which leaves the empty entry the
read
+ // above avoids.
+ let first_shed = match existing {
+ Some(entry) => {
+ let first = entry.shed == 0;
+ entry.shed = entry.shed.saturating_add(1);
+ first
+ }
+ None => !self.shard_park_shedding.replace(true),
+ };
if first_shed {
tracing::warn!(
shard = self.id,
