You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/07/24 22:32:54 UTC
(iggy) 01/12: fix(consensus): bound pipeline asserts by configured depth, not const
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/iggy-server-hardening
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 12df22e920aeb89ca8790dd9f5a532fc89b04e16
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Jul 24 10:54:20 2026 +0200
fix(consensus): bound pipeline asserts by configured depth, not const
metadata.prepare_queue_depth accepts values up to 4096, but
Pipeline::verify(), the view-change rebuild, and loopback push
asserted against PIPELINE_PREPARE_QUEUE_MAX=32. verify() also
checked the request queue against a const 64 while bootstrap
sizes it at depth*2. Any configured depth above 32 panicked at
runtime.
Expose prepare_queue_max() on the Pipeline trait and snapshot it
into VsrConsensus at construction, before the pipeline moves into
its RefCell, so assert sites need no re-borrow. All bounds now
compare against the configured capacity; the const remains as the
default depth.
---
core/consensus/src/impls.rs | 80 ++++++++++++++++++++++++++++---------
core/consensus/src/lib.rs | 5 +++
core/consensus/src/plane_helpers.rs | 74 ++++++++++++++++++++++++++++++++--
3 files changed, 138 insertions(+), 21 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 8869f870f..b041b0df8 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -122,16 +122,18 @@ impl Sequencer for LocalSequencer {
}
}
-/// TODO The below numbers need to be added a consensus config
-/// TODO understand how to configure these numbers.
-/// Maximum number of prepares that can be in-flight in the pipeline.
+/// Default in-flight prepare-queue depth.
///
-/// Sized to absorb a synchronized client burst (e.g. the 20-way
-/// concurrent-creation race tests across TCP/QUIC/WebSocket) without
-/// `PipelineFull`-rejecting and disconnecting clients that cannot replay in
-/// time. At depth 8 the QUIC burst wedges the metadata consensus even in
-/// release. Stays well under the journal's `SLOT_COUNT` (1024) and the inbox
-/// capacity headroom.
+/// [`LocalPipeline::new`] uses it, and the server-ng config default
+/// (`DEFAULT_METADATA_PREPARE_QUEUE_DEPTH`) is static-asserted equal to it at
+/// bootstrap. Operators raise the running bound via `[metadata]
+/// prepare_queue_depth`; the pipeline then carries its own capacity (see
+/// [`LocalPipeline::with_capacities`]).
+///
+/// Sized to absorb a synchronized client burst (the 20-way concurrent-creation
+/// race tests across TCP/QUIC/WebSocket) without `PipelineFull`-rejecting
+/// clients that cannot replay in time; a depth of 8 wedged the QUIC burst even
+/// in release. Stays well under the journal's slot count and the inbox headroom.
pub const PIPELINE_PREPARE_QUEUE_MAX: usize = 32;
/// Max accepted-but-not-yet-prepared requests buffered behind a full
@@ -447,7 +449,7 @@ impl LocalPipeline {
}
/// Find a message by op number and checksum (immutable).
- // Pipeline bounded at PIPELINE_PREPARE_QUEUE_MAX (8) entries; index always fits in usize.
+ // op - head_op is bounded by the configured prepare-queue depth; index always fits in usize.
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn message_by_op_and_checksum(&self, op: u64, checksum: u128) -> Option<&PipelineEntry> {
@@ -478,7 +480,7 @@ impl LocalPipeline {
}
/// Find a message by op number only.
- // Pipeline bounded at PIPELINE_PREPARE_QUEUE_MAX (8) entries; index always fits in usize.
+ // op - head_op is bounded by the configured prepare-queue depth; index always fits in usize.
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn message_by_op(&self, op: u64) -> Option<&PipelineEntry> {
@@ -494,7 +496,7 @@ impl LocalPipeline {
/// Get mutable reference to a message entry by op number.
/// Returns None if op is not in the pipeline.
- // Pipeline bounded at PIPELINE_PREPARE_QUEUE_MAX (8) entries; index always fits in usize.
+ // op - head_op is bounded by the configured prepare-queue depth; index always fits in usize.
#[allow(clippy::cast_possible_truncation)]
pub fn message_by_op_mut(&mut self, op: u64) -> Option<&mut PipelineEntry> {
let head_op = self.prepare_queue.front()?.header.op;
@@ -531,8 +533,8 @@ impl LocalPipeline {
/// If any invariant is violated.
pub fn verify(&self) {
// Check capacity limits
- assert!(self.prepare_queue.len() <= PIPELINE_PREPARE_QUEUE_MAX);
- assert!(self.request_queue.len() <= PIPELINE_REQUEST_QUEUE_MAX);
+ assert!(self.prepare_queue.len() <= self.prepare_queue_max);
+ assert!(self.request_queue.len() <= self.request_queue_max);
// Verify prepare queue hash chain
if let Some(head) = self.prepare_queue.front() {
@@ -622,6 +624,10 @@ impl Pipeline for LocalPipeline {
self.prepare_count()
}
+ fn prepare_queue_max(&self) -> usize {
+ self.prepare_queue_max
+ }
+
fn verify(&self) {
Self::verify(self);
}
@@ -809,6 +815,10 @@ where
last_prepare_checksum: Cell<u128>,
pipeline: RefCell<P>,
+ /// Snapshot of the pipeline's in-flight prepare capacity, taken at
+ /// construction. Bounds the loopback queue and the view-change rebuild
+ /// range without re-borrowing `pipeline`.
+ prepare_queue_max: usize,
message_bus: B,
loopback_queue: RefCell<VecDeque<Message<GenericHeader>>>,
@@ -896,6 +906,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// across groups. Consider using a proper hash (e.g., Murmur3) of
// (replica_id, namespace) for production.
let timeout_seed = u128::from(replica) ^ u128::from(namespace);
+ let prepare_queue_max = pipeline.prepare_queue_max();
Self {
cluster,
replica,
@@ -912,8 +923,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
last_timestamp: Cell::new(0),
last_prepare_checksum: Cell::new(0),
pipeline: RefCell::new(pipeline),
+ prepare_queue_max,
message_bus,
- loopback_queue: RefCell::new(VecDeque::with_capacity(PIPELINE_PREPARE_QUEUE_MAX)),
+ loopback_queue: RefCell::new(VecDeque::with_capacity(prepare_queue_max)),
start_view_change_from_all_replicas: RefCell::new(BitSet::with_capacity(REPLICAS_MAX)),
probe_attempts: Cell::new(0),
do_view_change_from_all_replicas: RefCell::new(dvc_quorum_array_empty()),
@@ -2436,13 +2448,13 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// incoming PrepareOk messages can be matched and commits can proceed.
if max_commit < new_op {
assert!(
- (new_op - max_commit) <= PIPELINE_PREPARE_QUEUE_MAX as u64,
+ (new_op - max_commit) <= self.prepare_queue_max as u64,
"view change: uncommitted range {}..={} ({} ops) exceeds pipeline capacity ({}); \
DVC winner claims more in-flight ops than the pipeline can hold",
max_commit + 1,
new_op,
new_op - max_commit,
- PIPELINE_PREPARE_QUEUE_MAX,
+ self.prepare_queue_max,
);
actions.push(VsrAction::RebuildPipeline {
from_op: max_commit + 1,
@@ -2554,7 +2566,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// TODO: Route SVC/DVC self-messages through loopback once VsrAction dispatch is implemented.
pub(crate) fn push_loopback(&self, message: Message<GenericHeader>) {
assert!(
- self.loopback_queue.borrow().len() < PIPELINE_PREPARE_QUEUE_MAX,
+ self.loopback_queue.borrow().len() < self.prepare_queue_max,
"loopback queue overflow: {} items",
self.loopback_queue.borrow().len()
);
@@ -2949,6 +2961,38 @@ mod pipeline_entry_tests {
let mut entry = PipelineEntry::new(header);
assert!(entry.take_reply_sender().is_none());
}
+
+ /// A pipeline configured deeper than [`PIPELINE_PREPARE_QUEUE_MAX`] must
+ /// verify a full queue instead of tripping the capacity assert: the bound
+ /// tracks the configured depth, not the default const.
+ #[test]
+ #[allow(clippy::cast_possible_truncation)]
+ fn given_prepare_queue_depth_above_default_when_verify_should_not_panic() {
+ let depth = PIPELINE_PREPARE_QUEUE_MAX * 2;
+ let mut pipeline = LocalPipeline::with_capacities(depth, depth * 2);
+
+ let mut parent = 0u128;
+ for op in 1..=depth as u64 {
+ let checksum = u128::from(op);
+ let header = PrepareHeader {
+ command: Command2::Prepare,
+ size: std::mem::size_of::<PrepareHeader>() as u32,
+ op,
+ parent,
+ checksum,
+ ..Default::default()
+ };
+ pipeline.push(PipelineEntry::new(header));
+ parent = checksum;
+ }
+
+ assert!(
+ pipeline.prepare_queue_full(),
+ "queue filled to the configured depth"
+ );
+ // Would panic on the old `len() <= PIPELINE_PREPARE_QUEUE_MAX` assert.
+ pipeline.verify();
+ }
}
#[cfg(test)]
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index 9c68f17ac..50d9e2260 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -52,6 +52,11 @@ pub trait Pipeline {
fn len(&self) -> usize;
+ /// In-flight prepare-queue capacity. `VsrConsensus` snapshots it at
+ /// construction to size the loopback queue and to bound the uncommitted
+ /// range a new primary may rebuild after a view change.
+ fn prepare_queue_max(&self) -> usize;
+
fn verify(&self);
/// True iff either queue carries `client_id`. Used by metadata-plane
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index e385566a4..f9b36e554 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -683,9 +683,9 @@ mod tests {
consensus.init();
// Diverge the frontiers: applied (commit_min=5) lags known-committed
- // (commit_max=13) by more than PIPELINE_PREPARE_QUEUE_MAX (8). op is at
- // 13 (>= commit_max), so the op clamp on the DVC commit is a no-op here
- // and the carried value is commit_max. The clamp itself is covered by
+ // (commit_max=13). op is at 13 (>= commit_max), so the op clamp on the
+ // DVC commit is a no-op here and the carried value is commit_max. The
+ // clamp itself is covered by
// `do_view_change_commit_clamped_to_op_when_commit_max_exceeds_op`.
consensus.advance_commit_max(13);
consensus.sequencer().set_sequence(13);
@@ -911,6 +911,74 @@ mod tests {
);
}
+ /// A DVC winner may claim an uncommitted range up to the *configured*
+ /// prepare depth. With a pipeline deeper than the default const, the new
+ /// primary schedules the rebuild rather than panicking on the old
+ /// `PIPELINE_PREPARE_QUEUE_MAX` bound.
+ #[test]
+ #[allow(clippy::cast_possible_truncation)]
+ fn given_view_change_range_above_default_when_complete_as_primary_should_rebuild() {
+ use iggy_binary_protocol::{DoViewChangeHeader, StartViewChangeHeader};
+
+ let depth = crate::PIPELINE_PREPARE_QUEUE_MAX * 2;
+ // Strictly above the default const, still within the configured depth.
+ let winner_op = (crate::PIPELINE_PREPARE_QUEUE_MAX + 8) as u64;
+
+ // 3 replicas, replica 0 is primary for view 3 (3 % 3 = 0).
+ let consensus = VsrConsensus::new(
+ 1,
+ 0,
+ 3,
+ 0,
+ NoopBus,
+ LocalPipeline::with_capacities(depth, depth * 2),
+ );
+ consensus.init();
+
+ // SVC from replica 1 moves replica 0 into view 3 and records its own DVC.
+ let svc = StartViewChangeHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 0,
+ size: 0,
+ view: 3,
+ release: 0,
+ command: Command2::StartViewChange,
+ replica: 1,
+ reserved_frame: [0; 66],
+ namespace: 0,
+ reserved: [0; 120],
+ };
+ let _ = consensus.handle_start_view_change(PlaneKind::Metadata, &svc);
+
+ // DVC from replica 2 claims a log head far past commit, forming quorum.
+ let dvc = DoViewChangeHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 0,
+ size: 0,
+ view: 3,
+ release: 0,
+ command: Command2::DoViewChange,
+ replica: 2,
+ reserved_frame: [0; 66],
+ op: winner_op,
+ commit: 0,
+ namespace: 0,
+ log_view: 0,
+ reserved: [0; 100],
+ };
+ let actions = consensus.handle_do_view_change(PlaneKind::Metadata, &dvc);
+
+ assert!(
+ actions.iter().any(|action| matches!(
+ action,
+ VsrAction::RebuildPipeline { from_op: 1, to_op } if *to_op == winner_op
+ )),
+ "expected RebuildPipeline over the full uncommitted range"
+ );
+ }
+
#[test]
fn send_prepare_ok_sends_to_bus_when_not_primary() {
// Replica 1, view 0; primary=0, so send_or_loopback takes bus path.