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 <[email protected]> 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.
