diff --git a/Cargo.lock b/Cargo.lock index 54c50c7cd7..a6dbe3746f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8488,6 +8488,7 @@ dependencies = [ "left-right", "message_bus", "paste", + "rmp", "rmp-serde", "serde", "server_common", diff --git a/Cargo.toml b/Cargo.toml index 4e4d18e400..c334d47d41 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -278,6 +278,7 @@ reqwest-tracing = "0.7.1" ring = "0.17.14" ringbuffer = "0.16.0" rmcp = "2.2.0" +rmp = "0.8.15" rmp-serde = "1.3.1" rolling-file = "0.2.0" rust-embed = "8.12.0" diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index e55c9c3b9f..dac2e38926 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -265,10 +265,20 @@ impl PipelineEntry { #[derive(Debug)] pub struct RequestEntry { pub message: Message, - // TODO: populate from monotonic clock at push, promote to `pub` for - // age-based filtering. Currently `0`; `pub(crate)` blocks sort-on-stub. - #[allow(dead_code)] - pub(crate) received_at: i64, + /// When the request was parked, in microseconds from the consensus-injected + /// clock ([`VsrConsensus::clock_realtime_micros`]). `0` until + /// [`VsrConsensus::push_queued_request`] stamps it, which is the only path + /// that parks an entry in production. + /// + /// Read against `clock_realtime_micros` at promotion for the queue wait: + /// what age-based shedding would filter on, and the queueing half of + /// end-to-end commit latency. + /// + /// Deliberately the plain clock read rather than + /// [`VsrConsensus::next_monotonic_timestamp`]: this must not consume the + /// prepare-stamping monotonic sequence, or parking a request would perturb + /// the timestamps replicated to every backup. + pub received_at: u64, /// In-process reply subscriber, carried through the queue so promotion /// can hand it to the pipeline entry (see [`PipelineEntry::with_sender`]). /// `None` = network path. Dropping a queued entry (view-change reset, @@ -310,6 +320,42 @@ impl RequestEntry { } } +impl VsrConsensus +where + B: MessageBus, + P: Pipeline, +{ + /// Park a request that could not take a prepare slot, stamping its arrival + /// time so the promotion side can measure how long it waited. + /// + /// # Errors + /// The entry itself, when the request queue is at its depth bound. + pub fn push_queued_request(&self, mut entry: RequestEntry) -> Result<(), RequestEntry> { + entry.received_at = self.clock_realtime_micros(); + self.pipeline.borrow_mut().push_request(entry) + } +} + +/// Outcome of [`VsrConsensus::rollback_pipelined_prepare`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PrepareRollback { + /// Sequencer, parent chain, and pipeline entry all restored. The op is free + /// again and the next request reuses it. + Unwound, + /// This replica never pre-advanced for the prepare, so there is nothing to + /// undo. A backup advances only AFTER its append succeeds. + NotPreAdvanced, + /// Refused: a sibling prepare was pipelined during the append await and is + /// already projected off the failed op, carrying `sequence` past it. + /// + /// Both outcomes are bad here and unwinding is the worse one: the sibling is + /// live in the pipeline and already chained to a prepare the WAL will never + /// hold, so rewinding underneath it would additionally hand its op number out + /// to the next request. The hole is not closable locally; a view change is + /// what repairs it. + Overtaken { sequence: u64 }, +} + /// Two-queue pipeline: in-flight prepares + buffered requests. #[derive(Debug)] pub struct LocalPipeline { @@ -484,6 +530,22 @@ impl LocalPipeline { self.prepare_queue.back() } + /// Drop the newest prepare when it is `op`, returning it. + /// + /// `None` (and no mutation) when the tail is a different op. The queue holds + /// a consecutive run, so removing anything but the tail would punch a hole in + /// it and break every `message_by_op` index computation; a caller whose op is + /// no longer the tail has been overtaken and must not unwind. + /// + /// The one caller is the journal-append rollback + /// ([`VsrConsensus::rollback_pipelined_prepare`]). + pub fn remove_prepare_tail(&mut self, op: u64) -> Option { + if self.prepare_queue.back()?.header.op != op { + return None; + } + self.prepare_queue.pop_back() + } + /// Find a message by op number and checksum (immutable). // op - head_op is bounded by the configured prepare-queue depth; index always fits in usize. #[must_use] @@ -628,6 +690,10 @@ impl Pipeline for LocalPipeline { Self::pop_message(self) } + fn remove_tail(&mut self, op: u64) -> Option { + Self::remove_prepare_tail(self, op) + } + fn clear(&mut self) { Self::clear(self); } @@ -687,6 +753,10 @@ impl Pipeline for LocalPipeline { fn pop_request(&mut self) -> Option { Self::pop_request(self) } + + fn request_queue_len(&self) -> usize { + Self::request_queue_len(self) + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -1177,7 +1247,12 @@ impl> VsrConsensus { self.status.set(Status::Normal); let mut timeouts = self.timeouts.borrow_mut(); if self.is_primary() { - timeouts.start(TimeoutKind::Prepare); + // Prepare is deliberately NOT armed here: a fresh primary has an + // empty pipeline and nothing to retransmit, and the timer is owned by + // the pipeline's edges from this point on ("ticking iff the pipeline + // is non-empty", see `sync_prepare_timeout`). The first + // `push_prepare_entry` arms it, timed from that push rather than from + // boot. timeouts.start(TimeoutKind::CommitMessage); } else { timeouts.start(TimeoutKind::NormalHeartbeat); @@ -1418,17 +1493,117 @@ impl> VsrConsensus { self.status.get() } - // TODO(hubcio): returning &RefCell

leaks interior mutability - callers - // could hold a Ref/RefMut across an .await and cause a runtime panic. - // We had this problem with slab + ECS. + /// Run `f` against the pipeline. + /// + /// The borrow cannot escape `f`, so it cannot be held across an `.await` and + /// alias a sibling task's `borrow_mut` into a `BorrowMutError` panic. Same + /// shape, and the same reason, as `IggyPartitions::with_partition`. Prefer + /// the named accessors below; this is for the few callers that need several + /// reads under one borrow. + pub fn with_pipeline(&self, f: impl FnOnce(&P) -> R) -> R { + f(&self.pipeline.borrow()) + } + + /// [`Self::with_pipeline`] for a mutating operation. + /// + /// Callers that pop or clear should prefer [`Self::pop_committed_prepare`] / + /// [`Self::clear_pipeline`], which also keep the prepare timeout's + /// ticking-iff-non-empty invariant. This form does not. + pub fn with_pipeline_mut(&self, f: impl FnOnce(&mut P) -> R) -> R { + f(&mut self.pipeline.borrow_mut()) + } + + /// Whether the prepare queue is at its depth bound; callers route to + /// [`Self::push_queued_request`] on `true`. + #[must_use] + pub fn pipeline_is_full(&self) -> bool { + self.pipeline.borrow().is_full() + } + + #[must_use] + pub fn pipeline_is_empty(&self) -> bool { + self.pipeline.borrow().is_empty() + } + + /// In-flight prepare count. + #[must_use] + pub fn pipeline_len(&self) -> usize { + self.pipeline.borrow().len() + } + + /// Requests parked waiting for a prepare slot. + #[must_use] + pub fn request_queue_len(&self) -> usize { + self.pipeline.borrow().request_queue_len() + } + + /// Whether either queue already carries a request from `client_id`: the + /// metadata plane's in-flight dedup. + #[must_use] + pub fn pipeline_has_message_from_client(&self, client_id: u128) -> bool { + self.pipeline.borrow().has_message_from_client(client_id) + } + + /// Header of the oldest in-flight prepare. #[must_use] - pub const fn pipeline(&self) -> &RefCell

{ - &self.pipeline + pub fn pipeline_head_header(&self) -> Option { + self.pipeline.borrow().head().map(|entry| entry.header) } + /// Whether an in-flight prepare matches `(op, checksum)`: the ack paths' test + /// that the frame they are about to count still describes a live entry. #[must_use] - pub const fn pipeline_mut(&mut self) -> &mut RefCell

{ - &mut self.pipeline + pub fn pipeline_holds_entry(&self, op: u64, checksum: u128) -> bool { + self.pipeline + .borrow() + .entry_by_op_and_checksum(op, checksum) + .is_some() + } + + /// Promote the oldest parked request, if any. + pub fn pop_queued_request(&self) -> Option { + self.pipeline.borrow_mut().pop_request() + } + + /// Pop the pipeline head once its op has committed, keeping the prepare + /// timeout's lifecycle in step. + /// + /// The timeout measures the age of the oldest un-acked prepare, so draining + /// the head has to either stop it (nothing left to retransmit) or restart it + /// (the next entry becomes the oldest, and it must be timed from now rather + /// than inheriting the drained entry's elapsed ticks). Arming happens in + /// [`Self::push_prepare_entry`]; between the two the invariant is "ticking + /// iff the pipeline is non-empty". + pub fn pop_committed_prepare(&self) -> Option { + let popped = self.pipeline.borrow_mut().pop(); + if popped.is_some() { + self.sync_prepare_timeout(); + } + popped + } + + /// Drop every in-flight prepare and parked request, and disarm the prepare + /// timeout with them (a view change re-prepares from the new primary; there + /// is nothing left here to retransmit). + pub fn clear_pipeline(&self) { + self.pipeline.borrow_mut().clear(); + self.timeouts.borrow_mut().stop(TimeoutKind::Prepare); + } + + /// Re-establish "prepare timeout ticking iff the pipeline is non-empty". + /// + /// Stops the timer on an empty pipeline; otherwise restarts it so it times + /// the current oldest entry from now. Exposed for the plane-side drains that + /// pop through [`Pipeline`] directly (`drain_committable_prefix`) rather than + /// through [`Self::pop_committed_prepare`]. + pub fn sync_prepare_timeout(&self) { + let empty = self.pipeline.borrow().is_empty(); + let mut timeouts = self.timeouts.borrow_mut(); + if empty { + timeouts.stop(TimeoutKind::Prepare); + } else { + timeouts.reset(TimeoutKind::Prepare); + } } /// Push a pre-built [`PipelineEntry`]; start prepare timeout if idle. @@ -1484,6 +1659,47 @@ impl> VsrConsensus { } } + /// Undo the [`Self::push_prepare_entry`] pre-advance for a prepare whose + /// journal append failed, so the op it claimed is handed back. + /// + /// The pre-advance runs the sequencer ahead of the WAL on purpose, so that a + /// sibling `on_request` racing the append await cannot project a duplicate op. + /// The cost is that a failed append leaves the op claimed with nothing durable + /// behind it: the next prepare chains off a phantom, the WAL takes a permanent + /// hole at that op, and the divergence rides the handoff bundle out to peers. + /// + /// Rolls the sequencer back to `header.op - 1` and the parent chain to + /// `header.parent`, both of which the header records from the moment it was + /// projected, and drops the pipeline entry so the reclaimed op is free rather + /// than colliding with a live entry. Dropping the entry drops its reply + /// sender, so a waiting client observes `Canceled` instead of hanging until + /// the request times out. + /// + /// The observed prepare timestamp is deliberately NOT rolled back: + /// [`Self::next_monotonic_timestamp`] only ever needs a lower bound, and + /// lowering it back could re-stamp a value a peer already observed. + /// + /// See [`PrepareRollback`] for the outcomes. + pub fn rollback_pipelined_prepare(&self, header: &PrepareHeader) -> PrepareRollback { + if !self.is_primary() { + return PrepareRollback::NotPreAdvanced; + } + let sequence = self.sequencer.current_sequence(); + if sequence != header.op { + return PrepareRollback::Overtaken { sequence }; + } + let removed = self.pipeline.borrow_mut().remove_tail(header.op); + debug_assert!( + removed.is_some(), + "sequencer at op {} but the pipeline tail is not that op", + header.op + ); + drop(removed); + self.sequencer.set_sequence(header.op.saturating_sub(1)); + self.set_last_prepare_checksum(header.parent); + PrepareRollback::Unwound + } + /// Push `message` with in-band reply subscriber. /// /// Like [`Consensus::pipeline_message`], but entry is built via @@ -1965,8 +2181,31 @@ impl> VsrConsensus { /// Advance to `view + 1` and start a view change (own SVC counted). fn start_election(&self, plane: PlaneKind, reason: ViewChangeReason) -> Vec { + self.enter_view_change(plane, self.view.get() + 1, reason) + } + + /// Enter `Status::ViewChange` at `new_view`: count this replica's own SVC, + /// arm the view-change timers, and schedule the SVC broadcast. + /// + /// The own-SVC bookkeeping is a direct insert rather than an SVC delivered to + /// self through the loopback. Deciding to change view is a local state + /// transition, not a message this replica happens to address to itself, and + /// it has to be atomic with the view/status writes above it: a self-message + /// would land on a later pump drain, leaving a window where the replica has + /// entered a view change without counting itself. On a solo group that window + /// IS the whole quorum. `PrepareOk` loops through the loopback because it + /// genuinely is a message to a peer that happens to be this replica. + /// + /// The three callers that used to inline this sequence were the actual + /// duplication: an election timeout, an SVC for a higher view, and a DVC for + /// a higher view, differing only in `reason`. + fn enter_view_change( + &self, + plane: PlaneKind, + new_view: u32, + reason: ViewChangeReason, + ) -> Vec { let old_view = self.view.get(); - let new_view = old_view + 1; self.view.set(new_view); self.status.set(Status::ViewChange); @@ -2144,17 +2383,17 @@ impl> VsrConsensus { /// Only acts on the primary in normal status with a non-empty pipeline. /// Resets the timeout with backoff on each firing. fn handle_prepare_timeout(&self) -> Vec { - // TODO(prepare-timeout): tighten the timer lifecycle: disarm in - // the ack path the moment quorum drains the pipeline and rearm - // for the next-oldest prepare when one commits with others still - // pending, giving the invariant "ticking iff pipeline non-empty" - // and a timeout that always measures the current oldest - // prepare's age. Ours arms once per idle->busy transition and - // disarms lazily below, so a prepare pushed late into an armed - // window can be retransmitted before it is `PREPARE_TICKS` old. - // Also worth special-casing "all remote acks present, own - // journal write is the laggard" by retrying the local write - // instead of retransmitting. + // The timer's lifecycle is maintained at the pipeline's own edges + // (`push_prepare_entry` arms, `sync_prepare_timeout` disarms on empty and + // restarts on a new head), so by the time this fires the timer is already + // measuring the current oldest prepare rather than inheriting a drained + // entry's elapsed ticks. The stop below is now a backstop for a pipeline + // emptied by a path that skipped that maintenance, not the primary + // disarm. + // + // TODO(prepare-timeout): special-case "all remote acks present, own + // journal write is the laggard" by retrying the local write instead of + // retransmitting to peers that already acked. // // Every early return below must stop or back off the timeout. // `fired()` stays true until the timer is rearmed, so returning @@ -2269,47 +2508,11 @@ impl> VsrConsensus { // If SVC is for a higher view, advance to that view if msg_view > self.view.get() { - let old_view = self.view.get(); - self.view.set(msg_view); - self.status.set(Status::ViewChange); - self.reset_view_change_state(); - self.sent_own_start_view_change.set(true); - self.start_view_change_from_all_replicas - .borrow_mut() - .insert(self.replica as usize); - - // Update timeouts - { - let mut timeouts = self.timeouts.borrow_mut(); - timeouts.stop(TimeoutKind::NormalHeartbeat); - timeouts.start(TimeoutKind::StartViewChangeMessage); - timeouts.start(TimeoutKind::ViewChangeStatus); - timeouts.start(TimeoutKind::RequestStartViewMessage); - } - - emit_sim_event( - SimEventKind::ViewChangeStarted, - &ViewChangeLogEvent { - replica: ReplicaLogContext::from_consensus(self, plane), - old_view, - new_view: msg_view, - reason: ViewChangeReason::ReceivedStartViewChange, - }, - ); - - // Send our own SVC - let action = VsrAction::SendStartViewChange { - view: msg_view, - group: self.group, - }; - emit_sim_event( - SimEventKind::ControlMessageScheduled, - &ControlActionLogEvent::from_vsr_action( - ReplicaLogContext::from_consensus(self, plane), - &action, - ), - ); - actions.push(action); + actions.extend(self.enter_view_change( + plane, + msg_view, + ViewChangeReason::ReceivedStartViewChange, + )); } // Record the SVC from sender @@ -2471,47 +2674,11 @@ impl> VsrConsensus { // If DVC is for a higher view, advance to that view if msg_view > self.view.get() { - let old_view = self.view.get(); - self.view.set(msg_view); - self.status.set(Status::ViewChange); - self.reset_view_change_state(); - self.sent_own_start_view_change.set(true); - self.start_view_change_from_all_replicas - .borrow_mut() - .insert(self.replica as usize); - - // Update timeouts - { - let mut timeouts = self.timeouts.borrow_mut(); - timeouts.stop(TimeoutKind::NormalHeartbeat); - timeouts.start(TimeoutKind::StartViewChangeMessage); - timeouts.start(TimeoutKind::ViewChangeStatus); - timeouts.start(TimeoutKind::RequestStartViewMessage); - } - - emit_sim_event( - SimEventKind::ViewChangeStarted, - &ViewChangeLogEvent { - replica: ReplicaLogContext::from_consensus(self, plane), - old_view, - new_view: msg_view, - reason: ViewChangeReason::ReceivedDoViewChange, - }, - ); - - // Send our own SVC - let action = VsrAction::SendStartViewChange { - view: msg_view, - group: self.group, - }; - emit_sim_event( - SimEventKind::ControlMessageScheduled, - &ControlActionLogEvent::from_vsr_action( - ReplicaLogContext::from_consensus(self, plane), - &action, - ), - ); - actions.push(action); + actions.extend(self.enter_view_change( + plane, + msg_view, + ViewChangeReason::ReceivedDoViewChange, + )); } // Only the primary candidate processes DVCs for quorum @@ -3428,8 +3595,13 @@ impl> VsrConsensus { /// Enqueue a self-addressed message for processing in the next loopback drain. /// - /// Currently only `PrepareOk` messages are routed here (via `send_or_loopback`). - // TODO: Route SVC/DVC self-messages through loopback once VsrAction dispatch is implemented. + /// Only `PrepareOk` reaches here (via `send_or_loopback`), and deliberately: + /// it is a message to a peer that happens to be this replica, so a one-drain + /// delay costs nothing. A replica's own SVC/DVC is not that shape -- it is the + /// local decision to change view, recorded synchronously with the view and + /// status writes in [`Self::enter_view_change`]. Routing it here instead would + /// leave a window where the replica has entered a view change without counting + /// itself, which on a solo group is the entire quorum. pub(crate) fn push_loopback(&self, message: Message) { assert!( self.loopback_queue.borrow().len() < self.prepare_queue_max, @@ -4255,7 +4427,7 @@ mod timestamp_clamp_tests { } #[cfg(test)] -mod state_transfer_stage_tests { +mod vsr_consensus_tests { use super::*; #[test] @@ -4465,6 +4637,241 @@ mod state_transfer_stage_tests { let _ = consensus.handle_normal_heartbeat_timeout(PlaneKind::Metadata); assert_eq!(consensus.status(), Status::Recovering); } + + /// A prepare as the primary projects one: `parent` is the chain value the + /// sequencer stood on before this op, which is exactly what a rollback restores. + #[allow(clippy::cast_possible_truncation)] + fn projected_prepare(op: u64, parent: u128) -> Message { + Message::::new(size_of::()).transmute_header(|_, new| { + *new = PrepareHeader { + command: Command2::Prepare, + size: size_of::() as u32, + op, + parent, + ..Default::default() + }; + }) + } + + /// Pipeline a prepare the way `on_request` does, pre-advancing the sequencer + /// and the parent chain ahead of the journal append. + fn pipeline(consensus: &VsrConsensus, message: &Message) { + consensus.pipeline_message(PlaneKind::Metadata, message); + } + + use crate::drain_committable_prefix; + use iggy_binary_protocol::Operation; + + /// Clock frozen at a fixed instant, so a stamp read off it is assertable. + struct FrozenClock(u64); + + impl clock::Clock for FrozenClock { + type Realtime = IggyTimestamp; + + fn realtime(&self) -> Self::Realtime { + IggyTimestamp::from(self.0) + } + } + + /// A client request as the admission path hands it to the request queue. + #[allow(clippy::cast_possible_truncation)] + fn client_request(client: u128) -> Message { + Message::::new(size_of::()).transmute_header( + |_, new| { + *new = RoutedRequestHeader { + command: Command2::Request, + size: size_of::() as u32, + client, + session: 1, + request: 1, + operation: Operation::CreateStream, + ..Default::default() + }; + }, + ) + } + + /// Whether the prepare retransmit timer is armed. + fn prepare_ticking(consensus: &VsrConsensus) -> bool { + consensus.timeouts.borrow().is_ticking(TimeoutKind::Prepare) + } + + #[test] + fn prepare_timeout_ticks_exactly_while_the_pipeline_is_non_empty() { + // The lifecycle invariant: armed by the first push, disarmed the moment + // the pipeline drains. Previously it armed at `init` on an empty pipeline + // and only disarmed lazily, when the timeout itself fired and found + // nothing to retransmit. + let consensus = VsrConsensus::new(1, 0, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + assert!( + !prepare_ticking(&consensus), + "a fresh primary has nothing to retransmit" + ); + + let first = projected_prepare(1, 0); + pipeline(&consensus, &first); + assert!(prepare_ticking(&consensus), "the first push arms the timer"); + + let second = projected_prepare(2, 0); + pipeline(&consensus, &second); + + // Committing the head leaves op 2 in flight, so the timer stays armed -- + // now measuring op 2 rather than carrying op 1's elapsed ticks. + consensus.advance_commit_max(1); + assert_eq!(drain_committable_prefix(&consensus).len(), 1); + assert!( + prepare_ticking(&consensus), + "a remaining prepare keeps the timer armed" + ); + + consensus.advance_commit_max(2); + assert_eq!(drain_committable_prefix(&consensus).len(), 1); + assert!( + !prepare_ticking(&consensus), + "draining the last prepare disarms the timer without waiting for it to fire" + ); + } + + #[test] + fn parking_a_request_stamps_its_arrival_from_the_injected_clock() { + // The queue wait is `clock_realtime_micros() - received_at` at promotion, + // so the stamp has to come from the same injected clock (virtual under + // the simulator) and must not consume the prepare-stamping monotonic + // sequence. + const NOW: u64 = 4_242; + let consensus = VsrConsensus::with_clock( + 1, + 0, + 3, + 0, + StageNoopBus, + LocalPipeline::new(), + ConsensusClock::new(Rc::new(FrozenClock(NOW))), + ); + consensus.init(); + let before = consensus.next_monotonic_timestamp(); + + consensus + .push_queued_request(RequestEntry::new(client_request(1))) + .expect("empty request queue accepts one"); + + let entry = consensus + .pop_queued_request() + .expect("the just-parked request"); + assert_eq!(entry.received_at, NOW, "stamped from the injected clock"); + assert_eq!( + consensus.next_monotonic_timestamp(), + before + 1, + "parking must not consume the prepare-stamping monotonic sequence" + ); + } + + #[test] + fn clearing_the_pipeline_disarms_the_prepare_timeout() { + // A view change re-prepares from the new primary, so nothing left here is + // worth retransmitting. + let consensus = VsrConsensus::new(1, 0, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + pipeline(&consensus, &projected_prepare(1, 0)); + assert!(prepare_ticking(&consensus)); + + consensus.clear_pipeline(); + assert!(consensus.pipeline_is_empty()); + assert!(!prepare_ticking(&consensus)); + } + + #[test] + fn rollback_hands_back_the_op_a_failed_append_claimed() { + // Without this, the sequencer keeps claiming op 8 while the WAL stops at 7: + // the next request projects op 9 over a hole that no repair path refills. + const PARENT: u128 = 0xfeed; + let consensus = VsrConsensus::new(1, 0, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + consensus.sequencer().set_sequence(7); + consensus.set_last_prepare_checksum(PARENT); + + let message = projected_prepare(8, PARENT); + pipeline(&consensus, &message); + assert_eq!(consensus.sequencer().current_sequence(), 8); + + assert_eq!( + consensus.rollback_pipelined_prepare(message.header()), + PrepareRollback::Unwound + ); + assert_eq!(consensus.sequencer().current_sequence(), 7); + assert_eq!(consensus.last_prepare_checksum(), PARENT); + assert!( + consensus.pipeline_is_empty(), + "the reclaimed op must not stay live in the pipeline; the next request reuses it" + ); + } + + #[test] + fn rollback_is_refused_once_a_sibling_took_the_next_op() { + // The race the pre-advance exists for: a request pipelined while op 8's + // append was in flight already chained op 9 off it. Rewinding would hand + // op 9's number back out while op 9 is still live, so the refusal is the + // safe answer and the caller escalates. + let consensus = VsrConsensus::new(1, 0, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + consensus.sequencer().set_sequence(7); + + let first = projected_prepare(8, 0); + pipeline(&consensus, &first); + let sibling = projected_prepare(9, 0); + pipeline(&consensus, &sibling); + + assert_eq!( + consensus.rollback_pipelined_prepare(first.header()), + PrepareRollback::Overtaken { sequence: 9 } + ); + assert_eq!(consensus.sequencer().current_sequence(), 9); + assert_eq!( + consensus.pipeline_len(), + 2, + "a refused rollback must not touch the pipeline" + ); + } + + #[test] + fn rollback_finds_nothing_to_undo_on_a_backup() { + // Replica 1 is a backup at view 0, and a backup advances only after its + // append succeeds, so a failed append left nothing pre-advanced. + let consensus = VsrConsensus::new(1, 1, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + consensus.sequencer().set_sequence(7); + + let message = projected_prepare(8, 0); + assert_eq!( + consensus.rollback_pipelined_prepare(message.header()), + PrepareRollback::NotPreAdvanced + ); + assert_eq!(consensus.sequencer().current_sequence(), 7); + } + + #[test] + fn rollback_cancels_the_client_awaiting_the_dropped_prepare() { + // The write was never made durable, so the caller must learn it failed + // instead of parking until its request times out. + use futures::FutureExt as _; + + let consensus = VsrConsensus::new(1, 0, 3, 0, StageNoopBus, LocalPipeline::new()); + consensus.init(); + consensus.sequencer().set_sequence(7); + + let message = projected_prepare(8, 0); + let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &message); + + assert_eq!( + consensus.rollback_pipelined_prepare(message.header()), + PrepareRollback::Unwound + ); + assert!( + matches!(receiver.now_or_never(), Some(Err(crate::oneshot::Canceled))), + "dropping the entry must cancel its awaiter" + ); + } } #[cfg(test)] diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs index 02cca677b0..f81eade31f 100644 --- a/core/consensus/src/lib.rs +++ b/core/consensus/src/lib.rs @@ -34,6 +34,10 @@ pub trait Pipeline { fn pop(&mut self) -> Option; + /// Drop the newest entry when it is `op`, returning it; `None` and no + /// mutation otherwise. Unwinds a push that turned out not to be durable. + fn remove_tail(&mut self, op: u64) -> Option; + fn clear(&mut self); fn entry_by_op(&self, op: u64) -> Option<&Self::Entry>; @@ -52,6 +56,9 @@ pub trait Pipeline { fn len(&self) -> usize; + /// Requests parked waiting for a prepare slot (the second queue). + fn request_queue_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. diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs index 329ca18d94..3dd51405b2 100644 --- a/core/consensus/src/metadata_helpers.rs +++ b/core/consensus/src/metadata_helpers.rs @@ -102,11 +102,7 @@ where { // In-flight dedup: a live prepare from this client absorbs the retry. // Pump delivers the reply at commit. - if consensus - .pipeline() - .borrow() - .has_message_from_client(client_id) - { + if consensus.pipeline_has_message_from_client(client_id) { tracing::debug!( client_id, request, @@ -262,11 +258,7 @@ where P: Pipeline, { // In-flight dedup. - if consensus - .pipeline() - .borrow() - .has_message_from_client(client_id) - { + if consensus.pipeline_has_message_from_client(client_id) { tracing::debug!(client_id, "register_preflight: in-flight prepare, drop"); return false; } diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index 87f3b65e6e..344336bcf9 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -415,15 +415,15 @@ where return false; } - let pipeline = consensus.pipeline().borrow(); let mut new_commit = consensus.commit_max(); - while let Some(entry) = pipeline.entry_by_op(new_commit + 1) { - if !entry.ok_quorum_received { - break; + consensus.with_pipeline(|pipeline| { + while let Some(entry) = pipeline.entry_by_op(new_commit + 1) { + if !entry.ok_quorum_received { + break; + } + new_commit += 1; } - new_commit += 1; - } - drop(pipeline); + }); if new_commit > consensus.commit_max() { consensus.advance_commit_max(new_commit); @@ -447,17 +447,27 @@ where { let commit = consensus.commit_max(); let mut drained = Vec::new(); - let mut pipeline = consensus.pipeline().borrow_mut(); - while let Some(head_op) = pipeline.head().map(|entry| entry.header.op) { - if head_op > commit { - break; + consensus.with_pipeline_mut(|pipeline| { + while let Some(head_op) = pipeline.head().map(|entry| entry.header.op) { + if head_op > commit { + break; + } + + let entry = pipeline + .pop() + .expect("drain_committable_prefix: head exists"); + drained.push(entry); } + }); - let entry = pipeline - .pop() - .expect("drain_committable_prefix: head exists"); - drained.push(entry); + // Popping through the pipeline directly bypasses + // `VsrConsensus::pop_committed_prepare`, so re-establish the prepare + // timeout's ticking-iff-non-empty invariant here: an emptied pipeline + // disarms it, and a remaining head becomes the entry the timer measures, + // timed from now rather than inheriting the drained entry's elapsed ticks. + if !drained.is_empty() { + consensus.sync_prepare_timeout(); } drained @@ -478,10 +488,8 @@ where P: Pipeline, { let commit = consensus.commit_max(); - let pipeline = consensus.pipeline().borrow(); - pipeline - .head() - .map(|entry| entry.header) + consensus + .pipeline_head_header() .filter(|header| header.op <= commit) } @@ -1971,11 +1979,7 @@ mod tests { assert_eq!(drained_ops, vec![5, 6]); assert_eq!( - consensus - .pipeline() - .borrow() - .head() - .map(|entry| entry.header.op), + consensus.pipeline_head_header().map(|header| header.op), Some(7) ); } diff --git a/core/metadata/Cargo.toml b/core/metadata/Cargo.toml index cd83798bc6..bfb94458e9 100644 --- a/core/metadata/Cargo.toml +++ b/core/metadata/Cargo.toml @@ -48,6 +48,7 @@ journal = { workspace = true } left-right = { workspace = true } message_bus = { workspace = true } paste = { workspace = true } +rmp = { workspace = true } rmp-serde = { workspace = true } serde = { workspace = true, features = ["derive"] } server_common = { workspace = true } diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index c4bb620e08..0beff44193 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -27,8 +27,8 @@ use crate::stm::{ConsensusGroupAllocator, StateMachine}; use consensus::{ CLIENTS_TABLE_MAX, Canceled, ClientTable, ClientTableSnapshot, CommitLogEvent, CommitReply, Consensus, EvictionContext, Pipeline, PipelineEntry, Plane, PlaneIdentity, PlaneKind, - PreflightOutcome, Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, - VsrConsensus, ack_preflight, ack_quorum_reached, apply_preflight_consensus_plane, + PreflightOutcome, PrepareRollback, Project, ReplicaLogContext, RequestLogEvent, Sequencer, + SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached, apply_preflight_consensus_plane, build_eviction_message, build_reply_message, build_reply_message_with, build_result_rejection_reply, emit_sim_event, fence_old_prepare_by_commit, is_caught_up_primary, panic_if_hash_chain_would_break_in_same_view, peek_committable_head, @@ -44,7 +44,8 @@ use iggy_binary_protocol::requests::topics::CreateTopicRequest as WireCreateTopi use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as PersistedCreateTopicRequest; use iggy_binary_protocol::{ Command2, ConsensusHeader, EvictionReason, GenericHeader, Operation, PrepareHeader, - PrepareOkHeader, ReplyHeader, RoutedRequestHeader, WireDecode, WireEncode, WireName, + PrepareOkHeader, ProtocolVersion, ReplyHeader, RoutedRequestHeader, WireDecode, WireEncode, + WireName, }; use iggy_common::IggyError; use iggy_common::UserId; @@ -190,15 +191,17 @@ impl IggySnapshot { /// /// The checksum comes from the file's bytes, never from re-encoding what was /// decoded: the pairing must survive a schema change. Adding a trailing - /// `#[serde(default)]` field is the repo's forward-compatible migration, and an - /// older file re-encodes with one MORE msgpack array element after it, so a - /// re-encode checksum would diverge on the first boot of the new build and refuse - /// every checkpointed node with its WAL prefix already drained. + /// `#[serde(default)]` field is the repo's forward-compatible migration (see + /// [`SNAPSHOT_FORMAT_VERSION`](crate::stm::snapshot::SNAPSHOT_FORMAT_VERSION) for + /// the rules), and an older file re-encodes with one MORE msgpack array element + /// after it, so a re-encode checksum would diverge on the first boot of the new + /// build and refuse every checkpointed node with its WAL prefix already drained. /// /// # Errors /// `SnapshotError::ChecksumMismatch` if the file carries an integrity trailer that - /// does not match its payload, or `SnapshotError` if the file cannot be read or - /// deserialized. + /// does not match its payload, `SnapshotError::UnsupportedFormatVersion` if it was + /// written in a format version this build does not read, or `SnapshotError` if the + /// file cannot be read or deserialized. pub fn load(path: &Path) -> Result<(Self, u128), SnapshotError> { let data = std::fs::read(path)?; let (payload, checksum) = split_trailer(&data, path)?; @@ -996,11 +999,8 @@ where // Two-queue admission: prepare slot then project+replicate; prepare // full + request room then buffer; both full then drop+warn (SDK // retries via read-timeout). - if consensus.pipeline().borrow().is_full() { - let push_result = consensus - .pipeline() - .borrow_mut() - .push_request(consensus::RequestEntry::new(message)); + if consensus.pipeline_is_full() { + let push_result = consensus.push_queued_request(consensus::RequestEntry::new(message)); if push_result.is_err() { warn!( target: "iggy.metadata.diag", @@ -1198,19 +1198,14 @@ where // violates VSR tail-ahead-of-head, recoverable only via hash-chain // fence + view change (burns a view). // - // TODO(hubcio): the primary path violates the invariant in the - // comment above. `consensus::impls::push_prepare_entry` pre-advances - // `sequencer.set_sequence(header.op)` and - // `set_last_prepare_checksum(header.checksum)` BEFORE this append. - // If the append below returns `Err`, sequencer + checksum stay - // advanced while the WAL holds no matching entry: the next prepare - // chains off a phantom op, cluster state diverges, and the - // `MetadataHandoff::Waiter` factory bundle propagates the divergence - // to peers. Fix: rollback `sequencer.set_sequence` + - // `set_last_prepare_checksum` to their captured prior values on - // append failure (preferred per CLAUDE.md "no panics in libraries"), - // or abort the shard. + // On the primary the pre-advance in `push_prepare_entry` already claimed + // this op, so a failed append has to hand it back or the next prepare + // chains off a phantom (see `rollback_pipelined_prepare`). The rollback is + // refused when a sibling prepare was pipelined during the await and has + // already been projected off this op; that log cannot be repaired from + // here, so it is reported and left to a view change. if let Err(e) = journal.handle().append(message.clone()).await { + let rollback = consensus.rollback_pipelined_prepare(&header); error!( target: "iggy.metadata.diag", plane = "metadata", @@ -1218,8 +1213,20 @@ where op = header.op, operation = ?header.operation, error = %e, + rollback = ?rollback, "journal append failed" ); + if let PrepareRollback::Overtaken { sequence } = rollback { + error!( + target: "iggy.metadata.diag", + plane = "metadata", + replica_id = consensus.replica(), + op = header.op, + sequence, + "journal append failed after a sibling prepare was projected off this op; \ + the local log has a hole this replica cannot close, awaiting view change" + ); + } return; } @@ -1269,11 +1276,7 @@ where } { - let pipeline = consensus.pipeline().borrow(); - if pipeline - .entry_by_op_and_checksum(header.op, header.prepare_checksum) - .is_none() - { + if !consensus.pipeline_holds_entry(header.op, header.prepare_checksum) { debug!( target: "iggy.metadata.diag", plane = "metadata", @@ -1578,7 +1581,7 @@ where /// # Panics /// If called on a shard without consensus (state transfer is a shard-0 /// concern). - #[allow(clippy::future_not_send)] + #[allow(clippy::future_not_send, clippy::too_many_lines)] pub async fn install_state_transfer( &self, snapshot_bytes: &[u8], @@ -1594,9 +1597,20 @@ where .as_ref() .expect("install_state_transfer: consensus only exists on shard 0"); + // Refuses a format version this build does not read, ahead of every frontier + // move below: the bytes come from a peer, so its build picked the shape. let snapshot = IggySnapshot::decode(snapshot_bytes)?; let snapshot_seq = snapshot.sequence_number(); + // The one place a snapshot crosses builds, so the only place the release + // stamp answers a question the local logs cannot. + tracing::info!( + snapshot_seq, + format_version = snapshot.snapshot().version, + release_format = %ProtocolVersion(snapshot.snapshot().release_format), + "decoded a transferred metadata snapshot" + ); + // Manifest coherence. `commit_op` and `table_frontier` arrive from the // serving peer and are applied to THIS replica's frontiers, so a // malformed descriptor would move them somewhere the artifacts do not @@ -1887,11 +1901,7 @@ where // commit a second register and bump the epoch past the first reply's. // Surface pre-synthesis. Scans both the prepare queue and the request // queue, so a register absorbed below dedups its own replays. - if consensus - .pipeline() - .borrow() - .has_message_from_client(client_id) - { + if consensus.pipeline_has_message_from_client(client_id) { return Err(MetadataSubmitError::InProgress); } @@ -1920,14 +1930,9 @@ where // re-runs `register_preflight` and so applies the ownership gate) as // soon as the in-flight batch drains, and the await below resolves // exactly like the direct dispatch would. - if !is_caught_up_primary(consensus) || consensus.pipeline().borrow().is_full() { + if !is_caught_up_primary(consensus) || consensus.pipeline_is_full() { let (entry, receiver) = consensus::RequestEntry::with_subscriber(request); - if consensus - .pipeline() - .borrow_mut() - .push_request(entry) - .is_err() - { + if consensus.push_queued_request(entry).is_err() { // Both queues full: honest terminal backpressure. return Err(MetadataSubmitError::PipelineFull); } @@ -2095,11 +2100,7 @@ where return Err(MetadataSubmitError::NotPrimary); } - if consensus - .pipeline() - .borrow() - .has_message_from_client(client_id) - { + if consensus.pipeline_has_message_from_client(client_id) { return Err(MetadataSubmitError::InProgress); } @@ -2115,14 +2116,9 @@ where // Prepare queue full: absorb into the request queue with this // caller's reply subscriber, promoted as // commits free slots. - if consensus.pipeline().borrow().is_full() { + if consensus.pipeline_is_full() { let (entry, receiver) = consensus::RequestEntry::with_subscriber(request); - if consensus - .pipeline() - .borrow_mut() - .push_request(entry) - .is_err() - { + if consensus.push_queued_request(entry).is_err() { return Err(MetadataSubmitError::PipelineFull); } return match receiver.await { @@ -2212,14 +2208,10 @@ where }, ); } - if consensus - .pipeline() - .borrow() - .has_message_from_client(internal_client_id) - { + if consensus.pipeline_has_message_from_client(internal_client_id) { return Err(MetadataSubmitError::InProgress); } - if consensus.pipeline().borrow().is_full() { + if consensus.pipeline_is_full() { return Err(MetadataSubmitError::PipelineFull); } @@ -2295,7 +2287,7 @@ where ); } - if consensus.pipeline().borrow().is_full() { + if consensus.pipeline_is_full() { return Err(MetadataSubmitError::PipelineFull); } @@ -2418,14 +2410,9 @@ where // with the committed reply. Only a full request queue is terminal // (`TransientNotAccepted`, re-issuable anywhere: the request never // entered a queue). - if consensus.pipeline().borrow().is_full() { + if consensus.pipeline_is_full() { let (entry, receiver) = consensus::RequestEntry::with_subscriber(message); - if consensus - .pipeline() - .borrow_mut() - .push_request(entry) - .is_err() - { + if consensus.push_queued_request(entry).is_err() { return Ok(build_result_rejection_reply( &request_header, consensus.commit_max(), @@ -2565,8 +2552,7 @@ where // Snapshot durable, self-unacked pending ops, dropping the pipeline and // journal borrows before the `send_prepare_ok` awaits below. let mut headers: Vec = Vec::new(); - { - let pipeline = consensus.pipeline().borrow(); + consensus.with_pipeline(|pipeline| { let from = consensus.commit_max() + 1; let to = consensus.sequencer().current_sequence(); for op in from..=to { @@ -2582,7 +2568,7 @@ where headers.push(header); } } - } + }); if headers.is_empty() { return; } @@ -2685,21 +2671,18 @@ where // Revalidate after the await: a sibling driver may have // committed this op (and more) while we were parked. - let head_is_ours = consensus.pipeline().borrow().head().is_some_and(|head| { - head.header.op == prepare_header.op - && head.header.checksum == prepare_header.checksum + let head_is_ours = consensus.pipeline_head_header().is_some_and(|head| { + head.op == prepare_header.op && head.checksum == prepare_header.checksum }); if !head_is_ours { continue; } let mut entry = consensus - .pipeline() - .borrow_mut() - .pop() + .pop_committed_prepare() .expect("on_ack: revalidated head exists"); - let pipeline_depth = consensus.pipeline().borrow().len(); + let pipeline_depth = consensus.pipeline_len(); let event = CommitLogEvent { replica: ReplicaLogContext::from_consensus(consensus, PlaneKind::Metadata), op: prepare_header.op, @@ -2861,10 +2844,8 @@ where return; } let stranded_commits = consensus.commit_min() < consensus.commit_max(); - let promotable_requests = { - let pipeline = consensus.pipeline().borrow(); - !pipeline.request_queue_is_empty() && !pipeline.is_full() - }; + let promotable_requests = consensus + .with_pipeline(|pipeline| !pipeline.request_queue_is_empty() && !pipeline.is_full()); if !stranded_commits && !promotable_requests { return; } @@ -2889,10 +2870,10 @@ where // one commit window drains the moment the window closes. Promoted // prepares are un-quorum'd, so they never re-close the gate here. loop { - if consensus.pipeline().borrow().is_full() { + if consensus.pipeline_is_full() { break; } - let req = consensus.pipeline().borrow_mut().pop_request(); + let req = consensus.pop_queued_request(); let Some(mut req) = req else { break }; let client_id = req.message.header().client; @@ -5172,7 +5153,7 @@ mod tests { "mid-window register must park in the request queue, not error" ); assert_eq!( - consensus.pipeline().borrow().request_queue_len(), + consensus.request_queue_len(), 1, "register buffered in the request queue" ); @@ -5190,7 +5171,7 @@ mod tests { assert!(resumed, "B's commit must complete and promote the register"); assert_eq!(consensus.commit_min(), 1, "B's op committed"); assert_eq!( - consensus.pipeline().borrow().request_queue_len(), + consensus.request_queue_len(), 0, "promotion emptied the request queue" ); @@ -5305,7 +5286,7 @@ mod tests { // C's register lands in the window: absorbed into the request queue. let mut register = Box::pin(md.submit_register_in_process(CLIENT_C, ACTING_USER)); assert!(register.as_mut().poll(&mut cx).is_pending()); - assert_eq!(consensus.pipeline().borrow().request_queue_len(), 1); + assert_eq!(consensus.request_queue_len(), 1); // The committing driver dies at its await — the hyper-disconnect // analogue. Commit and promotion are now stranded: op 1 is quorum'd @@ -5314,7 +5295,7 @@ mod tests { drop(driver); assert_eq!(consensus.commit_max(), 1); assert_eq!(consensus.commit_min(), 0); - assert_eq!(consensus.pipeline().borrow().request_queue_len(), 1); + assert_eq!(consensus.request_queue_len(), 1); assert!( register.as_mut().poll(&mut cx).is_pending(), "queued register must still be parked with no driver alive" @@ -5325,11 +5306,7 @@ mod tests { // (its self-ack lands on the loopback). md.resume_stranded_commits().await; assert_eq!(consensus.commit_min(), 1, "stranded op 1 applied"); - assert_eq!( - consensus.pipeline().borrow().request_queue_len(), - 0, - "queued register promoted" - ); + assert_eq!(consensus.request_queue_len(), 0, "queued register promoted"); // Commit the promoted register (production: pump loopback drain) // and the parked caller resolves with its session. @@ -5360,4 +5337,81 @@ mod tests { assert_eq!(md.client_table.borrow().get_epoch(CLIENT_C), Some(2)); assert!(is_caught_up_primary(consensus)); } + + #[compio::test] + async fn failed_journal_append_hands_the_op_back_instead_of_leaving_a_phantom() { + // The primary claims its op before the append (`push_prepare_entry`), so a + // failed append used to leave the sequencer one ahead of the WAL forever: + // the next request projected over the hole, and no repair path refilled it. + const CLIENT: u128 = 1; + const SESSION: u64 = 1; + const ACTING_USER: u32 = 7; + + let dir = tempfile::tempdir().unwrap(); + let journal = + journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0) + .await + .unwrap(); + let consensus = VsrConsensus::new( + 1, + 0, + 1, + server_common::sharding::METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + None, + TestMux::default(), + None, + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table.borrow_mut().commit_register( + CLIENT, + ACTING_USER, + register_reply(CLIENT, SESSION), + ); + + let projected = md + .prepare_request(create_stream_request(CLIENT, 1, "s1")) + .expect("CreateStream is client-allowed"); + let op = projected.header().op; + let parent = projected.header().parent; + let sequence_before = consensus.sequencer().current_sequence(); + + // Forces the append to fail deterministically, before any disk write: the + // buffer carries eight bytes of slack past the header's `size`, which + // `PrepareJournal::append` refuses rather than write slack that would + // mis-frame the recovery scan. Any append failure reaches the same arm. + let size = projected.header().size as usize; + let mut padded = Message::::new(size + 8); + padded.as_mut_slice()[..size].copy_from_slice(projected.as_slice()); + + consensus.pipeline_message(PlaneKind::Metadata, &padded); + assert_eq!(consensus.sequencer().current_sequence(), op); + + md.on_replicate(padded).await; + + assert_eq!( + consensus.sequencer().current_sequence(), + sequence_before, + "the claimed op must be handed back so the next request reuses it" + ); + assert_eq!(consensus.last_prepare_checksum(), parent); + assert!( + consensus.pipeline_is_empty(), + "the undurable prepare must not stay live in the pipeline" + ); + #[allow(clippy::cast_possible_truncation)] + let journaled = md.journal.as_ref().unwrap().handle().header(op as usize); + assert!( + journaled.is_none(), + "the append failed, so the WAL must hold nothing at that op" + ); + } } diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs index 3b2e009afd..12d0cd8449 100644 --- a/core/metadata/src/impls/recovery.rs +++ b/core/metadata/src/impls/recovery.rs @@ -794,6 +794,7 @@ fn verify_checkpoint_pairing( mod tests { use super::*; use crate::impls::metadata::checkpoint_checksum; + use crate::stm::snapshot::SNAPSHOT_FORMAT_VERSION; use consensus::CLIENTS_TABLE_MAX; use iggy_binary_protocol::consensus::{Command2, Operation}; use journal::Journal; @@ -1672,47 +1673,63 @@ mod tests { } #[compio::test] - async fn recover_accepts_snapshot_written_before_a_trailing_default_field() { - // Appending a `#[serde(default)]` field is this repo's forward-compatible - // snapshot migration, and msgpack encodes structs positionally: a file the - // previous build wrote decodes fine (the default fills the missing element) - // but re-encodes with one MORE element. A pairing checksum recomputed by - // re-encoding the decoded snapshot would therefore diverge on the FIRST boot - // of the new build and refuse every checkpointed node, with the WAL prefix - // already drained. Hashing the bytes on disk is what makes that upgrade boot. - // - // Emulated in the direction the migration runs: strip the trailing element off - // a current-shape file, which is what the pre-`client_table` build wrote. + async fn recover_refuses_a_snapshot_from_another_format_version() { + // A snapshot shape is only as trustworthy as the version stamped on it, so a + // foreign one refuses boot rather than restoring whatever msgpack happens to + // make of the bytes. Everything else about the directory is healthy: the + // superblock pairs with the file on disk, so the refusal can only be the + // version. + const CHECKPOINT_OP: u64 = 42; + let mut snapshot = IggySnapshot::new(CHECKPOINT_OP); + snapshot.snapshot_mut().version = SNAPSHOT_FORMAT_VERSION + 1; + let foreign = snapshot.encode().unwrap(); + + let dir = tempdir().unwrap(); + let metadata_dir = dir.path().join("metadata"); + std::fs::create_dir_all(&metadata_dir).unwrap(); + std::fs::write(metadata_dir.join("snapshot.bin"), &foreign).unwrap(); + let state = vsr_state_with_checkpoint(CHECKPOINT_OP, checkpoint_checksum(&foreign)); + { + let superblock = PingPongSuperblock::open(&metadata_dir).await.unwrap(); + superblock.write(&state.to_bytes()).await.unwrap(); + } + + match recover::( + dir.path(), + CLUSTERED, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + CLIENTS_TABLE_MAX, + |_| {}, + ) + .await + { + Err(RecoveryError::Snapshot(SnapshotError::UnsupportedFormatVersion { + found, + expected, + })) => { + assert_eq!(found, SNAPSHOT_FORMAT_VERSION + 1); + assert_eq!(expected, SNAPSHOT_FORMAT_VERSION); + } + Err(other) => panic!("expected a format-version refusal, got {other}"), + Ok(_) => panic!("expected a foreign format version to refuse boot"), + } + } + + #[compio::test] + async fn recover_pairs_the_checkpoint_against_the_bytes_on_disk() { + // The pairing checksum is taken over the file's bytes, never over a re-encode + // of the decoded snapshot. Re-encoding would tie recovery to + // decode-then-encode staying byte-identical across every serde and rmp + // release, and a divergence there would refuse boot on every checkpointed node + // with its WAL prefix already drained. const CHECKPOINT_OP: u64 = 42; let encoded = IggySnapshot::new(CHECKPOINT_OP).encode().unwrap(); - assert_eq!( - encoded[0] & 0xf0, - 0x90, - "snapshot must encode as a msgpack fixarray for this emulation" - ); - assert_eq!( - *encoded.last().unwrap(), - 0xC0, - "the trailing snapshot field must encode as nil here; adjust the emulation \ - if the last field stops being an Option" - ); - let mut legacy = vec![0x90 | ((encoded[0] & 0x0f) - 1)]; - legacy.extend_from_slice(&encoded[1..encoded.len() - 1]); - - let decoded = IggySnapshot::decode(&legacy).unwrap(); - assert_eq!(decoded.sequence_number(), CHECKPOINT_OP); - assert_ne!( - decoded.encode().unwrap(), - legacy, - "the emulated legacy file must NOT round-trip byte-identically, else this \ - test cannot distinguish the two checksum sources" - ); let dir = tempdir().unwrap(); let metadata_dir = dir.path().join("metadata"); std::fs::create_dir_all(&metadata_dir).unwrap(); - std::fs::write(metadata_dir.join("snapshot.bin"), &legacy).unwrap(); - let state = vsr_state_with_checkpoint(CHECKPOINT_OP, checkpoint_checksum(&legacy)); + std::fs::write(metadata_dir.join("snapshot.bin"), &encoded).unwrap(); + let state = vsr_state_with_checkpoint(CHECKPOINT_OP, checkpoint_checksum(&encoded)); { let superblock = PingPongSuperblock::open(&metadata_dir).await.unwrap(); superblock.write(&state.to_bytes()).await.unwrap(); @@ -1730,7 +1747,7 @@ mod tests { assert_eq!(recovered.snapshot_checkpoint.0, CHECKPOINT_OP); assert_eq!( recovered.snapshot_checkpoint.1, - checkpoint_checksum(&legacy), + checkpoint_checksum(&encoded), "the verified pairing must be the checksum of the bytes on disk" ); } diff --git a/core/metadata/src/stm/authz.rs b/core/metadata/src/stm/authz.rs index 00ea174161..bbd09fa52c 100644 --- a/core/metadata/src/stm/authz.rs +++ b/core/metadata/src/stm/authz.rs @@ -56,7 +56,6 @@ use iggy_binary_protocol::requests::users::ChangePasswordRequest; use iggy_binary_protocol::{Operation, PrepareHeader, WireDecode, WireIdentifier}; use iggy_common::{IggyError, variadic}; use server_common::Message; -use std::mem::size_of; /// Gate a committed prepare, then apply it. A denial commits as an /// `Unauthorized` no-op (the gate never mutates state); an allow proceeds to @@ -137,7 +136,7 @@ pub(crate) fn authorize( if user_id == ROOT_USER_ID { return None; } - let body = &prepare.as_slice()[size_of::()..header.size as usize]; + let body = prepare.body(); match header.operation { // Streams. `create_stream` is unscoped; the rest resolve the stream id. diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs index 698dbd3c02..cd363cb24b 100644 --- a/core/metadata/src/stm/mod.rs +++ b/core/metadata/src/stm/mod.rs @@ -412,18 +412,25 @@ macro_rules! collect_handlers { fn parse(input: Self::Input) -> Result<::iggy_common::Either, Self::Error> { use ::iggy_binary_protocol::WireDecode; use ::iggy_common::Either; - use ::iggy_binary_protocol::{Operation, PrepareHeader}; - match input.header().operation { + use ::iggy_binary_protocol::Operation; + + // Both scalars copied out of one header read. `header()` + // re-validates the bit pattern on each call, and the borrow has + // to end before the pass-through arm can move `input` on. + let (operation, timestamp) = { + let header = input.header(); + (header.operation, header.timestamp) + }; + + match operation { $( Operation::$operation => { - // TODO: FIXME, zero allocation operation construction. - let header = *input.header(); - let body = ::bytes::Bytes::copy_from_slice( - &input.as_slice()[core::mem::size_of::()..header.size as usize] - ); - let cmd = [<$operation Request>]::decode_from(&body) + // Decoded straight off the backing buffer. Every field a + // request keeps is owned, so nothing borrows past this + // call and the body never needs a copy of its own. + let cmd = [<$operation Request>]::decode_from(input.body()) .map_err(|_| ::iggy_common::IggyError::InvalidCommand)?; - let ts = ::iggy_common::IggyTimestamp::from(header.timestamp); + let ts = ::iggy_common::IggyTimestamp::from(timestamp); Ok(Either::Left([<$state Command>]::$operation(cmd, ts))) }, )* diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs index e9fadca173..de98e323ef 100644 --- a/core/metadata/src/stm/snapshot.rs +++ b/core/metadata/src/stm/snapshot.rs @@ -15,12 +15,52 @@ // specific language governing permissions and limitations // under the License. +use iggy_binary_protocol::IGGY_PROTOCOL_VERSION; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use std::fmt; use crate::stm::stream::StreamsSnapshot; use crate::stm::user::UsersSnapshot; +/// The version of the snapshot format in use, reserved for breaking changes. +/// +/// One version means exactly one serialized shape, and [`MetadataSnapshot::decode`] +/// accepts nothing else. Bump it in the same change that alters the shape: append, +/// remove, reorder, retype, or redefine the meaning of any field, at any depth +/// under [`MetadataSnapshot`]. There is no accepted range and no per-version +/// translation. A snapshot this build cannot read is refused, not best-effort +/// decoded. +/// +/// Nothing softer would hold. msgpack encodes a struct positionally, so a field one +/// build appends reaches another as an unexplained extra array element; without the +/// version the reader either fails on an unrelated msgpack error or, for a +/// same-length change, silently reads one field's bytes as another's. +/// +/// Bumping it invalidates every `snapshot.bin` already on disk, and a node whose +/// snapshot is refused refuses boot. That is deliberate. The metadata plane is +/// pre-production, so the cost is clearing a data directory; once it ships, a bump +/// needs an explicit translation path added here alongside it. +/// +/// Version 2: `status` sits at reply-header offset 216 (version 1 carried a +/// `namespace` word before it), which the client table's cached replies embed as raw +/// wire bytes msgpack cannot introspect. +pub const SNAPSHOT_FORMAT_VERSION: u32 = 2; + +/// The release that wrote a snapshot: the packed `iggy_binary_protocol` semver of +/// this build. [`iggy_binary_protocol::ProtocolVersion`] documents the packing and +/// renders it as `major.minor.patch`. +/// +/// Provenance, never a gate. The format version says whether the bytes are +/// readable; this says which build produced them, which matters most for a snapshot +/// that arrived over state transfer from a machine whose build this node otherwise +/// has no record of. +/// +/// A build constant, identical on every replica, so two replicas holding identical +/// state still serialize identically (see [`MetadataSnapshot`]). +pub const SNAPSHOT_RELEASE_FORMAT: u32 = IGGY_PROTOCOL_VERSION; + +const _: () = assert!(SNAPSHOT_RELEASE_FORMAT > 0); + #[derive(Debug)] pub enum SnapshotError { /// Serialization failed. @@ -42,6 +82,12 @@ pub enum SnapshotError { ChecksumMismatch { expected: u128, actual: u128 }, /// Snapshot file is too short to contain a valid checksum. Truncated { size: u64 }, + /// The snapshot was written in a format version this build does not read: a + /// different build wrote it, on this disk or on a state-transfer peer. Refuse it + /// rather than let msgpack read one field's bytes as another's. The version is + /// peeked ahead of the rest of the payload, so this fires even when nothing past + /// it is recognizable. + UnsupportedFormatVersion { found: u32, expected: u32 }, /// A state-transfer descriptor's frontiers contradict its artifacts: a /// `commit_op` below the snapshot the same offer ships, or a client-table /// frontier above that commit point. Both are impossible from a @@ -53,10 +99,6 @@ pub enum SnapshotError { commit_op: u64, table_frontier: u64, }, - /// The snapshot was written under a different format version. Refuse it - /// rather than reinterpret embedded raw bytes (the client table's cached - /// replies are wire `ReplyHeader` frames) under the wrong layout. - UnsupportedVersion { found: u32, supported: u32 }, } /// Stage at which snapshot persistence failed. @@ -97,6 +139,13 @@ impl fmt::Display for SnapshotError { "snapshot file truncated: {size} bytes (too short for checksum)" ) } + Self::UnsupportedFormatVersion { found, expected } => { + write!( + f, + "snapshot format version {found} is incompatible with this build, which \ + reads version {expected}" + ) + } Self::IncoherentManifest { snapshot_seq, commit_op, @@ -108,13 +157,6 @@ impl fmt::Display for SnapshotError { commit_op {commit_op}, table frontier {table_frontier}" ) } - Self::UnsupportedVersion { found, supported } => { - write!( - f, - "unsupported metadata snapshot version {found}; this build reads only \ - version {supported}" - ) - } } } } @@ -127,8 +169,8 @@ impl std::error::Error for SnapshotError { Self::Io(e) | Self::Persist { source: e, .. } => Some(e), Self::ChecksumMismatch { .. } | Self::Truncated { .. } - | Self::IncoherentManifest { .. } - | Self::UnsupportedVersion { .. } => None, + | Self::UnsupportedFormatVersion { .. } + | Self::IncoherentManifest { .. } => None, } } } @@ -150,17 +192,14 @@ impl From for SnapshotError { /// replicas with identical state must serialize identically. Regression guards: /// `stream::tests::populated_streams_snapshot_reencode_is_byte_stable` and /// `impls::metadata::tests::populated_snapshot_reencode_and_checksum_are_stable`. -/// Current [`MetadataSnapshot::version`]. Bump whenever the serialized form -/// changes meaning without changing shape -- in particular the client table's -/// cached replies, which are embedded as raw `ReplyHeader` wire bytes msgpack -/// cannot introspect. Version 2: `status` sits at reply-header offset 216 -/// (version 1 carried a `namespace` word before it). -pub const METADATA_SNAPSHOT_VERSION: u32 = 2; - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MetadataSnapshot { - /// Snapshot format version; [`MetadataSnapshot::decode`] refuses any other - /// value (see [`METADATA_SNAPSHOT_VERSION`]). + /// The version of the snapshot format in use, reserved for breaking changes. + /// See [`SNAPSHOT_FORMAT_VERSION`]; [`Self::decode`] refuses anything else. + /// + /// First field deliberately: msgpack encodes the struct positionally, so this is + /// the one element [`peek_format_version`] can pull out before it knows the rest + /// of the layout. pub version: u32, /// Timestamp when the snapshot was created (microseconds since epoch). pub created_at: u64, @@ -173,10 +212,11 @@ pub struct MetadataSnapshot { pub streams: Option, /// Client-table state (sessions + at-most-once dedup replies) at the checkpoint /// op. Not a state machine, but folded in so a restart that drained the WAL - /// prefix still recovers pre-checkpoint sessions. `#[serde(default)]` so a - /// snapshot predating this field decodes as `None`. - #[serde(default)] + /// prefix still recovers pre-checkpoint sessions. pub client_table: Option, + /// The release that wrote this snapshot, as a packed `iggy_binary_protocol` + /// semver. Provenance only, never a gate. See [`SNAPSHOT_RELEASE_FORMAT`]. + pub release_format: u32, } impl Default for MetadataSnapshot { @@ -190,7 +230,7 @@ impl MetadataSnapshot { #[must_use] pub const fn new(sequence_number: u64) -> Self { Self { - version: METADATA_SNAPSHOT_VERSION, + version: SNAPSHOT_FORMAT_VERSION, // Deterministic placeholder. The real creation time is stamped by // `Snapshot::create` from the consensus-injected clock (see // `VsrConsensus::clock_realtime_micros`) so a replayed simulator @@ -201,6 +241,7 @@ impl MetadataSnapshot { users: None, streams: None, client_table: None, + release_format: SNAPSHOT_RELEASE_FORMAT, } } @@ -212,24 +253,51 @@ impl MetadataSnapshot { rmp_serde::to_vec(self).map_err(SnapshotError::Serialize) } - /// Decode a snapshot from msgpack bytes. + /// Decode a snapshot from msgpack bytes, refusing a format version this build + /// does not read. + /// + /// The version is checked before the payload is deserialized, so a snapshot + /// whose layout past that field is unknown is refused by version rather than by + /// whichever later field happened to misparse. /// /// # Errors - /// Returns `SnapshotError::Deserialize` if msgpack deserialization fails, - /// or `SnapshotError::UnsupportedVersion` if the snapshot was written - /// under a different format version. + /// [`SnapshotError::UnsupportedFormatVersion`] when the stamped version is not + /// [`SNAPSHOT_FORMAT_VERSION`], or [`SnapshotError::Deserialize`] if msgpack + /// deserialization fails. pub fn decode(bytes: &[u8]) -> Result { - let snapshot: Self = rmp_serde::from_slice(bytes).map_err(SnapshotError::Deserialize)?; - if snapshot.version != METADATA_SNAPSHOT_VERSION { - return Err(SnapshotError::UnsupportedVersion { - found: snapshot.version, - supported: METADATA_SNAPSHOT_VERSION, + // Bytes carrying no readable version are not a snapshot at all, so they fall + // through to the deserializer, whose error names what actually went wrong. + if let Some(version) = peek_format_version(bytes) + && version != SNAPSHOT_FORMAT_VERSION + { + return Err(SnapshotError::UnsupportedFormatVersion { + found: version, + expected: SNAPSHOT_FORMAT_VERSION, }); } - Ok(snapshot) + rmp_serde::from_slice(bytes).map_err(SnapshotError::Deserialize) } } +/// The format version stamped in encoded snapshot bytes, read without decoding the +/// rest of them. +/// +/// `None` for anything that is not a msgpack array whose first element is an +/// unsigned integer, which covers every input that is not a snapshot. +/// +/// Reading the version on its own is what keeps the check working across a layout +/// change: deserializing the whole struct first would let some later field fail and +/// report a msgpack error naming no version, which is the difference between "this +/// file came from a newer build" and "this file is corrupt". `journal::superblock` +/// reads its own version field the same way, ahead of the length and checksum it +/// cannot yet trust. +#[must_use] +pub fn peek_format_version(encoded: &[u8]) -> Option { + let mut cursor = encoded; + rmp::decode::read_array_len(&mut cursor).ok()?; + rmp::decode::read_int(&mut cursor).ok() +} + /// Trait for metadata snapshot implementations. /// /// This is the high-level interface that concrete snapshot types (e.g. `IggySnapshot`) @@ -463,28 +531,110 @@ mod tests { let decoded = MetadataSnapshot::decode(&encoded).unwrap(); assert_eq!(decoded.sequence_number, 42); + assert_eq!(decoded.version, SNAPSHOT_FORMAT_VERSION); + assert_eq!(decoded.release_format, SNAPSHOT_RELEASE_FORMAT); assert!(decoded.users.is_none()); assert!(decoded.streams.is_none()); assert!(decoded.client_table.is_none()); } - // The client table's cached replies are embedded as raw `ReplyHeader` - // wire bytes msgpack cannot introspect, so a snapshot from a different - // format version must be refused, never reinterpreted under the current - // header layout. #[test] - fn decode_refuses_a_snapshot_from_another_format_version() { - let mut snapshot = MetadataSnapshot::new(42); - snapshot.version = METADATA_SNAPSHOT_VERSION - 1; + fn a_build_decodes_what_it_writes() { + // Exact-equality versioning makes this the one thing that could go wrong + // silently: a stamp the writer picks and the reader rejects would refuse every + // node its own snapshot on the next boot. + let encoded = MetadataSnapshot::new(5).encode().unwrap(); + MetadataSnapshot::decode(&encoded).expect("a build must read its own snapshot"); + } - let encoded = snapshot.encode().unwrap(); - assert!(matches!( - MetadataSnapshot::decode(&encoded), - Err(SnapshotError::UnsupportedVersion { - found, - supported: METADATA_SNAPSHOT_VERSION, - }) if found == METADATA_SNAPSHOT_VERSION - 1 - )); + #[test] + fn release_stamp_is_a_build_constant() { + // Two replicas holding identical state must serialize identically, which a + // per-node or per-write release stamp would break. + assert_eq!( + MetadataSnapshot::new(1).encode().unwrap(), + MetadataSnapshot::new(1).encode().unwrap() + ); + } + + #[test] + fn encoded_snapshot_leads_with_the_version_element() { + // `peek_format_version` rests on this shape: a msgpack array (`rmp_serde`'s + // compact struct encoding) whose first element is `version`. Switching to + // `to_vec_named` would encode a map instead and silently blind the peek, so + // pin it here. + let encoded = MetadataSnapshot::new(3).encode().unwrap(); + assert_eq!( + encoded[0] & 0xf0, + 0x90, + "snapshot must encode as a msgpack fixarray" + ); + assert_eq!( + peek_format_version(&encoded), + Some(SNAPSHOT_FORMAT_VERSION), + "the first array element must be the format version" + ); + } + + #[test] + fn peek_format_version_ignores_bytes_that_are_not_a_snapshot() { + assert_eq!(peek_format_version(&[]), None); + // A msgpack string, not an array. + assert_eq!(peek_format_version(&[0xa1, b'x']), None); + // An array whose first element is not an unsigned integer. + assert_eq!(peek_format_version(&[0x91, 0xc0]), None); + } + + #[test] + fn decode_refuses_any_other_format_version() { + // Newer and older alike: there is no accepted window, so both directions are + // the same refusal. Version 0 is what a zeroed or absent stamp reads as. + for forged in [SNAPSHOT_FORMAT_VERSION + 1, 0] { + let mut snapshot = MetadataSnapshot::new(9); + snapshot.version = forged; + let encoded = snapshot.encode().unwrap(); + + match MetadataSnapshot::decode(&encoded) { + Err(SnapshotError::UnsupportedFormatVersion { found, expected }) => { + assert_eq!(found, forged); + assert_eq!(expected, SNAPSHOT_FORMAT_VERSION); + } + other => panic!("expected an unsupported-version refusal, got {other:?}"), + } + } + } + + #[test] + fn decode_refuses_a_shape_change_by_version_not_by_msgpack() { + // The reason the version is peeked instead of read off the decoded struct: a + // future build's layout need not decode at all under this one, and the refusal + // must still name the version rather than whichever field happened to + // misparse. Emulated by appending an element, which is what a field append + // looks like on the wire. + let mut snapshot = MetadataSnapshot::new(4); + snapshot.version = SNAPSHOT_FORMAT_VERSION + 1; + let current = snapshot.encode().unwrap(); + assert_eq!(current[0] & 0xf0, 0x90, "expected a msgpack fixarray"); + let mut future = vec![0x90 | ((current[0] & 0x0f) + 1)]; + future.extend_from_slice(¤t[1..]); + future.push(0xc0); + + match MetadataSnapshot::decode(&future) { + Err(SnapshotError::UnsupportedFormatVersion { found, .. }) => { + assert_eq!(found, SNAPSHOT_FORMAT_VERSION + 1); + } + other => panic!("expected an unsupported-version refusal, got {other:?}"), + } + } + + #[test] + fn decode_reports_deserialization_when_no_version_is_readable() { + // Bytes carrying no version field are corruption, not a version skew, and the + // msgpack error names that far better than a fabricated version would. + match MetadataSnapshot::decode(&[0xa3, b'n', b'o', b'!']) { + Err(SnapshotError::Deserialize(_)) => {} + other => panic!("expected a deserialization error, got {other:?}"), + } } #[test] diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs index 7dffbeb8ef..00011b8797 100644 --- a/core/metadata/src/stm/user.rs +++ b/core/metadata/src/stm/user.rs @@ -617,11 +617,21 @@ impl StateHandler for UpdatePermissionsRequest { } } -// TODO(hubcio): Serialize proper reply (e.g. generated raw token from the -// primary-side mint) instead of empty Bytes. The raw token is currently -// generated only at the request-rewrite step on the primary and dropped; -// surfacing it back to the client needs a side-channel out of -// `maybe_rewrite_pat_request`. +/// The success reply here is deliberately empty: the raw token the caller needs +/// is the one thing this apply must never see. +/// +/// The primary mints the raw token and its hash at ingress (server-ng +/// `pat::rewrite_pat_request_for_user`) and replicates only the hash. Minting +/// inside this apply would call `ring::rand` on every replica and diverge the +/// token index, and replicating the raw token would persist a live credential in +/// every WAL and snapshot. So the raw token leaves the primary by a side channel +/// (`maybe_rewrite_pat_request` returns it alongside the rewritten request) and +/// the home shard splices it into this op's reply as a typed +/// `RawPersonalAccessTokenResponse` (server-ng `responses::build_raw_pat_reply`). +/// +/// One consequence rides on that: the secret exists only on the wire of the +/// original reply, so a replayed request cannot be served from the client-table +/// cache. `impls::metadata::unreplayable_secret_refusal` refuses it instead. impl StateHandler for CreatePersonalAccessTokenRequest { type State = UsersInner; fn apply(&self, state: &mut UsersInner, timestamp: IggyTimestamp) -> ApplyReply { diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index e5ed479a19..8dc9761699 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -36,7 +36,7 @@ use crate::{ PollingConsumer, }; use consensus::{ - CommitLogEvent, Consensus, PartitionDiagEvent, Pipeline, PipelineEntry, PlaneKind, Project, + CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry, PlaneKind, Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request, build_reply_message, drain_committable_prefix, emit_namespace_progress_event, @@ -1996,11 +1996,9 @@ where // Two-queue: prepare slot -> project+replicate; prepare full + // request room -> buffer; both full -> drop+warn (client retries // via read-timeout). - if consensus.pipeline().borrow().is_full() { - let push_result = consensus - .pipeline() - .borrow_mut() - .push_request(consensus::RequestEntry::new(message)); + if consensus.pipeline_is_full() { + let push_result = + consensus.push_queued_request(consensus::RequestEntry::new(message)); if push_result.is_err() { emit_partition_diag( tracing::Level::WARN, @@ -2054,7 +2052,7 @@ where #[allow(clippy::future_not_send)] pub async fn drain_request_queue_into_prepares(&mut self, slots_freed: usize) { for _ in 0..slots_freed { - let req = self.consensus().pipeline().borrow_mut().pop_request(); + let req = self.consensus().pop_queued_request(); let Some(req) = req else { break }; let prepare = { @@ -2353,7 +2351,7 @@ where SimEventKind::NamespaceProgressUpdated, &ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), header.op, - consensus.pipeline().borrow().len(), + consensus.pipeline_len(), ); } @@ -2379,11 +2377,7 @@ where return; } - let pipeline = consensus.pipeline().borrow(); - if pipeline - .entry_by_op_and_checksum(header.op, header.prepare_checksum) - .is_none() - { + if !consensus.pipeline_holds_entry(header.op, header.prepare_checksum) { emit_partition_diag( tracing::Level::DEBUG, &PartitionDiagEvent::new( @@ -2413,7 +2407,7 @@ where SimEventKind::NamespaceProgressUpdated, &ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), consensus.commit_min(), - consensus.pipeline().borrow().len(), + consensus.pipeline_len(), ); } } @@ -2446,7 +2440,7 @@ where SimEventKind::NamespaceProgressUpdated, &ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), consensus.commit_min(), - consensus.pipeline().borrow().len(), + consensus.pipeline_len(), ); } } @@ -3134,7 +3128,7 @@ where self.consensus.advance_commit_min(prepare_header.op); - let pipeline_depth = self.consensus.pipeline().borrow().len(); + let pipeline_depth = self.consensus.pipeline_len(); let event = CommitLogEvent { replica: ReplicaLogContext::from_consensus(&self.consensus, PlaneKind::Partitions), op: prepare_header.op, @@ -5415,7 +5409,7 @@ mod tests { ); } assert_eq!( - partition.consensus().pipeline().borrow().len(), + partition.consensus().pipeline_len(), 0, "denied delete must not replicate" ); @@ -5430,7 +5424,7 @@ mod tests { .on_request(delete_offset_request(client_id, 8, consumer_id)) .await; assert_eq!( - partition.consensus().pipeline().borrow().len(), + partition.consensus().pipeline_len(), 1, "existing offset delete must replicate" ); @@ -5613,10 +5607,14 @@ mod tests { /// leaving live groups untouched. The returned `Vec` is what the /// reconciler unlinks off-borrow, so it carries no partition reference. /// - /// TODO: a true cross-task interleave (pump reallocs the partitions vec - /// while the reconciler awaits the unlink) needs a two-future sim oracle - /// that does not exist yet; this covers the synchronous removal contract - /// the off-borrow split relies on. + /// Scope: the synchronous removal contract the off-borrow split relies on. + /// The cross-task interleave it enables -- a pump mutating the partitions vec + /// while a sibling task is parked mid-await -- is covered on the simulator's + /// deterministic executor, against the debug borrow tripwire, by + /// `simulator::tests::shell_detects_partition_borrow_held_across_await` + /// (`swap_remove`) and + /// `shell_detects_partition_borrow_held_across_a_pump_realloc` (a growing + /// `push`, which relocates every element). #[compio::test] async fn reclaim_dead_group_offsets_drops_dead_keeps_live() { let mut partition = test_partition(); diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 45c0df4b53..85f6f98e78 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -301,6 +301,22 @@ where suspend.await; } + /// TEST / SIMULATOR ONLY. Address of the partitions vec's heap buffer. + /// + /// Lets a test prove that an [`Self::insert`] actually REALLOCATED rather + /// than landing in spare capacity. The distinction is the whole point of the + /// realloc half of PR #3557: `swap_remove` invalidates one slot's reference, + /// while a growing `push` moves every element and invalidates all of them. A + /// test that only checked "insert happened" would pass on a push into spare + /// capacity, which moves nothing and proves nothing. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn buffer_addr(&self) -> usize { + // Safety: read-only reborrow of the same pump-only vec every accessor + // here goes through; nothing is handed out past this statement. + unsafe { (*self.partitions.get()).as_ptr() as usize } + } + /// Get mutable partition by namespace directly. Tombstone-gated like /// [`Self::get_by_ns`]. #[allow(clippy::mut_from_ref)] diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index fd9e17e6f4..75e36fc56c 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -2637,7 +2637,7 @@ where // promoted mid-transfer. (`last_prepare_checksum` needs nothing: it is // only read as a `parent:` stamp when building a prepare.) consensus.sequencer().set_sequence(commit_op); - consensus.pipeline().borrow_mut().clear(); + consensus.clear_pipeline(); consensus.advance_commit_max(commit_op); self.observed_view = self.consensus().view(); self.repair = None; diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 15e290532f..e940a21db9 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -2323,20 +2323,21 @@ fn restore_metadata_consensus( commit_watermark, restored_op, "re-pipelining recovered uncommitted metadata suffix" ); - let mut pipeline = consensus.pipeline().borrow_mut(); - #[allow(clippy::cast_possible_truncation)] - for op in (commit_watermark + 1)..=restored_op { - let Some(header) = journal.header(op as usize) else { - warn!( - op, - "recovered journal suffix has a gap; stopping re-pipeline" - ); - break; - }; - let mut entry = PipelineEntry::new(*header); - entry.add_ack(topology.self_replica_id); - pipeline.push(entry); - } + consensus.with_pipeline_mut(|pipeline| { + #[allow(clippy::cast_possible_truncation)] + for op in (commit_watermark + 1)..=restored_op { + let Some(header) = journal.header(op as usize) else { + warn!( + op, + "recovered journal suffix has a gap; stopping re-pipeline" + ); + break; + }; + let mut entry = PipelineEntry::new(*header); + entry.add_ack(topology.self_replica_id); + pipeline.push(entry); + } + }); } consensus diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index aa198b8d4a..040cfc20a6 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -3158,7 +3158,7 @@ mod tests { use partitions::{IggyPartitions, PartitionsConfig}; use server_common::iobuf::Frozen; use server_common::sharding::ShardId; - use server_common::{MESSAGE_ALIGN, Message}; + use server_common::{MESSAGE_ALIGN, Message, MessageBag}; use shard::metrics::ShardMetrics; use shard::shards_table::PapayaShardsTable; use shard::{ @@ -3660,7 +3660,7 @@ mod tests { *new_header = header; new_header.group = namespace.inner(); }); - shard.on_message(request.into_generic()).await; + shard.on_message(MessageBag::Request(request)).await; let replies = bus.client_replies.borrow(); assert_eq!( @@ -3737,7 +3737,7 @@ mod tests { new_header.group = namespace.inner(); }); // Namespace neither materialised nor tombstoned: the frame parks. - shard.on_message(request.into_generic()).await; + shard.on_message(MessageBag::Request(request)).await; shard.enqueue_reconcile_op(ReconcileOp::ConfirmRemove { namespace }); shard.apply_reconcile_ops(); diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index b08cf52290..1305bcf7b8 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -1175,8 +1175,7 @@ mod tests { PurgeTopicRequest, }; use iggy_binary_protocol::{ - Command2, GenericHeader, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader, - WireIdentifier, + Command2, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader, WireIdentifier, }; use message_bus::IggyMessageBus; use metadata::IggyMetadata; @@ -1186,8 +1185,8 @@ mod tests { use metadata::stm::stream::Streams; use metadata::stm::user::Users; use partitions::{IggyPartitions, PartitionsConfig}; - use server_common::Message; use server_common::sharding::{IggyNamespace, ShardId}; + use server_common::{Message, MessageBag}; use shard::shards_table::{PapayaShardsTable, ShardsTable, calculate_shard_assignment}; use shard::{IggyShard, PartitionConsensusConfig, ReconcileOp, ShardIdentity}; use std::mem::size_of; @@ -1252,7 +1251,7 @@ mod tests { /// 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 { + fn build_partition_prepare(namespace: IggyNamespace, op: u64) -> MessageBag { let header_size = size_of::(); let mut msg = Message::::new(header_size); let header = bytemuck::checked::try_from_bytes_mut::( @@ -1264,7 +1263,7 @@ mod tests { header.operation = Operation::SendMessages; header.group = namespace.inner(); header.op = op; - msg.into_generic() + MessageBag::Prepare(msg) } async fn park_one_prepare(shard: &TestShard, namespace: IggyNamespace, op: u64) { @@ -1276,7 +1275,7 @@ mod tests { /// 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 { + fn build_partition_request(namespace: IggyNamespace) -> MessageBag { build_partition_request_sized(namespace, 0) } @@ -1290,10 +1289,7 @@ mod tests { /// [`build_partition_request`] with `body_len` trailing payload bytes, so a /// test can drive the park buffer's byte budget rather than its frame cap. - fn build_partition_request_sized( - namespace: IggyNamespace, - body_len: usize, - ) -> Message { + fn build_partition_request_sized(namespace: IggyNamespace, body_len: usize) -> MessageBag { let header_size = size_of::(); let total_size = header_size + body_len; let mut msg = Message::::new(total_size); @@ -1310,7 +1306,7 @@ mod tests { header.session = 1; header.request = TEST_REQUEST_ID; header.client = TEST_CLIENT_ID; - msg.into_generic() + MessageBag::Request(msg) } fn assignment(partition_id: u32, consensus_group_id: u64) -> CreatedPartitionAssignment { diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs index 7adb4a46da..e2c0af29b0 100644 --- a/core/server/src/responses.rs +++ b/core/server/src/responses.rs @@ -27,7 +27,7 @@ use crate::bootstrap::{ShellBus, ShellShard}; use crate::cluster_meta::ClusterRoster; use crate::session_manager::SessionManager; use crate::wire::{transport_kind_to_wire, usize_to_u32}; -use bytes::{BufMut, Bytes, BytesMut}; +use bytes::{Bytes, BytesMut}; use consensus::{MetadataHandle, VsrConsensus}; use iggy_binary_protocol::PrepareHeader; use iggy_binary_protocol::codes::{ @@ -1303,6 +1303,38 @@ pub(crate) fn build_deny_reply( /// Server build version advertised in the login-register response. const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION"); +/// Build a metadata reply carrying `payload` behind a success result section. +/// +/// Every metadata reply is result-framed, and the SDK strips that section off +/// unconditionally (`split_metadata_result`) before decoding what follows. A +/// payload written without the leading zero count therefore has its first four +/// bytes eaten as a result count, and the decode fails or, worse, succeeds on the +/// shifted remainder. That is not hypothetical: the raw-PAT reply shipped once +/// without the prefix and broke SDK decoding. This is the only way to emit a +/// success reply with a body, so the prefix cannot be forgotten again. +fn build_result_framed_reply( + request_header: &RoutedRequestHeader, + client_id: u128, + session: u64, + commit: u64, + payload: &impl WireEncode, +) -> Message { + let mut encoded = BytesMut::with_capacity(payload.encoded_size()); + payload.encode(&mut encoded); + build_reply_with_body( + request_header, + client_id, + session, + commit, + RESULT_COUNT_LEN + encoded.len(), + |out| { + let (count, body) = out.split_at_mut(RESULT_COUNT_LEN); + count.copy_from_slice(&0u32.to_le_bytes()); + body.copy_from_slice(&encoded); + }, + ) +} + pub(crate) fn build_login_register_reply( request_header: &RoutedRequestHeader, client_id: u128, @@ -1310,28 +1342,16 @@ pub(crate) fn build_login_register_reply( commit: u64, user_id: u32, ) -> Message { - // Result-framed like every metadata reply: a zero result-count (success) - // followed by the `LoginRegisterResponse` payload. A transient Register - // instead ships a `[count=1][index=0][TransientNotCommitted]` frame - // (`build_transient_reply`), which the SDK decodes and replays. The matching - // strip is in the SDK `split_metadata_result` (Register is result-framed). + // A transient Register instead ships a `[count=1][index=0] + // [TransientNotCommitted]` frame (`build_transient_reply`), which the SDK + // decodes and replays. let payload = LoginRegisterResponse { user_id, session, server_protocol_version: IGGY_PROTOCOL_VERSION, server_version: WireName::new(SERVER_VERSION).expect("SERVER_VERSION is 1-255 bytes"), - } - .to_bytes(); - let mut body = Vec::with_capacity(RESULT_COUNT_LEN + payload.len()); - body.extend_from_slice(&[0u8; RESULT_COUNT_LEN]); - body.extend_from_slice(&payload); - build_reply_from_bytes( - request_header, - client_id, - session, - commit, - &Bytes::from(body), - ) + }; + build_result_framed_reply(request_header, client_id, session, commit, &payload) } pub(crate) fn build_reply_from_bytes( @@ -1401,19 +1421,12 @@ pub(crate) fn build_raw_pat_reply( } let token = WireName::new(raw.as_str()).map_err(|_| IggyError::InvalidFormat)?; let response = RawPersonalAccessTokenResponse { token }; - // The SDK strips a leading result section from every metadata reply - // (`split_metadata_result`), so the spliced body must carry the success - // section like any committed metadata reply; a bare token body would be - // read as a garbage result code. - let mut body = BytesMut::with_capacity(RESULT_COUNT_LEN + response.encoded_size()); - body.put_u32_le(0); - response.encode(&mut body); - let reply = build_reply_from_bytes( + let reply = build_result_framed_reply( request_header, request_header.client, request_header.session, commit, - &body.freeze(), + &response, ); Ok(reply.into_generic()) } @@ -1651,6 +1664,27 @@ mod tests { header } + #[test] + fn login_register_reply_carries_the_success_result_prefix() { + // The other `build_result_framed_reply` caller. The SDK strips the result + // section off every metadata reply, so a payload emitted without the prefix + // loses its first four bytes to a phantom result count -- the decode break + // the raw-PAT reply shipped once. Pin it on both callers, not just the one + // that regressed. + let mut header = pat_request_header(); + header.operation = Operation::Register; + let reply = build_login_register_reply(&header, 42, 7, 9, 5); + + let header_len = std::mem::size_of::(); + let body = &reply.as_slice()[header_len..reply.header().size as usize]; + assert_eq!(result_code(body), Some(0)); + + let payload = LoginRegisterResponse::decode_from(&body[RESULT_COUNT_LEN..]) + .expect("login-register payload decodes past the result section"); + assert_eq!(payload.user_id, 5); + assert_eq!(payload.session, 7); + } + /// A committed metadata reply whose body is the given result section /// (`[count][{index, result}]*`), as the commit path emits it. fn committed_reply(result_body: &[u8]) -> Message { diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs index 66f973a999..f591345175 100644 --- a/core/server_common/src/consensus_message.rs +++ b/core/server_common/src/consensus_message.rs @@ -335,6 +335,24 @@ where >::as_slice(&self.backing) } + /// The frame body: the bytes after the header, up to the header's `size`. + /// Empty for a header-only frame. + /// + /// Borrowed from the backing buffer rather than copied out of it. A + /// `WireDecode` parse takes `&[u8]` and keeps only the fields it decodes, so + /// copying the body first buys nothing and costs a memcpy per frame on paths + /// that run per replicated op (`metadata::stm`'s apply, `metadata::stm::authz`). + /// + /// # Panics + /// If `size` does not span the header or overruns the buffer. + /// [`TryFrom`](Message::try_from) rejects both, so every received frame + /// satisfies this; a buffer from [`Message::new`] must have its header stamped + /// first. + #[must_use] + pub fn body(&self) -> &[u8] { + &self.as_slice()[size_of::()..self.header().size() as usize] + } + pub fn as_mut_slice(&mut self) -> &mut [u8] { >::as_mut_slice(&mut self.backing) } @@ -555,6 +573,76 @@ pub enum MessageBag { } impl MessageBag { + /// `(operation, group)`: everything the shard router needs to pick a + /// target, read off the already-typed header without consuming the bag. + /// + /// `group` is a plain field on every consensus header rather than a + /// [`ConsensusHeader`] method, which is why this is a match and not a trait + /// call. `RepairPrepare` reads through its wrapped prepare. + #[must_use] + pub fn routing(&self) -> (Operation, u64) { + match self { + Self::Request(message) => (message.header().operation, message.header().group), + Self::Prepare(message) => (message.header().operation, message.header().group), + Self::PrepareOk(message) => (message.header().operation, message.header().group), + Self::StartViewChange(message) => { + (message.header().operation(), message.header().group) + } + Self::DoViewChange(message) => (message.header().operation(), message.header().group), + Self::StartView(message) => (message.header().operation(), message.header().group), + Self::Commit(message) => (message.header().operation(), message.header().group), + Self::RequestStartView(message) => { + (message.header().operation(), message.header().group) + } + Self::RequestPrepares(message) => { + (message.header().operation(), message.header().group) + } + Self::RepairPrepare(message) => { + (message.header().0.operation, message.header().0.group) + } + Self::RepairRangeReply(message) => { + (message.header().operation(), message.header().group) + } + Self::RequestStateTransfer(message) => { + (message.header().operation(), message.header().group) + } + Self::StateTransferTarget(message) => { + (message.header().operation(), message.header().group) + } + Self::RequestStateChunk(message) => { + (message.header().operation(), message.header().group) + } + Self::StateChunk(message) => (message.header().operation(), message.header().group), + } + } + + /// Discard the classification and hand back the underlying frame. + /// + /// Type-erasure only: the backing bytes are untouched, so a later + /// [`MessageBag::try_from`] reclassifies to the same variant. Callers that + /// need the frame in a generic container (the parked-frame buffer) use this; + /// the dispatch path keeps the bag so it never re-parses. + #[must_use] + pub fn into_generic(self) -> Message { + match self { + Self::Request(message) => message.into_generic(), + Self::Prepare(message) => message.into_generic(), + Self::PrepareOk(message) => message.into_generic(), + Self::StartViewChange(message) => message.into_generic(), + Self::DoViewChange(message) => message.into_generic(), + Self::StartView(message) => message.into_generic(), + Self::Commit(message) => message.into_generic(), + Self::RequestStartView(message) => message.into_generic(), + Self::RequestPrepares(message) => message.into_generic(), + Self::RepairPrepare(message) => message.into_generic(), + Self::RepairRangeReply(message) => message.into_generic(), + Self::RequestStateTransfer(message) => message.into_generic(), + Self::StateTransferTarget(message) => message.into_generic(), + Self::RequestStateChunk(message) => message.into_generic(), + Self::StateChunk(message) => message.into_generic(), + } + } + #[must_use] pub fn command(&self) -> Command2 { match self { @@ -896,6 +984,30 @@ mod tests { assert_eq!(generic.total_len(), 256); } + // body(): the slice every wire decode reads from + + #[test] + fn body_is_the_bytes_between_the_header_and_the_frame_size() { + // A 512-byte allocation holding a 260-byte frame: the accessor follows the + // header's `size`, never the buffer that happens to hold it. + const BODY: [u8; 4] = [1, 2, 3, 4]; + let frame_size = size_of::() + BODY.len(); + let mut owned = header_bytes_sized(Command2::Prepare, frame_size as u32, 512); + owned.as_mut_slice()[size_of::()..frame_size].copy_from_slice(&BODY); + + let message = Message::::try_from(owned).expect("valid generic"); + assert_eq!(message.body(), BODY); + } + + #[test] + fn body_is_empty_for_a_header_only_frame() { + // Header-only commands go through the same accessor, so `size` equal to the + // header must yield an empty slice rather than an inverted-range panic. + let owned = header_bytes_sized(Command2::Prepare, size_of::() as u32, 512); + let message = Message::::try_from(owned).expect("valid generic"); + assert!(message.body().is_empty()); + } + // try_as_typed: validation gates the unsafe ptr-cast reborrow #[test] diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index d898e6a909..93658412d4 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -690,9 +690,14 @@ pub enum ShardFrame { /// receiver still validates `target_shard == self.id` and drops /// frames stamped for the wrong shard (`MISROUTED`) to preserve the /// single-pump invariant under any caller bug. + /// + /// Carries the bag the router already classified, not the raw frame: the + /// receiving pump dispatches straight off the variant instead of re-running + /// `bytemuck::checked::try_from_bytes` plus the header's `validate()` on + /// bytes this process validated one hop ago. Consensus { target_shard: u16, - message: Message, + message: MessageBag, }, /// A connection setup or cross-shard forward frame. Drop recovery /// depends on the frame class: [`LifecycleFrame::ForwardReplicaSend`] @@ -703,12 +708,19 @@ pub enum ShardFrame { Lifecycle(LifecycleFrame), } +// Carrying the classified bag widens the Consensus variant (32 B against the +// 24 B of the `Message` it was classified from), which is free +// only while `LifecycleFrame` remains what sizes the union. Every shard inbox +// holds thousands of these, so a regression would surface as queue memory +// rather than as a failing test. +const _: () = assert!(std::mem::size_of::() == std::mem::size_of::()); + impl ShardFrame { /// Create a consensus frame addressed to `target_shard`. The sender /// is the routing authority; `accept_frame_for_self` compares this /// stamp against the receiving shard id in O(1). #[must_use] - pub const fn consensus(target_shard: u16, message: Message) -> Self { + pub const fn consensus(target_shard: u16, message: MessageBag) -> Self { Self::Consensus { target_shard, message, @@ -2257,17 +2269,11 @@ where /// /// Routes requests, replication messages, and acks to either the metadata /// plane or the partitions plane based on `PlaneIdentity::is_applicable`. - // - // TODO(hubcio): perf - this `MessageBag::try_from` is the second parse of - // the same frame; the first ran in `IggyShard::dispatch` (router.rs ~85) - // to extract (operation, namespace) for routing. The work here re-runs - // `bytemuck::checked::try_from_bytes` + per-header `validate()` on bytes - // already validated upstream. See the matching TODO in router.rs for the - // fix: thread the classified `MessageBag` through `ShardFrame::Consensus` - // so this function takes the bag directly and the match below dispatches - // without re-parsing. + /// + /// Takes the bag `IggyShard::dispatch` classified, so the frame is parsed + /// once per hop rather than once for routing and again for dispatch. #[allow(clippy::future_not_send)] - pub async fn on_message(&self, message: Message) + pub async fn on_message(&self, message: MessageBag) where B: MessageBus + 'static, MJ: JournalHandle, @@ -2286,8 +2292,8 @@ where >, T: ShardsTable, { - match MessageBag::try_from(message) { - Ok(MessageBag::Request(request)) => { + match message { + MessageBag::Request(request) => { let routing = (request.header().operation, request.header().group); match self.park_if_unmaterialised(request, routing.0, routing.1) { // The incarnation fence runs only here, on client traffic. @@ -2312,7 +2318,7 @@ where ParkOutcome::Parked => {} } } - Ok(MessageBag::Prepare(prepare)) => { + MessageBag::Prepare(prepare) => { let routing = (prepare.header().operation, prepare.header().group); // A tombstoned prepare still flows to the plane: replicated // traffic has no client awaiting a reply on this node, and @@ -2352,26 +2358,23 @@ where ParkOutcome::Overflow(_) | ParkOutcome::Parked => {} } } - Ok(MessageBag::PrepareOk(prepare_ok)) => self.on_ack(prepare_ok).await, - Ok(MessageBag::StartViewChange(msg)) => self.on_start_view_change(msg).await, - Ok(MessageBag::DoViewChange(msg)) => self.on_do_view_change(msg).await, - Ok(MessageBag::StartView(msg)) => self.on_start_view(msg).await, - Ok(MessageBag::Commit(ref msg)) => self.on_commit(msg).await, - Ok(MessageBag::RequestStartView(ref msg)) => self.on_request_start_view(msg).await, - Ok(MessageBag::RequestPrepares(ref msg)) => self.on_request_prepares(msg).await, - Ok(MessageBag::RepairPrepare(msg)) => self.on_repair_prepare(msg).await, - Ok(MessageBag::RepairRangeReply(ref msg)) => self.on_repair_range_reply(msg).await, - Ok(MessageBag::RequestStateTransfer(ref msg)) => { + MessageBag::PrepareOk(prepare_ok) => self.on_ack(prepare_ok).await, + MessageBag::StartViewChange(msg) => self.on_start_view_change(msg).await, + MessageBag::DoViewChange(msg) => self.on_do_view_change(msg).await, + MessageBag::StartView(msg) => self.on_start_view(msg).await, + MessageBag::Commit(ref msg) => self.on_commit(msg).await, + MessageBag::RequestStartView(ref msg) => self.on_request_start_view(msg).await, + MessageBag::RequestPrepares(ref msg) => self.on_request_prepares(msg).await, + MessageBag::RepairPrepare(msg) => self.on_repair_prepare(msg).await, + MessageBag::RepairRangeReply(ref msg) => self.on_repair_range_reply(msg).await, + MessageBag::RequestStateTransfer(ref msg) => { self.on_request_state_transfer(msg).await; } - Ok(MessageBag::StateTransferTarget(ref msg)) => { + MessageBag::StateTransferTarget(ref msg) => { self.on_state_transfer_target(msg).await; } - Ok(MessageBag::RequestStateChunk(ref msg)) => self.on_request_state_chunk(msg).await, - Ok(MessageBag::StateChunk(ref msg)) => self.on_state_chunk(msg).await, - Err(e) => { - tracing::warn!(shard = self.id, error = %e, "dropping unparsable consensus frame"); - } + MessageBag::RequestStateChunk(ref msg) => self.on_request_state_chunk(msg).await, + MessageBag::StateChunk(ref msg) => self.on_state_chunk(msg).await, } } @@ -2595,7 +2598,27 @@ where while let Some(frame) = remaining.next() { let passes = frame.passes; let parked_epoch = frame.epoch; - let Err(error) = sender.try_send(ShardFrame::consensus(self.id, frame.message)) else { + // Parked frames are stored generic (the buffer holds every variant + // in one Vec), so re-entering the pump costs one classify. That is + // the rare path -- a post-`CreateTopic` convergence window, not the + // per-message steady state the bag handoff exists for. + let bag = match MessageBag::try_from(frame.message) { + Ok(bag) => bag, + Err(error) => { + // The frame classified once already, on the way in, so this + // is unreachable short of memory corruption. Dropping it + // costs a client retry; panicking on the reconciler's path + // would take the shard down. + tracing::error!( + shard = self.id, + namespace_raw = namespace.inner(), + %error, + "parked partition frame no longer classifies; dropping it" + ); + continue; + } + }; + let Err(error) = sender.try_send(ShardFrame::consensus(self.id, bag)) else { continue; }; let (refused, disconnected) = match error { @@ -2608,7 +2631,7 @@ where let refused_frame = ParkedFrame { epoch: parked_epoch, passes, - message, + message: message.into_generic(), }; if disconnected { // Pump gone: re-parking holds the frame until process exit, and @@ -8126,10 +8149,11 @@ fn rebuild_pipeline_entries( ); } - let mut pipeline = consensus.pipeline().borrow_mut(); - for entry in entries { - pipeline.push(entry); - } + consensus.with_pipeline_mut(|pipeline| { + for entry in entries { + pipeline.push(entry); + } + }); } /// Snapshot this replica's uncommitted suffix into consensus, if the journal has diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index b683af2bc0..96b766a244 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -23,7 +23,7 @@ use crate::{IggyShard, LifecycleFrame, Receiver, RestorableMetadataStm, ShardFra use consensus::{MetadataHandle, PartitionsHandle}; use crossfire::TrySendError; use futures::FutureExt; -use iggy_binary_protocol::{ConsensusHeader, GenericHeader, Operation, PrepareHeader}; +use iggy_binary_protocol::{GenericHeader, Operation, PrepareHeader}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn}; @@ -37,76 +37,6 @@ use server_common::{Message, MessageBag}; /// so the simulator can advance its virtual clock in whole tick intervals. pub const CONSENSUS_TICK_INTERVAL: std::time::Duration = std::time::Duration::from_millis(10); -/// Decompose a [`MessageBag`] into the routing-relevant tuple -/// `(operation, namespace, generic_message)`. -/// -/// Single source of truth used by every dispatch entry point so the -/// operation / namespace extraction never drifts between call sites. -fn extract_routing(bag: MessageBag) -> (Operation, u64, Message) { - match bag { - MessageBag::Request(r) => { - let h = *r.header(); - (h.operation, h.group, r.into_generic()) - } - MessageBag::Prepare(p) => { - let h = *p.header(); - (h.operation, h.group, p.into_generic()) - } - MessageBag::PrepareOk(p) => { - let h = *p.header(); - (h.operation, h.group, p.into_generic()) - } - MessageBag::StartViewChange(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::DoViewChange(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::StartView(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::Commit(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::RequestStartView(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::RequestPrepares(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::RepairPrepare(m) => { - let h = *m.header(); - (h.0.operation, h.0.group, m.into_generic()) - } - MessageBag::RepairRangeReply(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::RequestStateTransfer(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::StateTransferTarget(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::RequestStateChunk(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - MessageBag::StateChunk(m) => { - let h = *m.header(); - (h.operation(), h.group, m.into_generic()) - } - } -} - /// Inter-shard dispatch logic. /// /// All messages — whether destined for a local or remote shard — are routed @@ -122,20 +52,12 @@ where /// Network-receive entry point. Classifies the raw /// `Message` and routes it to the owning shard via /// `route_typed`. - // - // TODO(hubcio): perf - this `MessageBag::try_from` is run twice per - // consensus frame: once here to extract (operation, namespace) for - // routing, and a second time on the receiving shard inside `on_message` - // (lib.rs ~560) to dispatch to the correct on_* handler. The second - // parse re-runs `bytemuck::checked::try_from_bytes` + per-header - // `validate()` on bytes already validated upstream. Measured ~50 ns/ - // frame; at 1M ops/sec/shard ~ 50 ms/sec/shard of pure re-validation. - // - // Fix: thread the classified bag through `ShardFrame::Consensus` (carry - // `MessageBag` instead of `Message`) so the inbox path - // matches directly with no second parse. Consensus variant grows from - // ~24 B to ~32 B, but `ShardFrame` total stays at 160 B (LifecycleFrame - // drives the union size). + /// + /// The only classify a frame gets: the bag rides + /// [`ShardFrame::Consensus`] to the owning shard, whose pump matches it + /// directly. Reading routing off it and then handing the bytes on generic + /// made every frame pay `bytemuck::checked::try_from_bytes` plus the + /// header's `validate()` twice. pub fn dispatch(&self, message: Message) { let bag = match MessageBag::try_from(message) { Ok(bag) => bag, @@ -154,8 +76,8 @@ where return; } }; - let (operation, namespace, generic) = extract_routing(bag); - self.route_typed(operation, namespace, generic); + let (operation, namespace) = bag.routing(); + self.route_typed(operation, namespace, bag); } /// Invoke the client-request handler directly, exactly as the client-fd @@ -176,7 +98,7 @@ where /// group is deterministic across the cluster. pub(crate) fn route_consensus_control( &self, - message: Message, + message: MessageBag, namespace_u64: u64, operation: Operation, ) { @@ -198,12 +120,7 @@ where /// - `METADATA_GROUP` -> shard 0. /// - packable `IggyNamespace::inner()` -> the shard owning that /// partition's consensus group. - fn route_typed( - &self, - operation: Operation, - namespace_u64: u64, - generic: Message, - ) { + fn route_typed(&self, operation: Operation, namespace_u64: u64, generic: MessageBag) { if operation.is_metadata() { self.try_send_to_target(0, generic, operation); return; @@ -259,12 +176,7 @@ where /// a trusted index) is dropped with `reason=unroutable` rather than /// panicking. Metadata frames always pass `target = 0` here, since /// `is_metadata` operations are owned by shard 0. - fn try_send_to_target( - &self, - target: u16, - message: Message, - operation: Operation, - ) { + fn try_send_to_target(&self, target: u16, message: MessageBag, operation: Operation) { let variant = if operation.is_partition() { frame_drop_variant::PARTITION } else { diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index d4c2ba0c79..e77d0832f0 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -2831,6 +2831,118 @@ mod tests { ); } + /// The realloc half of the PR #3557 class, and the stronger one: a pump + /// `insert` that grows the partitions vec MOVES every element, so a stale + /// reference to any partition dangles, not just the one a `swap_remove` + /// displaced. `shell_detects_partition_borrow_held_across_await` covers the + /// remove; this covers the grow, on the same two-task deterministic + /// interleave (reconciler-shaped reader parked with a borrow live, pump-shaped + /// task mutating the container underneath it). + /// + /// The buffer address is asserted to have MOVED, so the test cannot pass on a + /// push into spare capacity, which relocates nothing. + /// + /// Debug-only, like the tripwire it drives: `BorrowGuard` compiles out in + /// release. + #[cfg(debug_assertions)] + #[test] + fn shell_detects_partition_borrow_held_across_a_pump_realloc() { + use crate::executor::DetExecutor; + use consensus::PartitionsHandle; + use std::panic::{AssertUnwindSafe, catch_unwind}; + + server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + + let network_opts = packet::PacketSimulatorOptions { + node_count: 3, + client_count: 1, + seed: 0x5CED_0022, + ..packet::PacketSimulatorOptions::default() + }; + let mut sim = Simulator::new(3, std::iter::once(1u128), network_opts); + let ns_a = IggyNamespace::new(0, 0, 0); + let ns_b = IggyNamespace::new(0, 0, 1); + // The namespace the pump grows the vec with. Never materialised up front: + // inserting it IS the mutation under test. + let ns_grow = IggyNamespace::new(0, 0, 2); + sim.init_partition(ns_a); + sim.init_partition(ns_b); + + // BAD read: the borrow is live across the suspension, so the pump's + // growing insert lands while a stale reference to every partition is + // outstanding. `catch_unwind` builds the executor inline so unwinding + // drops the parked read's guard and restores the borrow count. + let tripped = catch_unwind(AssertUnwindSafe(|| { + let mut executor = DetExecutor::new(11); + let read = Rc::clone(&sim.replicas[0].shards[0]); + executor.spawn(async move { + read.plane + .partitions() + .hold_borrow_across_await(std::future::pending()) + .await; + }); + executor.run_until_stalled(POLL_BUDGET); // borrow acquired; task parks + let grow = Rc::clone(&sim.replicas[0].shards[0]); + executor.spawn(async move { + grow.init_partition(ns_grow, None, None); + }); + executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live + })) + .is_err(); + assert!( + tripped, + "a pump realloc under a live partition borrow went undetected: the \ + #3557 tripwire did not fire on the growing insert" + ); + // The tripwire asserts before `push`, so the vec is untouched: the two + // originals survive and the grow namespace never materialised. + let partitions = sim.replicas[0].shards[0].plane.partitions(); + assert_eq!( + partitions.len(), + 2, + "tripwire must abort the insert before it relocates the vec" + ); + assert!(!partitions.contains(&ns_grow)); + + // REAL read: `with_partition` drops the borrow before the suspension, so + // the identical schedule is sound and the grow applies. + let addr_before = partitions.buffer_addr(); + let mut executor = DetExecutor::new(11); + let read = Rc::clone(&sim.replicas[0].shards[0]); + executor.spawn(async move { + let _ = read + .plane + .partitions() + .with_partition(&ns_a, |_partition| ()); + std::future::pending::<()>().await; + }); + executor.run_until_stalled(POLL_BUDGET); + let grow = Rc::clone(&sim.replicas[0].shards[0]); + executor.spawn(async move { + grow.init_partition(ns_grow, None, None); + }); + executor.run_until_stalled(POLL_BUDGET); + + let partitions = sim.replicas[0].shards[0].plane.partitions(); + assert!( + partitions.contains(&ns_grow), + "correct with_partition read must leave the concurrent insert sound" + ); + assert_ne!( + partitions.buffer_addr(), + addr_before, + "the insert landed in spare capacity, so nothing moved and this test \ + proves nothing about a realloc; seed more partitions before the grow" + ); + // Every pre-existing partition is still addressable after the move, which + // is what a stale reference would have missed. + assert!(partitions.contains(&ns_a) && partitions.contains(&ns_b)); + } + /// Committed metadata prepare timestamps for `seed`: register plus two /// stream creates, read back from replica 0's metadata journal. fn metadata_prepare_timestamps(seed: u64) -> Vec {