From 72bf394ec7ecf0ed086802f43d824cf55844b3cd Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Fri, 17 Jul 2026 15:27:01 +0200 Subject: [PATCH 1/7] fix(metadata): fix metadata pipeline commit --- core/consensus/src/plane_helpers.rs | 22 ++ core/metadata/src/impls/metadata.rs | 302 +++++++++++++++++++++++++--- core/server-ng/src/http/state.rs | 21 +- core/server-ng/src/http/submit.rs | 31 ++- 4 files changed, 333 insertions(+), 43 deletions(-) diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index 0b454c078d..e385566a40 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -255,6 +255,28 @@ where drained } +/// Header of the pipeline head, iff its op is covered by the commit frontier. +/// +/// Peek-only counterpart of [`drain_committable_prefix`] for commit paths that +/// must survive their driving future being canceled between "committable" and +/// "applied" (see `IggyMetadata::on_ack`): the caller peeks here, performs its +/// awaits with the entry still in the pipeline, then — in a sync region — +/// revalidates that the head is still this exact entry before popping and +/// applying it. A driver dropped at an await strands nothing; a sibling driver +/// that committed the op first fails the caller's revalidation and re-peeks. +pub fn peek_committable_head(consensus: &VsrConsensus) -> Option +where + B: MessageBus, + P: Pipeline, +{ + let commit = consensus.commit_max(); + let pipeline = consensus.pipeline().borrow(); + pipeline + .head() + .map(|entry| entry.header) + .filter(|header| header.op <= commit) +} + /// Build reply for a committed prepare. /// /// Every field except `size` comes from `prepare_header`, bytes are identical diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 2cddd1c50f..dd955e16c9 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -27,9 +27,10 @@ use consensus::{ PipelineEntry, Plane, PlaneIdentity, PlaneKind, PreflightOutcome, 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, drain_committable_prefix, - emit_sim_event, fence_old_prepare_by_commit, is_caught_up_primary, - panic_if_hash_chain_would_break_in_same_view, pipeline_prepare_common, register_preflight, + 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, pipeline_prepare_common, + register_preflight, replicate_preflight, replicate_to_next_in_chain, request_preflight, send_eviction_to_client, send_prepare_ok as send_prepare_ok_common, }; @@ -796,22 +797,27 @@ where "ack quorum received" ); - let drained = drain_committable_prefix(consensus); - let drained_count = drained.len(); - if let (Some(first), Some(last)) = (drained.first(), drained.last()) { - debug!( - target: "iggy.metadata.diag", - plane = "metadata", - replica_id = consensus.replica(), - first_op = first.header.op, - last_op = last.header.op, - drained_count = drained_count, - "draining committed metadata prefix" - ); - } - - for mut entry in drained { - let prepare_header = entry.header; + // Commit loop: peek -> await journal read -> revalidate head -> + // sync {pop, apply, advance_commit_min}. + // + // The entry stays in the pipeline across the journal-read await, + // so a driver of this function that is dropped there — a hyper + // HTTP handler future canceled by peer disconnect, or a parked + // in-process submitter — strands nothing: the next driver + // (sibling submit, shard pump, repair tick) re-peeks the same + // head and commits it. Popping BEFORE the await loses the entry + // forever on cancellation (nothing re-applies a popped entry; + // `repair_primary_self_acks` is re-ack-only), pinning + // `commit_min` below `commit_max` and panicking the next commit + // with "commit_min must advance sequentially". + // + // Concurrent drivers are serialized by the head revalidation: + // only the driver that still finds its peeked header at the + // pipeline head after the await owns that op's commit; everyone + // else re-peeks and moves on to the next committable op. + let mut commits = 0usize; + let mut wire_replies: Vec<(CommitLogEvent, Message)> = Vec::new(); + while let Some(prepare_header) = peek_committable_head(consensus) { // TODO(hubcio): should we replace this with graceful fallback (warn + return)? // When journal compaction is implemented compaction could race // with this lookup if it removes entries below the commit number. @@ -826,6 +832,22 @@ 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 + }); + if !head_is_ours { + continue; + } + + let mut entry = consensus + .pipeline() + .borrow_mut() + .pop() + .expect("on_ack: revalidated head exists"); + let pipeline_depth = consensus.pipeline().borrow().len(); let event = CommitLogEvent { replica: ReplicaLogContext::from_consensus(consensus, PlaneKind::Metadata), @@ -841,8 +863,11 @@ where // proof the table is caught up. Table first, counter last: // panic mid-commit leaves the gate closed. // - // Invariant: no .await or panic between client_table.commit_* - // and advance_commit_min. Sync-only. + // Invariant: no .await or panic from the pop above through + // `advance_commit_min` and the subscriber fire below. + // Sync-only — this is what makes pop/apply/advance atomic on + // the single-threaded shard and keeps the head revalidation + // sound. let reply = if prepare_header.operation == Operation::Register { // Register: commit_register creates session, no SM. let reply = build_reply_message(&prepare_header, &bytes::Bytes::new()); @@ -911,10 +936,12 @@ where }; consensus.advance_commit_min(prepare_header.op); emit_sim_event(SimEventKind::OperationCommitted, &event); + commits += 1; // Fire subscriber BEFORE wire send. Slot already updated // (slot-first ordering, see take_reply_sender). Dropped - // receiver: ignored. + // receiver: ignored. Still inside the sync region, so an + // in-process awaiter is woken atomically with its commit. let had_in_process_subscriber = entry.has_reply_sender(); if let Some(sender) = entry.take_reply_sender() { let _ = sender.send(reply.clone()); @@ -926,33 +953,39 @@ where // the same socket. Sending both desyncs the SDK -- it reads // the first frame, fails to decode the typed body, and // leaves the second frame stuck in the socket buffer. - if had_in_process_subscriber { - continue; + if !had_in_process_subscriber { + wire_replies.push((event, reply)); } + } + // Wire replies AFTER the commit loop: this region may await, and + // a driver dropped here loses only reply frames — every commit + // above is applied and its reply cached in the client_table, so + // the SDK recovers it via request replay. + for (event, reply) in wire_replies { let generic_reply = reply.into_generic(); let reply_buffers = freeze_client_reply(generic_reply); emit_sim_event(SimEventKind::ClientReplyEmitted, &event); if let Err(e) = consensus .message_bus() - .send_to_client(prepare_header.client, reply_buffers) + .send_to_client(event.client_id, reply_buffers) .await { error!( - client = prepare_header.client, - op = prepare_header.op, - request_id = prepare_header.request, - operation = ?prepare_header.operation, + client = event.client_id, + op = event.op, + request_id = event.request_id, + operation = ?event.operation, %e, "client reply forward failed, no retransmit path; client will time out", ); } } - // Each commit frees one prepare slot, promote up to - // drained_count buffered requests so the pipeline stays busy. - self.drain_request_queue_into_prepares(drained_count).await; + // Each commit frees one prepare slot, promote up to that many + // buffered requests so the pipeline stays busy. + self.drain_request_queue_into_prepares(commits).await; } } } @@ -2687,4 +2720,209 @@ mod tests { "ServerDefault expiry must be stamped to the configured default at admission" ); } + + /// Bus whose `send_to_client` parks forever while `stall` is set, + /// recording each parked send in `stall_hits`. Models a client whose + /// connection writer stalled, so an `on_ack` driver suspends at a wire + /// send — an await the test can then cancel the driver at. + #[derive(Debug, Default)] + struct StallBus { + stall: std::cell::Cell, + stall_hits: std::cell::Cell, + } + + // Cell fields make the futures !Send; fine on the single-threaded shard. + #[allow(clippy::future_not_send)] + impl MessageBus for StallBus { + fn track_background(&self, _handle: JoinHandle<()>) {} + async fn send_to_client( + &self, + _client_id: u128, + _data: Frozen, + ) -> Result<(), SendError> { + if self.stall.get() { + self.stall_hits.set(self.stall_hits.get() + 1); + std::future::pending::<()>().await; + } + Ok(()) + } + async fn send_to_replica( + &self, + _replica: u8, + _data: Frozen, + ) -> Result<(), SendError> { + Ok(()) + } + fn set_connection_lost_fn(&self, _f: ConnectionLostFn) {} + fn set_replica_forward_fn(&self, _f: ReplicaForwardFn) {} + fn set_client_forward_fn(&self, _f: ClientForwardFn) {} + } + + fn create_stream_request(client: u128, request: u64, name: &str) -> Message { + let body = iggy_binary_protocol::requests::streams::CreateStreamRequest { + name: WireName::new(name).unwrap(), + } + .to_bytes(); + let header_size = size_of::(); + let total = header_size + body.len(); + let mut message = Message::::new(total); + { + let slice = message.as_mut_slice(); + slice[header_size..total].copy_from_slice(&body); + let header = + bytemuck::checked::from_bytes_mut::(&mut slice[..header_size]); + *header = RequestHeader { + command: Command2::Request, + operation: Operation::CreateStream, + size: u32::try_from(total).unwrap(), + client, + session: 1, + request, + user_id: 0, + namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE, + ..Default::default() + }; + } + message + } + + /// Reproduces the `commit_min must advance sequentially` shard-0 crash. + /// + /// `on_ack` drains (pops) the committable prefix off the pipeline and only + /// then applies it, with awaits in between (journal read, wire send). Any + /// driver of `on_ack` that is dropped at one of those awaits — a hyper + /// HTTP handler future canceled by peer disconnect (`http/state.rs`), or + /// any parked in-process submitter — strands the popped-but-unapplied + /// entries: nothing can re-apply them (`repair_primary_self_acks` is + /// re-ack-only, above `commit_max`), `commit_min` is pinned below + /// `commit_max` (every login rejected `NotCaughtUp`), and the next commit + /// that quorums panics the shard. + /// + /// The test parks a driver mid-commit exactly there, cancels it, and then + /// delivers the next ack. Correct behavior: the stranded op is still in + /// the pipeline and the late ack commits it and everything after it, in + /// order. Broken behavior: panic "expected 2, got 3". + #[compio::test] + async fn dropped_on_ack_driver_must_not_lose_popped_commits() { + use std::future::Future; + + const CLIENT: u128 = 1; + // The session is the Register op's commit number; it must be > 0 and + // sort at-or-under the commits of the three ops below (1..=3). + 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_CONSENSUS_NAMESPACE, + StallBus::default(), + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new(Some(consensus), Some(journal), 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), + |_| false, + ); + + // Three prepares through the real primary path: pipeline entry, WAL + // append, self-ack onto the loopback queue. + for (i, name) in ["s1", "s2", "s3"].iter().enumerate() { + let prepare = md + .prepare_request(create_stream_request(CLIENT, i as u64 + 1, name)) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + } + let mut loopback = Vec::new(); + consensus.drain_loopback_into(&mut loopback); + let mut acks = loopback + .into_iter() + .map(|message| { + message + .try_into_typed::() + .expect("loopback holds self PrepareOks") + }) + .collect::>(); + assert_eq!(acks.len(), 3, "one self-ack per replicated prepare"); + let ack3 = acks.pop().unwrap(); + let ack2 = acks.pop().unwrap(); + let ack1 = acks.pop().unwrap(); + + // Ack op 2 first: quorum for op 2 alone, but the contiguous prefix + // still starts at the un-acked op 1, so nothing commits yet. + md.on_ack(ack2).await; + assert_eq!(consensus.commit_max(), 0); + assert_eq!(consensus.commit_min(), 0); + + // Ack op 1: the quorum walk covers ops 1..=2, so this single driver + // commits both. Poll it by hand until it parks at a stalled wire + // send mid-`on_ack`, then cancel it — the moral equivalent of hyper + // dropping an HTTP handler future on peer disconnect. + consensus.message_bus().stall.set(true); + { + let mut driver = Box::pin(md.on_ack(ack1)); + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + let mut parked_at_send = false; + for _ in 0..1_000 { + assert!( + driver.as_mut().poll(&mut cx).is_pending(), + "driver must park at the stalled wire send, not complete" + ); + if consensus.message_bus().stall_hits.get() > 0 { + parked_at_send = true; + break; + } + // Let the runtime process the journal-read completion the + // driver is waiting on, then poll again. + compio::time::sleep(std::time::Duration::from_millis(1)).await; + } + assert!( + parked_at_send, + "driver never reached a wire send (commit_min={})", + consensus.commit_min() + ); + // Cancel mid-`on_ack`, with at least op 1 applied and the reply + // send in flight. + drop(driver); + } + consensus.message_bus().stall.set(false); + assert_eq!(consensus.commit_max(), 2, "quorum walk advanced commit_max"); + assert!( + consensus.commit_min() >= 1, + "driver applied op 1 before parking at the wire send" + ); + + // Whatever the canceled driver left behind must still be + // committable: delivering the ack for op 3 has to commit every + // remaining op, in order. The broken commit path lost op 2 with the + // dropped driver (popped, never applied) and panics here with + // "commit_min must advance sequentially: expected 2, got 3". + md.on_ack(ack3).await; + assert_eq!( + consensus.commit_min(), + 3, + "late ack must commit the stranded op 2 and then op 3" + ); + assert_eq!(consensus.commit_max(), 3); + assert!( + is_caught_up_primary(consensus), + "gate must reopen once the prefix is fully applied" + ); + } } diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs index 55df0d8e86..777518946e 100644 --- a/core/server-ng/src/http/state.rs +++ b/core/server-ng/src/http/state.rs @@ -28,6 +28,7 @@ use axum::http::{HeaderName, HeaderValue}; use axum::response::Response; use configs::server_ng::NgSystemConfig; use consensus::{MetadataHandle, VsrConsensus}; +use futures::channel::oneshot; use iggy_common::{ClusterMetadata, IggyTimestamp}; use message_bus::InstanceToken; use send_wrapper::SendWrapper; @@ -205,8 +206,26 @@ impl HttpInner { } // Shared Register entry point; on shard 0 (always, for HTTP) it runs // `submit_register_in_process` directly on the metadata owner. - let session = submit_register_on_owner(&self.shard, client_id, user_id) + // + // Detached so a client disconnect cannot cancel the Register + // mid-flight: the in-process submit drives shared consensus machinery + // (pipeline push, WAL append, the `on_ack` commit loop), and hyper + // drops this handler future the moment the HTTP peer disconnects. + // A canceled submit used to strand consensus state mid-await; now the + // detached task always drives it to completion and a disconnect only + // drops the receiver half (same discipline as `submit_committed`). + let (result_slot, committed) = oneshot::channel(); + let shard = Rc::clone(&self.shard); + compio::runtime::spawn(async move { + let result = submit_register_on_owner(&shard, client_id, user_id).await; + // A failed send means the handler died mid-await; the Register + // itself has already committed, which is what matters. + let _ = result_slot.send(result); + }) + .detach(); + let session = committed .await + .map_err(|_| AuthError::SessionUnavailable)? .map_err(|error| { warn!(?error, "server-ng HTTP: VSR Register submit failed"); AuthError::SessionUnavailable diff --git a/core/server-ng/src/http/submit.rs b/core/server-ng/src/http/submit.rs index 6e549a5992..f343f546f6 100644 --- a/core/server-ng/src/http/submit.rs +++ b/core/server-ng/src/http/submit.rs @@ -260,18 +260,29 @@ pub(in crate::http) async fn logout_session(state: &HttpInner, session: &Rc {} + Ok(Err(error)) => warn!( ?error, "server-ng HTTP: VSR Logout submit failed; slot lingers until eviction" - ); + ), + Err(_canceled) => warn!( + "server-ng HTTP: VSR Logout task dropped before replying; slot lingers until eviction" + ), } state.forget_session(session); } From e6d749f46594e642ac3608d2b91caaf94853a127 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Fri, 17 Jul 2026 16:00:07 +0200 Subject: [PATCH 2/7] cargo fmt --- core/metadata/src/impls/metadata.rs | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index b777c72de2..c27dcc9e6c 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -30,9 +30,8 @@ use consensus::{ 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, pipeline_prepare_common, - register_preflight, - replicate_preflight, replicate_to_next_in_chain, request_preflight, send_eviction_to_client, - send_prepare_ok as send_prepare_ok_common, + register_preflight, replicate_preflight, replicate_to_next_in_chain, request_preflight, + send_eviction_to_client, send_prepare_ok as send_prepare_ok_common, }; use iggy_binary_protocol::WireIdentifier; use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment; @@ -2817,12 +2816,10 @@ mod tests { 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 journal = + journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0) + .await + .unwrap(); let consensus = VsrConsensus::new( 1, 0, @@ -2833,7 +2830,13 @@ mod tests { ); consensus.init(); let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = - IggyMetadata::new(Some(consensus), Some(journal), None, TestMux::default(), None); + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + None, + ); let consensus = md.consensus.as_ref().unwrap(); md.client_table.borrow_mut().commit_register( From 8a0de0174011b95a702cba48bb5603145d543ba5 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Sun, 19 Jul 2026 21:32:39 +0200 Subject: [PATCH 3/7] further fixes --- Cargo.lock | 2 + core/configs/src/lib.rs | 4 +- core/configs/src/server_ng_config/defaults.rs | 14 + core/configs/src/server_ng_config/displays.rs | 14 +- core/configs/src/server_ng_config/metadata.rs | 183 +++++++++++ core/configs/src/server_ng_config/mod.rs | 1 + .../configs/src/server_ng_config/server_ng.rs | 2 + .../src/server_ng_config/validators.rs | 3 + core/consensus/src/impls.rs | 41 ++- core/journal/Cargo.toml | 1 + core/journal/src/prepare_journal.rs | 150 ++++++++- core/metadata/Cargo.toml | 1 + core/metadata/src/impls/metadata.rs | 311 +++++++++++++++++- core/metadata/src/impls/recovery.rs | 48 ++- core/server-ng/config.toml | 19 ++ core/server-ng/src/bootstrap.rs | 26 +- 16 files changed, 775 insertions(+), 45 deletions(-) create mode 100644 core/configs/src/server_ng_config/metadata.rs diff --git a/Cargo.lock b/Cargo.lock index 22cbf3cfe2..84f6ad7bdc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7658,6 +7658,7 @@ version = "0.1.0" dependencies = [ "bytemuck", "compio", + "futures", "iggy_binary_protocol", "server_common", "tempfile", @@ -8368,6 +8369,7 @@ dependencies = [ "bytes", "compio", "consensus", + "futures", "iggy_binary_protocol", "iggy_common", "journal", diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs index 4d454f56f3..c9322f002d 100644 --- a/core/configs/src/lib.rs +++ b/core/configs/src/lib.rs @@ -30,6 +30,6 @@ pub use server_config::{ tcp, validators, websocket, }; pub use server_ng_config::{ - COMPONENT_NG, message_bus, quic as ng_quic, server_ng, sharding as ng_sharding, tcp as ng_tcp, - websocket as ng_websocket, + COMPONENT_NG, message_bus, metadata as ng_metadata, quic as ng_quic, server_ng, + sharding as ng_sharding, tcp as ng_tcp, websocket as ng_websocket, }; diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs index c7e7249345..1c9b20483e 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -28,6 +28,7 @@ //! bootstrap. use super::message_bus::MessageBusConfig; +use super::metadata::MetadataConfig; use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig}; use super::server_ng::NgSystemConfig; use super::server_ng::{ExtraConfig, ServerNgConfig}; @@ -62,11 +63,24 @@ impl Default for ServerNgConfig { http: HttpConfig::default(), telemetry: TelemetryConfig::default(), cluster: ClusterConfig::default(), + metadata: MetadataConfig::default(), message_bus: MessageBusConfig::default(), } } } +impl Default for MetadataConfig { + fn default() -> MetadataConfig { + // Read from the embedded TOML so the Default impl and the on-disk + // schema cannot drift (same pattern as MessageBusConfig below). + let metadata = &SERVER_NG_CONFIG.metadata; + MetadataConfig { + prepare_queue_depth: metadata.prepare_queue_depth as usize, + journal_slots: metadata.journal_slots as usize, + } + } +} + impl Default for QuicConfig { fn default() -> QuicConfig { QuicConfig { diff --git a/core/configs/src/server_ng_config/displays.rs b/core/configs/src/server_ng_config/displays.rs index 8fa0b5e3d4..022e4e311c 100644 --- a/core/configs/src/server_ng_config/displays.rs +++ b/core/configs/src/server_ng_config/displays.rs @@ -23,6 +23,7 @@ //! section formatter. use super::message_bus::MessageBusConfig; +use super::metadata::MetadataConfig; use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig}; use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig}; use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig}; @@ -34,7 +35,7 @@ impl Display for ServerNgConfig { f, "{{ consumer_group: {}, data_maintenance: {}, extra: {}, message_saver: {}, \ heartbeat: {}, system: {}, quic: {}, tcp: {}, http: {}, telemetry: {}, \ - message_bus: {} }}", + metadata: {}, message_bus: {} }}", self.consumer_group, self.data_maintenance, self.extra, @@ -45,11 +46,22 @@ impl Display for ServerNgConfig { self.tcp, self.http, self.telemetry, + self.metadata, self.message_bus, ) } } +impl Display for MetadataConfig { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{{ prepare_queue_depth: {}, journal_slots: {} }}", + self.prepare_queue_depth, self.journal_slots, + ) + } +} + impl Display for MessageBusConfig { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!( diff --git a/core/configs/src/server_ng_config/metadata.rs b/core/configs/src/server_ng_config/metadata.rs new file mode 100644 index 0000000000..bfcb253369 --- /dev/null +++ b/core/configs/src/server_ng_config/metadata.rs @@ -0,0 +1,183 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! On-disk schema for the metadata consensus plane (shard 0's VSR +//! replica: users, streams, topics, sessions). +//! +//! Two capacity knobs previously hardcoded in the runtime crates: +//! +//! - `prepare_queue_depth` -> `consensus::PIPELINE_PREPARE_QUEUE_MAX` +//! (the pipeline's in-flight prepare bound; submits beyond it bounce +//! with the transient "metadata prepare queue is full") +//! - `journal_slots` -> `journal::prepare_journal::DEFAULT_SLOT_COUNT` +//! (the WAL's in-memory index; committed-but-unsnapshotted headroom +//! between forced checkpoints) +//! +//! The two interlock through the forced-checkpoint margin +//! (`max(64, prepare_queue_depth)` at bootstrap): while a checkpoint +//! runs, up to a full prepare queue of already-pipelined ops appends +//! into that margin, and `validate` keeps `journal_slots` far enough +//! above it that checkpoints stay rare instead of back-to-back. +//! +//! The defaults are duplicated literals rather than imports so +//! `core/configs` does not grow build-time edges onto `core/consensus` +//! and `core/journal` (the runtime crates are the consumers of this +//! config, mirroring the `IOV_MAX_LIMIT_NG` precedent in +//! [`super::message_bus`]). `core/server-ng`'s bootstrap pins both +//! literals against the runtime constants with static asserts. + +use super::COMPONENT_NG; +use crate::ConfigurationError; +use configs::ConfigEnv; +use iggy_common::Validatable; +use serde::{Deserialize, Serialize}; + +/// Mirrors `consensus::PIPELINE_PREPARE_QUEUE_MAX`. +pub const DEFAULT_METADATA_PREPARE_QUEUE_DEPTH: usize = 32; + +/// Mirrors `journal::prepare_journal::DEFAULT_SLOT_COUNT`. +pub const DEFAULT_METADATA_JOURNAL_SLOTS: usize = 1024; + +/// Floor of the forced-checkpoint margin +/// (`metadata::SnapshotCoordinator::CHECKPOINT_MARGIN`); the effective +/// margin is `max(this, prepare_queue_depth)`. +pub const METADATA_CHECKPOINT_MARGIN_FLOOR: usize = 64; + +/// Upper bound on `prepare_queue_depth`. Every queued prepare pins a +/// full message buffer; four thousand in-flight metadata ops is far past +/// any sane deployment and a likely unit typo. +pub const MAX_METADATA_PREPARE_QUEUE_DEPTH: usize = 4096; + +/// Upper bound on `journal_slots`. Each slot costs index memory and every +/// checkpoint rewrites the live WAL suffix; a million slots is the sanity +/// ceiling, not a tuning target. +pub const MAX_METADATA_JOURNAL_SLOTS: usize = 1 << 20; + +/// Capacity tunables for the metadata consensus plane. +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +pub struct MetadataConfig { + /// Depth of the metadata prepare queue: how many uncommitted metadata + /// ops may be in flight at once. Submits beyond it are rejected with + /// the transient "metadata prepare queue is full" (SDK retries). + pub prepare_queue_depth: usize, + + /// Size of the metadata WAL's in-memory index, in slots (one + /// committed-but-unsnapshotted op per slot). Headroom between forced + /// checkpoints; more slots = rarer checkpoints, more memory, larger + /// per-checkpoint WAL rewrites. + pub journal_slots: usize, +} + +impl MetadataConfig { + /// The forced-checkpoint margin bootstrap installs for this config: + /// the built-in floor, raised to the configured prepare-queue depth. + #[must_use] + pub const fn checkpoint_margin(&self) -> usize { + if self.prepare_queue_depth > METADATA_CHECKPOINT_MARGIN_FLOOR { + self.prepare_queue_depth + } else { + METADATA_CHECKPOINT_MARGIN_FLOOR + } + } +} + +impl Validatable for MetadataConfig { + fn validate(&self) -> Result<(), ConfigurationError> { + if self.prepare_queue_depth == 0 { + eprintln!("{COMPONENT_NG} metadata.prepare_queue_depth must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.prepare_queue_depth > MAX_METADATA_PREPARE_QUEUE_DEPTH { + eprintln!( + "{COMPONENT_NG} metadata.prepare_queue_depth ({}) exceeds the maximum ({MAX_METADATA_PREPARE_QUEUE_DEPTH})", + self.prepare_queue_depth + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.journal_slots > MAX_METADATA_JOURNAL_SLOTS { + eprintln!( + "{COMPONENT_NG} metadata.journal_slots ({}) exceeds the maximum ({MAX_METADATA_JOURNAL_SLOTS})", + self.journal_slots + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + // The journal must comfortably out-size the checkpoint margin: + // at `journal_slots == margin` every single prepare would force a + // checkpoint, and below it the journal could wrap. 4x keeps + // checkpoints amortized over at least 3/4 of the journal. + let min_slots = 4 * self.checkpoint_margin(); + if self.journal_slots < min_slots { + eprintln!( + "{COMPONENT_NG} metadata.journal_slots ({}) must be >= 4 * max({METADATA_CHECKPOINT_MARGIN_FLOOR}, prepare_queue_depth) = {min_slots}", + self.journal_slots + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_are_valid() { + let config = MetadataConfig { + prepare_queue_depth: DEFAULT_METADATA_PREPARE_QUEUE_DEPTH, + journal_slots: DEFAULT_METADATA_JOURNAL_SLOTS, + }; + assert!(config.validate().is_ok()); + assert_eq!(config.checkpoint_margin(), METADATA_CHECKPOINT_MARGIN_FLOOR); + } + + #[test] + fn margin_tracks_deep_prepare_queue() { + let config = MetadataConfig { + prepare_queue_depth: 256, + journal_slots: 4096, + }; + assert!(config.validate().is_ok()); + assert_eq!(config.checkpoint_margin(), 256); + } + + #[test] + fn journal_must_outsize_margin() { + // Deep queue, journal kept at the old default: margin becomes 256, + // 4 * 256 = 1024 == journal_slots, boundary accepted... + let boundary = MetadataConfig { + prepare_queue_depth: 256, + journal_slots: 1024, + }; + assert!(boundary.validate().is_ok()); + // ...one slot fewer is refused. + let starved = MetadataConfig { + prepare_queue_depth: 256, + journal_slots: 1023, + }; + assert!(starved.validate().is_err()); + } + + #[test] + fn zero_depth_is_refused() { + let config = MetadataConfig { + prepare_queue_depth: 0, + journal_slots: DEFAULT_METADATA_JOURNAL_SLOTS, + }; + assert!(config.validate().is_err()); + } +} diff --git a/core/configs/src/server_ng_config/mod.rs b/core/configs/src/server_ng_config/mod.rs index e3196e521f..16de8fdea8 100644 --- a/core/configs/src/server_ng_config/mod.rs +++ b/core/configs/src/server_ng_config/mod.rs @@ -30,6 +30,7 @@ pub mod defaults; pub mod displays; pub mod message_bus; +pub mod metadata; pub mod quic; pub mod server_ng; pub mod sharding; diff --git a/core/configs/src/server_ng_config/server_ng.rs b/core/configs/src/server_ng_config/server_ng.rs index 0323f6df64..e60bcc894a 100644 --- a/core/configs/src/server_ng_config/server_ng.rs +++ b/core/configs/src/server_ng_config/server_ng.rs @@ -17,6 +17,7 @@ use super::COMPONENT_NG; use super::message_bus::MessageBusConfig; +use super::metadata::MetadataConfig; use super::quic::QuicConfig; use super::tcp::TcpConfig; use super::websocket::WebSocketConfig; @@ -76,6 +77,7 @@ pub struct ServerNgConfig { pub websocket: WebSocketConfig, pub telemetry: TelemetryConfig, pub cluster: ClusterConfig, + pub metadata: MetadataConfig, pub message_bus: MessageBusConfig, } diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index 0ab7939d3d..1c9745b8cf 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -80,6 +80,9 @@ impl Validatable for ServerNgConfig { self.cluster.validate().error(|e: &ConfigurationError| { format!("{COMPONENT_NG} (error: {e}) - failed to validate cluster config") })?; + self.metadata.validate().error(|e: &ConfigurationError| { + format!("{COMPONENT_NG} (error: {e}) - failed to validate metadata config") + })?; self.system .logging .validate() diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 30c8db27e2..72c95b80da 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -250,10 +250,17 @@ impl RequestEntry { /// Two-queue pipeline: in-flight prepares + buffered requests. #[derive(Debug)] pub struct LocalPipeline { - /// Uncommitted prepares; cap [`PIPELINE_PREPARE_QUEUE_MAX`]. + /// Uncommitted prepares; cap [`Self::prepare_queue_max`]. prepare_queue: VecDeque, - /// Requests awaiting a prepare slot; cap [`PIPELINE_REQUEST_QUEUE_MAX`]. + /// Requests awaiting a prepare slot; cap [`Self::request_queue_max`]. request_queue: VecDeque, + /// Depth bound for `prepare_queue`; [`PIPELINE_PREPARE_QUEUE_MAX`] + /// unless the operator overrode it (`[metadata]` in the server-ng + /// config). + prepare_queue_max: usize, + /// Depth bound for `request_queue`; [`PIPELINE_REQUEST_QUEUE_MAX`] + /// unless overridden alongside `prepare_queue_max`. + request_queue_max: usize, } impl Default for LocalPipeline { @@ -265,9 +272,31 @@ impl Default for LocalPipeline { impl LocalPipeline { #[must_use] pub fn new() -> Self { + Self::with_capacities(PIPELINE_PREPARE_QUEUE_MAX, PIPELINE_REQUEST_QUEUE_MAX) + } + + /// Pipeline with operator-tuned queue depths. + /// + /// Callers wiring this from config must keep the journal's + /// checkpoint margin >= `prepare_queue_max`: up to a full prepare + /// queue of already-pipelined ops appends while a forced checkpoint + /// runs, and the margin is what guarantees them journal room (see + /// `SnapshotCoordinator` in `core/metadata`). + /// + /// # Panics + /// If a depth is zero — a zero-depth pipeline can never admit an op. + #[must_use] + pub fn with_capacities(prepare_queue_max: usize, request_queue_max: usize) -> Self { + assert!( + prepare_queue_max > 0 && request_queue_max > 0, + "pipeline queue depths must be non-zero \ + (prepare={prepare_queue_max}, request={request_queue_max})" + ); Self { - prepare_queue: VecDeque::with_capacity(PIPELINE_PREPARE_QUEUE_MAX), - request_queue: VecDeque::with_capacity(PIPELINE_REQUEST_QUEUE_MAX), + prepare_queue: VecDeque::with_capacity(prepare_queue_max), + request_queue: VecDeque::with_capacity(request_queue_max), + prepare_queue_max, + request_queue_max, } } @@ -278,7 +307,7 @@ impl LocalPipeline { #[must_use] pub fn prepare_queue_full(&self) -> bool { - self.prepare_queue.len() >= PIPELINE_PREPARE_QUEUE_MAX + self.prepare_queue.len() >= self.prepare_queue_max } #[must_use] @@ -288,7 +317,7 @@ impl LocalPipeline { #[must_use] pub fn request_queue_full(&self) -> bool { - self.request_queue.len() >= PIPELINE_REQUEST_QUEUE_MAX + self.request_queue.len() >= self.request_queue_max } #[must_use] diff --git a/core/journal/Cargo.toml b/core/journal/Cargo.toml index 220eca23f2..fb76787a6b 100644 --- a/core/journal/Cargo.toml +++ b/core/journal/Cargo.toml @@ -35,6 +35,7 @@ iggy_binary_protocol = { workspace = true } server_common = { workspace = true } [dev-dependencies] +futures = { workspace = true } tempfile = { workspace = true } [lints.clippy] diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs index 2117d1341d..802b36d819 100644 --- a/core/journal/src/prepare_journal.rs +++ b/core/journal/src/prepare_journal.rs @@ -48,6 +48,11 @@ const MAX_ENTRY_SIZE: u64 = 64 * 1024 * 1024; /// hold. This may need to be tuned properly. pub(crate) const SLOT_COUNT: usize = 1024; +/// Default in-memory index size, in slots. Overridable per journal via +/// [`PrepareJournal::open_with_slots`] (operator knob: `[metadata] +/// journal_slots` in the server-ng config). +pub const DEFAULT_SLOT_COUNT: usize = SLOT_COUNT; + /// Error type for journal operations. #[derive(Debug)] #[allow(clippy::module_name_repetitions)] @@ -129,6 +134,19 @@ pub struct PrepareJournal { /// set; first-write-wins matches the actual semantics and avoids /// `RefCell` borrow-panic risk on the read fast path. poisoned: OnceCell, + /// True while a `drain()` is rewriting the WAL. Concurrent drains + /// share the one `wal.tmp` swap and race it (truncated tmp, ENOENT + /// on the losing rename, reads through a reopened fd), so overlap is + /// refused up front with `ResourceBusy` instead. The upper layer + /// serializes checkpoints anyway (metadata `journal_gate`); this is + /// the journal's own defense so no future caller can reintroduce the + /// race silently. + drain_in_flight: Cell, + /// Number of slots in the in-memory index; ops map to `op % slot_count`. + /// [`DEFAULT_SLOT_COUNT`] unless the operator overrode it. Larger values + /// let more committed-but-unsnapshotted entries accumulate between + /// checkpoints (more WAL churn headroom, more memory). + slot_count: usize, } /// Captured cause of journal poisoning. `stage` names the drain step @@ -149,8 +167,8 @@ impl fmt::Debug for PrepareJournal { } #[allow(clippy::cast_possible_truncation)] -const fn slot_for_op(op: u64) -> usize { - op as usize % SLOT_COUNT +const fn slot_for_op(op: u64, slot_count: usize) -> usize { + op as usize % slot_count } /// Repair a damaged WAL tail by truncating to `pos`, or surface a loud @@ -226,6 +244,17 @@ impl Drop for TmpFileGuard { } } +/// Clears `drain_in_flight` on every exit path of `drain()` — success, +/// `?` error, or caller cancellation at any await. A stuck flag would +/// refuse every future drain and let the journal fill for good. +struct DrainInFlightGuard<'a>(&'a Cell); + +impl Drop for DrainInFlightGuard<'_> { + fn drop(&mut self) { + self.0.set(false); + } +} + #[allow(clippy::cast_possible_truncation)] impl PrepareJournal { /// Open the WAL file in read-write mode, scanning forward to rebuild @@ -242,15 +271,44 @@ impl PrepareJournal { /// Returns `JournalError::Io` if the WAL file cannot be opened or read. #[allow(clippy::future_not_send)] pub async fn open(path: &Path, snapshot_op: u64) -> Result { + Self::open_with_slots(path, snapshot_op, DEFAULT_SLOT_COUNT).await + } + + /// [`Self::open`] with an operator-tuned index size. + /// + /// `slot_count` bounds how many committed-but-unsnapshotted entries the + /// journal holds before a forced checkpoint must reclaim WAL space; the + /// caller owns keeping it above its checkpoint margin + prepare-queue + /// depth (validated at config load for the server-ng `[metadata]` knob). + /// + /// # Errors + /// Returns `JournalError::Io` if the WAL file cannot be opened or read, + /// or (`InvalidInput`) if `slot_count` is zero. + #[allow(clippy::future_not_send)] + pub async fn open_with_slots( + path: &Path, + snapshot_op: u64, + slot_count: usize, + ) -> Result { + if slot_count == 0 { + return Err(JournalError::Io(io::Error::new( + io::ErrorKind::InvalidInput, + "journal slot_count must be non-zero", + ))); + } let storage = FileStorage::open(path).await?; - Self::scan(storage, snapshot_op).await + Self::scan(storage, snapshot_op, slot_count).await } #[allow(clippy::future_not_send)] - async fn scan(storage: FileStorage, snapshot_op: u64) -> Result { + async fn scan( + storage: FileStorage, + snapshot_op: u64, + slot_count: usize, + ) -> Result { let file_len = storage.file_len(); - let mut headers: Vec> = vec![None; SLOT_COUNT]; - let mut offsets: Vec> = vec![None; SLOT_COUNT]; + let mut headers: Vec> = vec![None; slot_count]; + let mut offsets: Vec> = vec![None; slot_count]; let mut last_op: Option = None; let mut pos: u64 = 0; let mut header_buf = vec![0u8; HEADER_SIZE]; @@ -306,7 +364,7 @@ impl PrepareJournal { break; } - let slot = slot_for_op(header.op); + let slot = slot_for_op(header.op, slot_count); // Note: Regarding duplicate op in WAL. We rewrite it with whichever // is the latest entry. @@ -334,6 +392,8 @@ impl PrepareJournal { last_op: Cell::new(last_op), snapshot_op: Cell::new(snapshot_op), poisoned: OnceCell::new(), + drain_in_flight: Cell::new(false), + slot_count, }) } @@ -433,7 +493,7 @@ impl PrepareJournal { let (offset, size) = { let headers = self.headers.borrow(); let offsets = self.offsets.borrow(); - let slot = slot_for_op(header.op); + let slot = slot_for_op(header.op, self.slot_count); let stored = match headers[slot].as_ref() { Some(h) if h.op == header.op => h, _ => return Ok(None), @@ -467,7 +527,7 @@ impl Journal for PrepareJournal { fn header(&self, idx: usize) -> Option> { let headers = self.headers.borrow(); Ref::filter_map(headers, |h| { - let slot = slot_for_op(idx as u64); + let slot = slot_for_op(idx as u64, self.slot_count); let header = h[slot].as_ref()?; if header.op == idx as u64 { Some(header) @@ -487,14 +547,14 @@ impl Journal for PrepareJournal { fn remaining_capacity(&self) -> Option { let Some(last) = self.last_op.get() else { - return Some(SLOT_COUNT); + return Some(self.slot_count); }; let snapshot = self.snapshot_op.get(); if last <= snapshot { - return Some(SLOT_COUNT); + return Some(self.slot_count); } let used = (last - snapshot) as usize; - Some(SLOT_COUNT.saturating_sub(used)) + Some(self.slot_count.saturating_sub(used)) } /// Remove entries with ops in `ops` from the journal, @@ -507,6 +567,17 @@ impl Journal for PrepareJournal { if let Some(state) = self.poisoned.get() { return Err(Self::poisoned_io_error(state)); } + // Overlapping drains race the `wal.tmp` swap below (truncate each + // other's tmp mid-write, lose the rename to ENOENT, read stale + // offsets through the winner's reopened fd). Refuse up front, + // before any WAL bytes move. + if self.drain_in_flight.replace(true) { + return Err(io::Error::new( + io::ErrorKind::ResourceBusy, + "drain already in flight: concurrent drains would race the WAL rewrite", + )); + } + let _drain_guard = DrainInFlightGuard(&self.drain_in_flight); let end_op = *ops.end(); // Partition slots into drained and live entries. @@ -515,7 +586,7 @@ impl Journal for PrepareJournal { { let headers = self.headers.borrow(); let offsets = self.offsets.borrow(); - for slot in 0..SLOT_COUNT { + for slot in 0..self.slot_count { if let (Some(h), Some(off)) = (&headers[slot], offsets[slot]) { if ops.contains(&h.op) { to_drain.push((*h, off)); @@ -615,11 +686,11 @@ impl Journal for PrepareJournal { let mut offsets = self.offsets.borrow_mut(); let mut pos: u64 = 0; for (header, _) in &live { - let slot = slot_for_op(header.op); + let slot = slot_for_op(header.op, self.slot_count); offsets[slot] = Some(pos); pos += u64::from(header.size); } - for slot in 0..SLOT_COUNT { + for slot in 0..self.slot_count { if let Some(h) = &headers[slot] && ops.contains(&h.op) { @@ -636,7 +707,7 @@ impl Journal for PrepareJournal { return Err(Self::poisoned_io_error(state)); } let header = *entry.header(); - let slot = slot_for_op(header.op); + let slot = slot_for_op(header.op, self.slot_count); // Slot collision must be detected BEFORE `write_append + fsync`: // a post-fsync panic would leave bytes durably on disk, and the @@ -705,7 +776,7 @@ impl Journal for PrepareJournal { let (size, offset) = { let headers = self.headers.borrow(); let offsets = self.offsets.borrow(); - let slot = slot_for_op(header.op); + let slot = slot_for_op(header.op, self.slot_count); let stored = headers[slot].as_ref()?; if stored.op != header.op { return None; @@ -1246,4 +1317,49 @@ mod tests { assert!(journal.header(2).is_some()); assert_eq!(journal.iter_headers_from(1).len(), 2); } + + #[compio::test] + async fn concurrent_drains_are_refused_not_raced() { + // Two drivers racing `drain()` share the one fixed `wal.tmp`: + // `File::create` truncates the other racer's tmp mid-write, the + // first rename consumes the path, and the loser surfaces ENOENT + // (production: `forced checkpoint failed ... snapshot I/O error: + // No such file or directory`) — or, with luckier timing, both + // renames "succeed" over each other's bytes. Contract: exactly + // one drain runs; a concurrent call is refused with + // `ResourceBusy` before it touches the WAL; the journal stays + // healthy either way. + let dir = tempdir().unwrap(); + let path = dir.path().join("journal.wal"); + let journal = PrepareJournal::open(&path, 0).await.unwrap(); + for op in 1..=8 { + journal.append(make_prepare(op, 64)).await.unwrap(); + } + + let (a, b) = futures::join!(journal.drain(1..=4), journal.drain(1..=4)); + let results = [a.map(|v| v.len()), b.map(|v| v.len())]; + let (winners, losers): (Vec<_>, Vec<_>) = + results.into_iter().partition(std::result::Result::is_ok); + assert_eq!( + winners.len(), + 1, + "exactly one concurrent drain may perform the WAL rewrite: {winners:?} / {losers:?}" + ); + assert_eq!(winners[0].as_ref().unwrap(), &4, "winner drains ops 1..=4"); + assert_eq!( + losers[0].as_ref().unwrap_err().kind(), + io::ErrorKind::ResourceBusy, + "loser must be refused up front, not fail mid-flight on the shared tmp: {losers:?}" + ); + + // The journal must remain fully usable after the refused call. + assert!(journal.poisoned.get().is_none(), "refusal must not poison"); + for op in 5..=8 { + assert!(journal.header(op).is_some(), "live op {op} lost"); + } + journal.append(make_prepare(9, 32)).await.unwrap(); + let h = *journal.header(9).unwrap(); + let entry = journal.entry_at(&h).await.unwrap().unwrap(); + assert_eq!(entry.header().op, 9); + } } diff --git a/core/metadata/Cargo.toml b/core/metadata/Cargo.toml index 357ca65d73..681ff73ead 100644 --- a/core/metadata/Cargo.toml +++ b/core/metadata/Cargo.toml @@ -57,6 +57,7 @@ tracing = { workspace = true } [dev-dependencies] bytes = { workspace = true } compio = { workspace = true } +futures = { workspace = true } tempfile = { workspace = true } [lints.clippy] diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index c27dcc9e6c..1a6603975b 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -203,11 +203,18 @@ impl Snapshot for IggySnapshot { pub struct SnapshotCoordinator { data_dir: std::path::PathBuf, create_snapshot: fn(&M, u64, u64) -> Result, + /// Remaining-journal-slots threshold at which a checkpoint is forced. + /// Defaults to [`Self::CHECKPOINT_MARGIN`]; bootstrap raises it to at + /// least the configured prepare-queue depth (see the static assert and + /// [`Self::set_checkpoint_margin`]). + checkpoint_margin: Cell, } impl SnapshotCoordinator { - /// Number of remaining journal slots at which a checkpoint is forced. - // TODO: tune this margin size + /// Default number of remaining journal slots at which a checkpoint is + /// forced. Must stay >= the prepare-queue depth: the ops already + /// pipelined while a checkpoint runs skip it and append into this + /// margin. const CHECKPOINT_MARGIN: usize = 64; #[must_use] @@ -218,9 +225,19 @@ impl SnapshotCoordinator { Self { data_dir, create_snapshot, + checkpoint_margin: Cell::new(Self::CHECKPOINT_MARGIN), } } + /// Raise (never lower) the forced-checkpoint margin. Bootstrap calls + /// this with the configured prepare-queue depth so a deeper pipeline + /// keeps its guarantee of journal room while a checkpoint runs; the + /// default margin stays the floor. + pub fn set_checkpoint_margin(&self, margin: usize) { + self.checkpoint_margin + .set(margin.max(Self::CHECKPOINT_MARGIN)); + } + /// Create a snapshot, persist it, and drain snapshotted entries from the /// journal to reclaim WAL space. /// @@ -270,7 +287,7 @@ impl SnapshotCoordinator { let needs_checkpoint = journal .handle() .remaining_capacity() - .is_some_and(|c| c <= Self::CHECKPOINT_MARGIN); + .is_some_and(|c| c <= self.checkpoint_margin.get()); if needs_checkpoint { self.checkpoint(stm, journal, commit_op, created_at).await?; @@ -281,6 +298,84 @@ impl SnapshotCoordinator { } } +// A checkpoint pauses journal reclamation, not admission: while one runs, +// up to a full prepare queue of already-pipelined ops still needs journal +// room (their `on_replicate` drivers skip the in-flight checkpoint and +// append). The margin must cover them or the journal wraps mid-checkpoint. +const _: () = + assert!(SnapshotCoordinator::<()>::CHECKPOINT_MARGIN >= consensus::PIPELINE_PREPARE_QUEUE_MAX); + +/// Single-shard async gate serializing the journal-mutation section of +/// `on_replicate` (forced checkpoint + WAL append). +/// +/// Many futures can drive `on_replicate` concurrently on one shard (the +/// pump loop, detached per-client submit tasks, repair). Ungated they race +/// `SnapshotCoordinator::checkpoint`: every driver crossing the +/// `remaining_capacity <= CHECKPOINT_MARGIN` boundary runs a full +/// checkpoint, and the concurrent `journal.drain()` calls collide on the +/// WAL rewrite — shared `wal.tmp`, ENOENT for every rename that loses, +/// short reads after the winner's reopen. Appends racing a drain are just +/// as unsound: the drain's live-set partition misses an append landing +/// mid-rewrite and the rewrite silently discards it. +/// +/// Not a general lock: single-threaded (`Cell`/`RefCell`, never `Sync`), +/// release wakes every waiter and poll order re-races (arrival-order FIFO +/// under `futures::join!`-style drivers), cancel-safe (dropping the guard +/// releases; dropping a waiter leaves only a stale waker). +struct LocalGate { + busy: Cell, + waiters: RefCell>, +} + +impl LocalGate { + const fn new() -> Self { + Self { + busy: Cell::new(false), + waiters: RefCell::new(Vec::new()), + } + } + + const fn acquire(&self) -> LocalGateAcquire<'_> { + LocalGateAcquire { gate: self } + } +} + +struct LocalGateAcquire<'a> { + gate: &'a LocalGate, +} + +impl<'a> std::future::Future for LocalGateAcquire<'a> { + type Output = LocalGateGuard<'a>; + + fn poll( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll { + if self.gate.busy.get() { + // Re-polls while still busy push a duplicate waker; the extra + // wake is spurious and harmless at pipeline-queue scale. + self.gate.waiters.borrow_mut().push(cx.waker().clone()); + std::task::Poll::Pending + } else { + self.gate.busy.set(true); + std::task::Poll::Ready(LocalGateGuard { gate: self.gate }) + } + } +} + +struct LocalGateGuard<'a> { + gate: &'a LocalGate, +} + +impl Drop for LocalGateGuard<'_> { + fn drop(&mut self) { + self.gate.busy.set(false); + for waker in self.gate.waiters.borrow_mut().drain(..) { + waker.wake(); + } + } +} + /// Failures shared by the in-process metadata submit helpers. /// /// Returned by [`IggyMetadata::submit_register_in_process`], @@ -378,6 +473,9 @@ pub struct IggyMetadata { pub allocator: ConsensusGroupAllocator, /// Snapshot coordinator - present when persistent checkpointing is configured. pub coordinator: Option>, + /// Serializes `on_replicate`'s journal-mutation section (forced + /// checkpoint + WAL append) across concurrent drivers. See [`LocalGate`]. + journal_gate: LocalGate, /// Per-client session state (sessions, dedup, eviction). Metadata-only. pub client_table: RefCell, /// Late-bound post-commit notifier. Fires once per committed normal op @@ -427,6 +525,7 @@ where mux_stm, allocator, coordinator, + journal_gate: LocalGate::new(), client_table: RefCell::new(ClientTable::new(CLIENTS_TABLE_MAX)), commit_notifier: RefCell::new(None), default_max_topic_size: Cell::new(u64::MAX), @@ -450,6 +549,16 @@ impl IggyMetadata { self.default_max_topic_size.set(max_topic_size_bytes); } + /// Raise the forced-checkpoint margin to cover a configured + /// prepare-queue depth (`[metadata] prepare_queue_depth`). Clamped to + /// the built-in floor by the coordinator; no-op on shards without a + /// coordinator. + pub fn set_checkpoint_margin(&self, margin: usize) { + if let Some(coordinator) = &self.coordinator { + coordinator.set_checkpoint_margin(margin); + } + } + /// Resolved byte value for `MaxTopicSize::ServerDefault`. #[must_use] pub const fn default_max_topic_size(&self) -> u64 { @@ -663,9 +772,25 @@ where panic_if_hash_chain_would_break_in_same_view(&previous, &header); } - if !self.checkpoint_if_needed(consensus, journal).await { - return; - } + // Serialize the journal-mutation section (forced checkpoint + append) + // across concurrent `on_replicate` drivers. Ungated, every driver + // crossing the checkpoint boundary ran its own checkpoint and the + // concurrent `drain()`s raced the WAL rewrite (`snapshot I/O error: + // No such file or directory`) — the single-node "metadata prepare + // queue is full" wedge. Held through the append so a drain can never + // rewrite the WAL out from under a racing append either. + let journal_gate = self.journal_gate.acquire().await; + + // Best-effort WAL reclamation. A failed checkpoint must NOT drop the + // prepare: `pipeline_message` already pushed the pipeline entry and + // pre-advanced the sequencer, so bailing out here leaves a phantom op + // that no repair path re-prepares — the commit frontier gaps behind + // it permanently and the pipeline wedges full. `CHECKPOINT_MARGIN >= + // PIPELINE_PREPARE_QUEUE_MAX` (static assert above) guarantees the + // append below still has room after a failed or skipped attempt; a + // journal that truly wraps is refused by append's slot-collision + // guard, not here. + self.checkpoint_if_needed(consensus, journal).await; // Backup: gap check (op == current_op + 1). // Primary: sequencer pre-advanced by push_prepare_entry (guards @@ -724,6 +849,9 @@ where return; } + // Journal mutation done; wire traffic below must not hold the gate. + drop(journal_gate); + // Durable; chain-replicate. `replicate` borrows + freezes; we keep // message for the sequencer/checksum bookkeeping below. self.replicate(&message).await; @@ -1731,10 +1859,17 @@ where Error = iggy_common::IggyError, >, { + /// Run a forced checkpoint when the journal is low on capacity. + /// + /// Diagnostic-only outcome: the caller holds the `journal_gate`, so this + /// is single-flight by construction, and a failure is deliberately not + /// surfaced as control flow — the prepare being replicated must proceed + /// to its append regardless (see the phantom-op comment at the call + /// site). The next prepare over the boundary simply retries. #[allow(clippy::future_not_send)] - async fn checkpoint_if_needed(&self, consensus: &VsrConsensus, journal: &J) -> bool { + async fn checkpoint_if_needed(&self, consensus: &VsrConsensus, journal: &J) { let Some(coordinator) = &self.coordinator else { - return true; + return; }; // Use commit_min (locally executed), not commit_max. WAL entries @@ -1757,9 +1892,8 @@ where checkpoint_op = snap_op, "forced checkpoint completed" ); - true } - Ok(false) => true, + Ok(false) => {} Err(e) => { error!( target: "iggy.metadata.diag", @@ -1767,9 +1901,8 @@ where replica_id = consensus.replica(), checkpoint_op = snap_op, error = %e, - "forced checkpoint failed" + "forced checkpoint failed; continuing without WAL reclamation" ); - false } } } @@ -2932,4 +3065,158 @@ mod tests { "gate must reopen once the prefix is fully applied" ); } + + /// Reproduces the single-node "metadata prepare queue is full" wedge + /// (laserdata cloud-core suite, 2026-07-18). + /// + /// `checkpoint_if_needed` runs inside `on_replicate`, once per submit. + /// Under a concurrent login/create burst, several `on_replicate` futures + /// cross the forced-checkpoint boundary (journal `remaining_capacity <= + /// CHECKPOINT_MARGIN`, i.e. op `SLOT_COUNT - MARGIN = 960`) together, and + /// every one of them runs a full checkpoint concurrently. The concurrent + /// `journal.drain()` calls race on the one fixed `wal.tmp`: the losers + /// surface `snapshot I/O error: No such file or directory` (or a short + /// read after the winner's reopen). Fatally, `on_replicate` then dropped + /// the loser's prepare — AFTER `pipeline_message` had pushed the pipeline + /// entry and pre-advanced the sequencer — so the op was never journaled, + /// never acked, never committed. The commit frontier gaps permanently: + /// logins first bounce `NotCaughtUp`, in-flight clients wedge + /// (`InProgress` on logout), and once the pipeline fills every submit is + /// rejected `PipelineFull` forever. + /// + /// Correct behavior: checkpoints are single-flight, a failed or skipped + /// checkpoint never discards a pipelined prepare, all racers' ops are + /// journaled and commit, and the caught-up gate reopens. + #[compio::test] + async fn concurrent_checkpoint_boundary_must_not_drop_prepares() { + const CLIENT: u128 = 1; + const SESSION: u64 = 1; + const ACTING_USER: u32 = 7; + /// One op below the forced-checkpoint trigger: the journal holds + /// 1024 slots and forces a checkpoint when 64 or fewer remain. + const FILL: u64 = 960; + + let dir = tempfile::tempdir().unwrap(); + // Bootstrap creates the metadata dir; the coordinator's snapshot + // persist expects it to exist. + std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).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_CONSENSUS_NAMESPACE, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + // `data_dir` present => SnapshotCoordinator armed, checkpoints live. + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + Some(dir.path().to_path_buf()), + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table.borrow_mut().commit_register( + CLIENT, + ACTING_USER, + register_reply(CLIENT, SESSION), + |_| false, + ); + + // Fill to one op under the boundary through the real primary path, + // acking each op so `commit_min` tracks `last_op` and the pipeline + // stays shallow — the steady state the production server was in. + let mut loopback = Vec::new(); + for i in 1..=FILL { + let prepare = md + .prepare_request(create_stream_request(CLIENT, i, &format!("s{i}"))) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + loopback.clear(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("one self-ack per prepare") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + md.on_ack(ack).await; + } + assert_eq!(consensus.commit_min(), FILL); + assert_eq!( + md.journal.as_ref().unwrap().remaining_capacity(), + Some(64), + "fill must stop exactly at the forced-checkpoint boundary" + ); + + // Three submits race across the boundary — the concurrent + // login/create burst from the incident. Every racer sees + // `remaining_capacity <= CHECKPOINT_MARGIN` before any drain + // completes. + let md_ref = &md; + let race = |request: u64, name: String| async move { + let prepare = md_ref + .prepare_request(create_stream_request(CLIENT, request, &name)) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md_ref.on_replicate(prepare).await; + }; + futures::join!( + race(FILL + 1, format!("s{}", FILL + 1)), + race(FILL + 2, format!("s{}", FILL + 2)), + race(FILL + 3, format!("s{}", FILL + 3)), + ); + + // Every racer's prepare must be durably journaled: a dropped one is + // unrepairable (nothing re-prepares it) and gaps the frontier. + let journal = md.journal.as_ref().unwrap(); + for op in FILL + 1..=FILL + 3 { + assert!( + journal.header(op as usize).is_some(), + "op {op} vanished from the WAL: a failed checkpoint dropped a pipelined prepare" + ); + } + // The checkpoint itself must have happened — once: WAL reclaimed, + // snapshot on disk. + assert!( + journal.remaining_capacity().unwrap() > 900, + "checkpoint must have drained the snapshotted prefix, got {:?}", + journal.remaining_capacity() + ); + assert!( + dir.path() + .join(crate::impls::METADATA_DIR) + .join("snapshot.bin") + .exists(), + "checkpoint must persist the snapshot" + ); + + // The self-acks commit all three racers; any gap here is the + // production wedge (commit frontier pinned, PipelineFull forever). + loopback.clear(); + consensus.drain_loopback_into(&mut loopback); + assert_eq!(loopback.len(), 3, "one self-ack per racer"); + for message in loopback.drain(..) { + let ack = message + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + md.on_ack(ack).await; + } + assert_eq!( + consensus.commit_min(), + FILL + 3, + "commit frontier must cross the checkpoint boundary" + ); + assert!( + is_caught_up_primary(consensus), + "caught-up gate must reopen after the boundary" + ); + } } diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs index f9a4d1132c..88413b14d6 100644 --- a/core/metadata/src/impls/recovery.rs +++ b/core/metadata/src/impls/recovery.rs @@ -129,6 +129,7 @@ pub struct RecoveredMetadata { pub async fn recover( data_dir: &Path, solo: bool, + journal_slots: usize, seed_baseline: impl FnOnce(&M), ) -> Result, RecoveryError> where @@ -160,7 +161,7 @@ where let journal_path = metadata_dir.join("journal.wal"); let watermark = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number); - let journal = PrepareJournal::open(&journal_path, watermark).await?; + let journal = PrepareJournal::open_with_slots(&journal_path, watermark, journal_slots).await?; // Intentional fail-fast: a bad entry aborts recovery and the operator // must repair or truncate the WAL before the node can boot again. @@ -266,7 +267,14 @@ mod tests { #[compio::test] async fn recover_empty_state() { let dir = tempdir().unwrap(); - let recovered = recover::(dir.path(), false, |_| {}).await.unwrap(); + let recovered = recover::( + dir.path(), + false, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); assert_eq!(recovered.last_applied_op, None); assert!(recovered.journal.last_op().is_none()); @@ -283,7 +291,14 @@ mod tests { .persist(&metadata_dir.join("snapshot.bin")) .unwrap(); - let recovered = recover::(dir.path(), false, |_| {}).await.unwrap(); + let recovered = recover::( + dir.path(), + false, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); assert_eq!( recovered .snapshot @@ -310,7 +325,14 @@ mod tests { journal.storage_ref().fsync().await.unwrap(); } - let recovered = recover::(dir.path(), false, |_| {}).await.unwrap(); + let recovered = recover::( + dir.path(), + false, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); assert!(recovered.snapshot.is_none()); // Op 3's entry proves commit=2; op 3 itself is journaled but not // provably committed, so it stays journal-only. @@ -344,7 +366,14 @@ mod tests { journal.storage_ref().fsync().await.unwrap(); } - let recovered = recover::(dir.path(), false, |_| {}).await.unwrap(); + let recovered = recover::( + dir.path(), + false, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); assert_eq!(recovered.last_applied_op, Some(3)); assert_eq!(recovered.last_journaled_op, Some(5)); assert_eq!(recovered.journal.last_op(), Some(5)); @@ -373,7 +402,14 @@ mod tests { journal.storage_ref().fsync().await.unwrap(); } - let recovered = recover::(dir.path(), false, |_| {}).await.unwrap(); + let recovered = recover::( + dir.path(), + false, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + |_| {}, + ) + .await + .unwrap(); // Replays ops 6-9 (snapshot at 5; op 10's entry proves commit=9). assert_eq!(recovered.last_applied_op, Some(9)); assert_eq!(recovered.last_journaled_op, Some(10)); diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index bb62085759..f29ebbb6a1 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -697,6 +697,25 @@ cert_file = "core/certs/iggy_cert.pem" key_file = "core/certs/iggy_key.pem" # Message bus configuration. +# Metadata consensus plane tunables (shard 0's VSR replica: users, +# streams, topics, sessions). Size these together: a deeper prepare queue +# admits more concurrent in-flight metadata ops (e.g. login storms), and +# the journal must hold enough slots that a forced checkpoint (triggered +# when remaining slots fall to the checkpoint margin, which itself is +# max(64, prepare_queue_depth)) stays rare. Validation enforces +# journal_slots >= 4 * max(64, prepare_queue_depth). +[metadata] +# Depth of the metadata prepare queue: how many uncommitted metadata ops +# may be in flight at once. Submits beyond it are rejected with the +# transient "metadata prepare queue is full" and retried by the SDK. +prepare_queue_depth = 32 + +# Size of the metadata WAL's in-memory index, in slots (one committed but +# not-yet-snapshotted op per slot). Larger values buy more headroom +# between forced checkpoints at the cost of memory and bigger WAL +# rewrites per checkpoint. +journal_slots = 1024 + # Tunables for the inter-shard / inter-replica internal bus that ships # consensus traffic between replicas and SDK-client traffic between # shards. These knobs are consensus-liveness-critical (max_batch gates diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 8a9ea14e6a..eaee9c873b 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -888,6 +888,7 @@ async fn shard_main( let recovered = recover::( data_dir, topology.replica_count == 1, + config.metadata.journal_slots, |mux_stm| { ensure_default_root_user(mux_stm); }, @@ -959,6 +960,7 @@ async fn shard_main( topology.self_replica_id, topology.replica_count, Rc::clone(&bus), + config.metadata.prepare_queue_depth, ); (Some(consensus), Some(journal), snapshot) } else { @@ -975,6 +977,10 @@ async fn shard_main( // message expiry) at admission; every shard's copy backs the same resolution in responses. metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64()); metadata.set_default_message_expiry(u64::from(config.system.topic.message_expiry)); + // Keep the forced-checkpoint margin >= the configured prepare-queue + // depth: ops already pipelined while a checkpoint runs append into that + // margin (config validation keeps journal_slots >= 4x this). + metadata.set_checkpoint_margin(config.metadata.checkpoint_margin()); let shard_metrics = ShardMetrics::for_shard(); // Notifier install deferred until after tick handler wires below. @@ -1675,6 +1681,20 @@ async fn build_shard_for_thread( Ok((shard, sessions)) } +// Pin the configs-crate default literals (duplicated there to avoid a +// build-time edge onto the runtime crates) against the runtime constants, +// mirroring the message_bus IOV_MAX pin. A drift on either side fails this +// crate's build until both are reconciled. +const _: () = assert!( + configs::ng_metadata::DEFAULT_METADATA_PREPARE_QUEUE_DEPTH + == consensus::PIPELINE_PREPARE_QUEUE_MAX +); +const _: () = assert!( + configs::ng_metadata::DEFAULT_METADATA_JOURNAL_SLOTS + == journal::prepare_journal::DEFAULT_SLOT_COUNT +); + +#[allow(clippy::too_many_arguments)] fn restore_metadata_consensus( journal: &PrepareJournal, restored_op: u64, @@ -1683,6 +1703,7 @@ fn restore_metadata_consensus( self_replica_id: u8, replica_count: u8, bus: Rc, + prepare_queue_depth: usize, ) -> VsrConsensus> { let mut consensus = VsrConsensus::new( cluster_id, @@ -1690,7 +1711,10 @@ fn restore_metadata_consensus( replica_count, server_common::sharding::METADATA_CONSENSUS_NAMESPACE, bus, - LocalPipeline::new(), + // Request queue keeps the stock 2x ratio over the prepare queue + // (32 -> 64 at defaults): buffered requests are cheap relative to + // in-flight prepares and drain as prepares commit. + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), ); let last_header = journal From 74fd75e40628d3e93fc9b6ebf0796ffa32edf036 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Sun, 19 Jul 2026 21:51:30 +0200 Subject: [PATCH 4/7] fix clippy --- core/metadata/src/impls/metadata.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 1a6603975b..351f69be71 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -3179,7 +3179,9 @@ mod tests { let journal = md.journal.as_ref().unwrap(); for op in FILL + 1..=FILL + 3 { assert!( - journal.header(op as usize).is_some(), + journal + .header(usize::try_from(op).expect("test ops fit in usize")) + .is_some(), "op {op} vanished from the WAL: a failed checkpoint dropped a pipelined prepare" ); } @@ -3203,7 +3205,7 @@ mod tests { loopback.clear(); consensus.drain_loopback_into(&mut loopback); assert_eq!(loopback.len(), 3, "one self-ack per racer"); - for message in loopback.drain(..) { + for message in loopback { let ack = message .try_into_typed::() .expect("loopback holds self PrepareOks"); From 8bfdd4285bcce316dcee454f630b17e22e33d2b3 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 20 Jul 2026 14:00:49 +0200 Subject: [PATCH 5/7] some moar fixes --- core/consensus/src/impls.rs | 59 +++- core/metadata/src/impls/metadata.rs | 475 ++++++++++++++++++++++++---- core/server-ng/src/dispatch.rs | 347 ++++++++++++++++++++ 3 files changed, 819 insertions(+), 62 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 72c95b80da..250b06d8bc 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -184,13 +184,20 @@ impl PipelineEntry { #[must_use] pub fn with_subscriber(header: PrepareHeader) -> (Self, Receiver>) { let (sender, receiver) = oneshot::channel(); - let entry = Self { + (Self::with_sender(header, sender), receiver) + } + + /// Entry adopting an existing reply sender — used when a request that + /// carried a subscriber through the request queue is promoted into a + /// prepare slot, so the original in-process awaiter keeps its receiver. + #[must_use] + pub fn with_sender(header: PrepareHeader, sender: Sender>) -> Self { + Self { header, ok_from_replicas: BitSet::with_capacity(REPLICAS_MAX), ok_quorum_received: false, reply_sender: Some(sender), - }; - (entry, receiver) + } } /// Take reply sender; caller fires after slot update (slot-first ordering). @@ -235,6 +242,11 @@ pub struct RequestEntry { // age-based filtering. Currently `0`; `pub(crate)` blocks sort-on-stub. #[allow(dead_code)] pub(crate) received_at: i64, + /// 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, + /// preflight rejection at promotion) wakes the receiver with `Canceled`. + pub(crate) reply_sender: Option>>, } impl RequestEntry { @@ -243,8 +255,32 @@ impl RequestEntry { Self { message, received_at: 0, + reply_sender: None, } } + + /// Queued request paired with a fresh receiver that resolves when the + /// promoted prepare commits (`Err(Canceled)` if the entry is dropped + /// first). The in-process absorption path: a submit that arrives while + /// the primary is mid-commit or the prepare queue is full parks here + /// instead of being bounced with a transient error. + #[must_use] + pub fn with_subscriber( + message: Message, + ) -> (Self, Receiver>) { + let (sender, receiver) = oneshot::channel(); + let entry = Self { + message, + received_at: 0, + reply_sender: Some(sender), + }; + (entry, receiver) + } + + /// Take the reply sender for hand-off to the promoted pipeline entry. + pub const fn take_reply_sender(&mut self) -> Option>> { + self.reply_sender.take() + } } /// Two-queue pipeline: in-flight prepares + buffered requests. @@ -1162,6 +1198,23 @@ impl> VsrConsensus { receiver } + /// [`Self::pipeline_message_with_subscriber`] for a promoted queued + /// request: adopts the sender the request carried through the request + /// queue instead of minting a fresh channel, so the awaiter that parked + /// at enqueue time resolves on this prepare's commit. + /// + /// # Panics + /// If not primary (mirrors [`Consensus::pipeline_message`]). + pub fn pipeline_message_with_sender( + &self, + plane: PlaneKind, + message: &Message, + sender: Sender>, + ) { + let entry = PipelineEntry::with_sender(*message.header(), sender); + self.push_prepare_entry(plane, message, entry); + } + #[must_use] pub const fn cluster(&self) -> u128 { self.cluster diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 351f69be71..e8031a5e58 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -942,7 +942,6 @@ where // only the driver that still finds its peeked header at the // pipeline head after the await owns that op's commit; everyone // else re-peeks and moves on to the next committable op. - let mut commits = 0usize; let mut wire_replies: Vec<(CommitLogEvent, Message)> = Vec::new(); while let Some(prepare_header) = peek_committable_head(consensus) { // TODO(hubcio): should we replace this with graceful fallback (warn + return)? @@ -1063,7 +1062,6 @@ where }; consensus.advance_commit_min(prepare_header.op); emit_sim_event(SimEventKind::OperationCommitted, &event); - commits += 1; // Fire subscriber BEFORE wire send. Slot already updated // (slot-first ordering, see take_reply_sender). Dropped @@ -1110,9 +1108,11 @@ where } } - // Each commit frees one prepare slot, promote up to that many - // buffered requests so the pipeline stays busy. - self.drain_request_queue_into_prepares(commits).await; + // Commits freed prepare slots and reopened the catch-up gate; + // promote buffered requests so the pipeline stays busy and + // absorbed submits (queued while this batch was mid-commit) + // dispatch immediately. + self.drain_request_queue_into_prepares().await; } } } @@ -1184,20 +1184,16 @@ where return Ok(session); } - // Status + catch-up gate (see doc). Split variants for telemetry: - // NotPrimary (try peer) vs NotCaughtUp (retry). Caller policy same. - if !is_caught_up_primary(consensus) { - return Err( - if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() { - MetadataSubmitError::NotCaughtUp - } else { - MetadataSubmitError::NotPrimary - }, - ); + // Wrong node: waiting or queueing cannot fix that, the client must + // re-route to the primary. + if !(consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing()) { + return Err(MetadataSubmitError::NotPrimary); } // Mirror wire-path register_preflight: a racing second prepare fails - // check_register on commit. Surface pre-synthesis. + // check_register on commit. 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() @@ -1206,12 +1202,6 @@ where return Err(MetadataSubmitError::InProgress); } - // TODO(pipeline-backpressure): in-process has no request_queue yet; - // terminal on full. Wire path buffers. - if consensus.pipeline().borrow().is_full() { - return Err(MetadataSubmitError::PipelineFull); - } - let request = build_register_request_message(consensus, client_id, user_id); // Wire path runs `RequestHeader::validate` at network boundary; // in-process skips it. debug_assert pins drift. @@ -1222,6 +1212,38 @@ where }, "build_register_request_message produced a header that fails validate()" ); + + // Catch-up gate (Register only: admitting one while a committed op + // is still unapplied races `commit_register`'s session-eq assert) or + // prepare queue full: absorb into the request queue instead of + // bouncing with a transient error. The queued + // entry carries this caller's reply subscriber; the commit path + // promotes it (`drain_request_queue_into_prepares`, which re-runs + // `register_preflight`) 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() { + let (entry, receiver) = consensus::RequestEntry::with_subscriber(request); + if consensus + .pipeline() + .borrow_mut() + .push_request(entry) + .is_err() + { + // Both queues full: honest terminal backpressure. + return Err(MetadataSubmitError::PipelineFull); + } + return match receiver.await { + Ok(reply) => Ok(reply.header().commit), + // Entry dropped before commit: view-change reset or a + // promotion-time preflight rejection. Same re-check as the + // direct path's cancel arm below. + Err(Canceled) => self + .client_table + .borrow() + .get_session(client_id) + .ok_or(MetadataSubmitError::Canceled), + }; + } // `prepare_request` only fails on `!is_client_allowed`; Register is // allowed, so unreachable. Panic loudly on regression instead of // smuggling through wire-eviction. @@ -1284,14 +1306,14 @@ where return Ok(consensus.commit_min()); } - if !is_caught_up_primary(consensus) { - return Err( - if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() { - MetadataSubmitError::NotCaughtUp - } else { - MetadataSubmitError::NotPrimary - }, - ); + // No catch-up gate here: a logout admitted mid-commit-window is + // safe — the wire path has always dispatched non-register ops + // without one, and the per-client dedup below covers the only + // logout-vs-logout race. It simply pipelines behind the in-flight + // batch and commits with it, so a one-shot client's session + // teardown is latency, never an error. + if !(consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing()) { + return Err(MetadataSubmitError::NotPrimary); } if consensus @@ -1302,10 +1324,6 @@ where return Err(MetadataSubmitError::InProgress); } - if consensus.pipeline().borrow().is_full() { - return Err(MetadataSubmitError::PipelineFull); - } - let request = build_logout_request_message(consensus, client_id, session, request); debug_assert!( { @@ -1314,6 +1332,31 @@ where }, "build_logout_request_message produced a header that fails validate()" ); + + // 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() { + let (entry, receiver) = consensus::RequestEntry::with_subscriber(request); + if consensus + .pipeline() + .borrow_mut() + .push_request(entry) + .is_err() + { + return Err(MetadataSubmitError::PipelineFull); + } + return match receiver.await { + Ok(reply) => Ok(reply.header().commit), + Err(Canceled) => { + if self.client_table.borrow().get_session(client_id).is_none() { + Ok(consensus.commit_min()) + } else { + Err(MetadataSubmitError::Canceled) + } + } + }; + } let prepare = self .prepare_request(request) .expect("Operation::Logout is client-allowed; prepare projection cannot fail"); @@ -1539,16 +1582,21 @@ where .as_ref() .expect("submit_request_in_process: consensus only exists on shard 0"); - // Not-primary / not-caught-up is transient: the same request replayed - // once a primary is caught up commits fine. Reply with the explicit - // transient frame (relayed to the socket by the home shard) so the - // client replays immediately rather than waiting out its read-timeout. + // Not-primary is transient: the same request replayed against the + // current primary commits fine. Reply with the explicit transient + // frame (relayed to the socket by the home shard) so the client + // replays immediately rather than waiting out its read-timeout. // `TransientNotAccepted` specifically: the request never entered the // pipeline here, so the client may re-issue it ANYWHERE -- including // under a fresh session after failing over to the current leader -- // without double-apply risk. (`TransientNotCommitted` conversely means // the outcome is unknown and only a same-session replay is safe.) - if !is_caught_up_primary(consensus) { + // + // No catch-up gate: a non-register op admitted mid-commit-window + // simply pipelines behind the in-flight batch (the wire path has + // always done this); the register-specific invariant is guarded in + // `submit_register_in_process` / `register_preflight`. + if !(consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing()) { return Ok(build_result_rejection_reply( &request_header, consensus.commit_max(), @@ -1590,15 +1638,38 @@ where PreflightOutcome::Drop => return Err(MetadataSubmitError::Canceled), } - // Pipeline full: backpressure, not failure. The request was not - // admitted, so `TransientNotAccepted` (re-issuable anywhere). + // Prepare queue full: backpressure, not failure. Absorb into the + // request queue with this caller's reply subscriber; the commit + // path promotes it as slots free up and the await below resolves + // 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() { - return Ok(build_result_rejection_reply( - &request_header, - consensus.commit_max(), - IggyError::TransientNotAccepted.as_code(), - ) - .into_generic()); + let (entry, receiver) = consensus::RequestEntry::with_subscriber(message); + if consensus + .pipeline() + .borrow_mut() + .push_request(entry) + .is_err() + { + return Ok(build_result_rejection_reply( + &request_header, + consensus.commit_max(), + IggyError::TransientNotAccepted.as_code(), + ) + .into_generic()); + } + return match receiver.await { + Ok(reply) => Ok(reply.into_generic()), + // Queued entry dropped (view-change reset) or promoted then + // canceled: outcome unknown, same-session replay only. + Err(Canceled) => Ok(build_result_rejection_reply( + &request_header, + consensus.commit_max(), + IggyError::TransientNotCommitted.as_code(), + ) + .into_generic()), + }; } // The acting-user RBAC stamp lives in the shared `prepare_request` @@ -1649,12 +1720,14 @@ where ) -> Result, Canceled> { consensus.verify_pipeline(); let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare); - // Re-check gate post-subscribe: `pipeline_message_with_subscriber` - // can drop the borrow. No commit-max advance flips the gate today; - // pin against a future await between check and dispatch. + // Register is the one op whose admission requires the catch-up gate + // (session-eq assert at commit); its submit path checks the gate and + // the check-to-dispatch section is synchronous. Non-register ops + // dispatch mid-window by design (they pipeline behind the in-flight + // batch, like the wire path always has). debug_assert!( - is_caught_up_primary(consensus), - "dispatch_prepare_and_await: gate flipped between check and dispatch" + prepare.header().operation != Operation::Register || is_caught_up_primary(consensus), + "dispatch_prepare_and_await: register dispatched with the catch-up gate closed" ); // `on_replicate` awaits: a sibling in-process submit may commit // (commit_min advances) or a view change may land (view advances) while @@ -1797,16 +1870,29 @@ where /// client's request between push and drain (Stale / Duplicate / /// `AlreadyRegistered`). Skipping produces a duplicate prepare and panics. #[allow(clippy::future_not_send)] - async fn drain_request_queue_into_prepares(&self, slots_freed: usize) { + async fn drain_request_queue_into_prepares(&self) { let consensus = self.consensus.as_ref().unwrap(); - for _ in 0..slots_freed { + // Promote while prepare slots exist. Requests are queued for two + // reasons — prepare queue full at arrival, or (in-process register) + // the catch-up gate was closed — so promotion is bounded by slots, + // not by how many commits just freed: a whole burst absorbed during + // 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() { + break; + } let req = consensus.pipeline().borrow_mut().pop_request(); - let Some(req) = req else { break }; + let Some(mut req) = req else { break }; let client_id = req.message.header().client; let session = req.message.header().session; let request = req.message.header().request; let operation = req.message.header().operation; + // If preflight or projection rejects below, dropping `req` (and + // the sender taken from it) wakes an in-process awaiter with + // `Canceled`; its submit path re-checks the client table. + let reply_sender = req.take_reply_sender(); let dispatch = if operation == Operation::Register { register_preflight(consensus, &self.client_table, client_id).await } else { @@ -1838,10 +1924,20 @@ where continue; } }; - pipeline_prepare_common(consensus, PlaneKind::Metadata, prepare, |prepare| { - self.on_replicate(prepare) - }) - .await; + // Mirror `pipeline_prepare_common`, threading the queued + // subscriber into the pipeline entry so the awaiter that parked + // at enqueue time resolves on this prepare's commit. + assert!(!consensus.is_follower(), "promotion: primary only"); + assert!(consensus.is_normal(), "promotion: status must be normal"); + assert!(!consensus.is_syncing(), "promotion: must not be syncing"); + consensus.verify_pipeline(); + match reply_sender { + Some(sender) => { + consensus.pipeline_message_with_sender(PlaneKind::Metadata, &prepare, sender); + } + None => consensus.pipeline_message(PlaneKind::Metadata, &prepare), + } + self.on_replicate(prepare).await; } } } @@ -3221,4 +3317,265 @@ mod tests { "caught-up gate must reopen after the boundary" ); } + + /// The exact window behind the historical "logout/unregister failed + /// ... primary not yet caught up on `commit_journal`" reports + /// (laserdata harness, 2026-07-17..20): a logout submitted while + /// ANOTHER client's op sits between quorum-ack (`commit_max` advanced + /// inside `on_ack`) and apply (`commit_min` behind, driver parked at + /// the journal read). + /// + /// New contract (queue absorption): a logout landing in + /// that window is NOT bounced with `NotCaughtUp` — non-register ops + /// carry no catch-up gate. It pipelines behind the in-flight batch, + /// and its submit's inline loopback pump commits both ops in order. + /// The parked sibling driver then resumes onto an already-drained + /// pipeline and exits via head-revalidation, exercising the + /// concurrent-driver safety of the commit loop. + #[compio::test] + async fn logout_in_mid_commit_window_commits_instead_of_rejecting() { + use std::future::Future; + + /// The client logging out; its session is already committed. + const CLIENT_A: u128 = 1; + /// The client whose in-flight commit closes the gate. + const CLIENT_B: u128 = 2; + 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_CONSENSUS_NAMESPACE, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + None, + ); + let consensus = md.consensus.as_ref().unwrap(); + for client in [CLIENT_A, CLIENT_B] { + md.client_table.borrow_mut().commit_register( + client, + ACTING_USER, + register_reply(client, SESSION), + |_| false, + ); + } + + // B's op: prepared, journaled, self-acked onto the loopback queue. + let prepare = md + .prepare_request(create_stream_request(CLIENT_B, 1, "s1")) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + let mut loopback = Vec::new(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("one self-ack per prepare") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + + // Open the window: the first poll of `on_ack` reaches quorum and + // advances commit_max synchronously, then parks at the journal + // read — commit_min has not moved. This is the exact server state + // every production NotCaughtUp line was emitted from. + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + let mut driver = Box::pin(md.on_ack(ack)); + assert!( + driver.as_mut().poll(&mut cx).is_pending(), + "driver must park at the journal read inside the commit" + ); + assert_eq!(consensus.commit_max(), 1, "quorum advanced commit_max"); + assert_eq!(consensus.commit_min(), 0, "apply has not landed yet"); + assert!( + !is_caught_up_primary(consensus), + "gate must be closed mid-commit" + ); + + // A's logout lands in the window. No gate for non-register ops: it + // pipelines behind B's committing op, and its inline loopback pump + // drives BOTH commits (B's op 1 via head-revalidated takeover, then + // its own op 2) before resolving. + let outcome = md.submit_logout_in_process(CLIENT_A, SESSION, 2).await; + assert_eq!( + outcome, + Ok(2), + "mid-window logout must commit and reply, never bounce NotCaughtUp" + ); + assert_eq!(consensus.commit_min(), 2, "both ops committed in order"); + assert!( + is_caught_up_primary(consensus), + "gate reopens once the batch drains" + ); + + // The parked sibling driver resumes onto a drained pipeline: the + // head it peeked is gone, revalidation sends it out without + // touching commit state. Drive it to completion to prove it. + let mut resumed = false; + for _ in 0..1_000 { + if driver.as_mut().poll(&mut cx).is_ready() { + resumed = true; + break; + } + // Let the runtime deliver the journal-read completion. + compio::time::sleep(std::time::Duration::from_millis(1)).await; + } + assert!(resumed, "parked driver must exit via head-revalidation"); + assert_eq!( + consensus.commit_min(), + 2, + "resumed driver commits nothing new" + ); + assert_eq!( + md.client_table.borrow().get_session(CLIENT_A), + None, + "session removed by the committed logout" + ); + } + + /// Register is the one op that still honors the catch-up gate (its + /// admission races `commit_register`'s session-eq assert against + /// committed-but-unapplied ops). New contract: a register arriving in + /// the mid-commit window is ABSORBED into the pipeline's request queue + /// with its reply subscriber attached, promoted by + /// the commit path once the batch drains, and the caller's await + /// resolves with the committed session — instead of the historical + /// `NotCaughtUp` bounce that one-shot CLI clients surfaced as + /// "Disconnected" login failures. + #[compio::test] + async fn register_in_mid_commit_window_is_queued_then_committed() { + use std::future::Future; + + /// The client whose in-flight commit closes the gate. + const CLIENT_B: u128 = 2; + /// The client registering mid-window. + const CLIENT_C: u128 = 3; + 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_CONSENSUS_NAMESPACE, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + None, + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table.borrow_mut().commit_register( + CLIENT_B, + ACTING_USER, + register_reply(CLIENT_B, SESSION), + |_| false, + ); + + // B's op journaled + self-acked; park its commit mid-window. + let prepare = md + .prepare_request(create_stream_request(CLIENT_B, 1, "s1")) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + let mut loopback = Vec::new(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("one self-ack per prepare") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + let mut driver = Box::pin(md.on_ack(ack)); + assert!(driver.as_mut().poll(&mut cx).is_pending()); + assert_eq!(consensus.commit_max(), 1); + assert_eq!(consensus.commit_min(), 0); + + // C's register lands in the window: absorbed, not bounced. + let mut register = Box::pin(md.submit_register_in_process(CLIENT_C, ACTING_USER)); + assert!( + register.as_mut().poll(&mut cx).is_pending(), + "mid-window register must park in the request queue, not error" + ); + assert_eq!( + consensus.pipeline().borrow().request_queue_len(), + 1, + "register buffered in the request queue" + ); + + // The committing driver drains its batch, then promotes the queued + // register into a prepare (its self-ack lands on the loopback). + let mut resumed = false; + for _ in 0..1_000 { + if driver.as_mut().poll(&mut cx).is_ready() { + resumed = true; + break; + } + compio::time::sleep(std::time::Duration::from_millis(1)).await; + } + 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(), + 0, + "promotion emptied the request queue" + ); + + // Commit the promoted register (production: the shard pump or any + // sibling submit drains this ack) and the parked caller resolves. + loopback.clear(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("promoted register must self-ack") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + md.on_ack(ack).await; + + let mut outcome = None; + for _ in 0..1_000 { + if let std::task::Poll::Ready(result) = register.as_mut().poll(&mut cx) { + outcome = Some(result); + break; + } + compio::time::sleep(std::time::Duration::from_millis(1)).await; + } + assert_eq!( + outcome.expect("absorbed register must resolve"), + Ok(2), + "queued register commits with the next batch; session = commit op" + ); + assert_eq!( + md.client_table.borrow().get_session(CLIENT_C), + Some(2), + "session created by the promoted register" + ); + } } diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 7d796f5035..bdc6a486de 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -2710,3 +2710,350 @@ where .as_ref() .and_then(std::rc::Weak::upgrade) } + +#[cfg(test)] +mod tests { + use super::*; + use consensus::{LocalPipeline, Plane as _, PlaneKind, VsrConsensus}; + use iggy_binary_protocol::requests::streams::CreateStreamRequest; + use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader}; + use iggy_common::variadic; + use journal::prepare_journal::PrepareJournal; + use message_bus::client_listener::RequestHandler; + use message_bus::fd_transfer::DupedFd; + use message_bus::installer::ConnectionInstaller; + use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind}; + use message_bus::replica::listener::MessageHandler; + use message_bus::{ + ClientConnectionLostFn, ClientForwardFn, ConnectionLostFn, JoinHandle, MessageBus, + ReplicaForwardFn, ReplicaHandshakeDoneFn, SendError, + }; + use metadata::impls::metadata::IggySnapshot; + use metadata::stm::stream::Streams; + use metadata::stm::user::Users; + use metadata::{IggyMetadata, MuxStateMachine}; + use partitions::{IggyPartitions, PartitionsConfig}; + use server_common::iobuf::Frozen; + use server_common::sharding::ShardId; + use server_common::{MESSAGE_ALIGN, Message}; + use shard::shards_table::PapayaShardsTable; + use shard::{IggyShard, PartitionConsensusConfig, ReplicaTopology, ShardIdentity}; + use std::cell::RefCell; + use std::future::Future; + use std::mem::size_of; + use std::rc::Rc; + + type TestMux = MuxStateMachine; + type TestShard = IggyShard; + + /// Records every client-bound reply instead of writing to a socket; + /// everything else is a no-op. The two `ShellBus` halves are stubbed. + #[derive(Debug, Clone, Default)] + struct SpyBus { + client_replies: Rc>>, + } + + #[allow(clippy::future_not_send)] + impl MessageBus for SpyBus { + fn track_background(&self, _handle: JoinHandle<()>) {} + async fn send_to_client( + &self, + client_id: u128, + _data: Frozen, + ) -> Result<(), SendError> { + self.client_replies.borrow_mut().push(client_id); + Ok(()) + } + async fn send_to_replica( + &self, + _replica: u8, + _data: Frozen, + ) -> Result<(), SendError> { + Ok(()) + } + fn set_connection_lost_fn(&self, _f: ConnectionLostFn) {} + fn set_replica_forward_fn(&self, _f: ReplicaForwardFn) {} + fn set_client_forward_fn(&self, _f: ClientForwardFn) {} + } + + impl ConnectionInstaller for SpyBus { + fn install_replica_inbound_fd( + &self, + _fd: DupedFd, + _on_message: MessageHandler, + _on_done: ReplicaHandshakeDoneFn, + ) { + } + fn install_replica_outbound_fd( + &self, + _fd: DupedFd, + _replica_id: u8, + _on_message: MessageHandler, + _on_done: ReplicaHandshakeDoneFn, + ) { + } + fn release_replica_handshake_slot(&self, _slot: u64) {} + fn clear_replica_dial_pending(&self, _replica_id: u8) {} + fn install_client_fd( + &self, + _fd: DupedFd, + _meta: ClientConnMeta, + _on_request: RequestHandler, + ) { + } + fn install_client_ws_fd( + &self, + _fd: DupedFd, + _meta: ClientConnMeta, + _on_request: RequestHandler, + ) { + } + fn client_meta(&self, _client_id: u128) -> Option> { + None + } + fn set_client_connection_lost_fn(&self, _f: ClientConnectionLostFn) {} + } + + /// Minimal committed `Register` reply for `ClientTable::commit_register` + /// (reads only `client` and `commit`). + fn register_reply(client: u128, session: u64) -> Message { + let header_size = size_of::(); + let mut reply = Message::::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::( + &mut reply.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes are a valid ReplyHeader"); + *header = ReplyHeader { + client, + request: 0, + commit: session, + command: Command2::Reply, + operation: Operation::Register, + ..Default::default() + }; + reply + } + + fn request_message( + operation: Operation, + client: u128, + session: u64, + request: u64, + body: &[u8], + ) -> Message { + let header_size = size_of::(); + let total = header_size + body.len(); + let mut message = Message::::new(total); + { + let slice = message.as_mut_slice(); + slice[header_size..total].copy_from_slice(body); + let header = + bytemuck::checked::from_bytes_mut::(&mut slice[..header_size]); + *header = RequestHeader { + command: Command2::Request, + operation, + size: u32::try_from(total).expect("test request fits u32"), + client, + session, + request, + user_id: 0, + namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE, + ..Default::default() + }; + } + message + } + + /// Raw prepare for the sibling op, standing in for the crate-private + /// `prepare_request` projection. `user_id` 0 skips the in-apply RBAC + /// gate (server-originated convention), so the op applies cleanly. + fn prepare_message( + operation: Operation, + client: u128, + request: u64, + body: &[u8], + ) -> Message { + let header_size = size_of::(); + let total = header_size + body.len(); + let mut message = Message::::new(total); + { + let slice = message.as_mut_slice(); + slice[header_size..total].copy_from_slice(body); + let header = + bytemuck::checked::from_bytes_mut::(&mut slice[..header_size]); + *header = PrepareHeader { + command: Command2::Prepare, + operation, + size: u32::try_from(total).expect("test prepare fits u32"), + op: 1, + view: 0, + client, + request, + user_id: 0, + checksum: 42, + namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE, + ..Default::default() + }; + } + message + } + + /// Desired-contract test for the production failure chain "CLI stream + /// create succeeded, logout failed: Disconnected" (laserdata harness + /// reports, 2026-07-17..20). + /// + /// Why the logout of a CLI invocation fails during ITS OWN successful + /// `stream create`: the catch-up gate is GLOBAL. The suite runs many + /// CLI invocations against one shared single-node server; each one is + /// three replicated ops (Register, work, Logout). When THIS client's + /// logout frame arrives, some SIBLING client's op is regularly sitting + /// between quorum-ack (`commit_max` advanced inside `on_ack`) and + /// apply (`commit_min` still behind, driver parked at the journal + /// read). `submit_logout_in_process` then rejects `NotCaughtUp`, and + /// `handle_logout_request` swallows the error: no reply frame, session + /// left bound. A one-shot CLI sees only a dead connection — "Problem + /// with server logout / Disconnected" — and exits non-zero although + /// its create committed; the harness retry then trips "already + /// exists". + /// + /// This test rebuilds that interleaving deterministically (client B = + /// the sibling parked mid-commit; client A = the CLI logging out) and + /// asserts the DESIRED contract instead of today's behavior: + /// + /// a client-initiated logout must always produce a reply frame and + /// unbind the transport session, even when the submit gate is + /// closed — teardown is best-effort, the VSR slot may lapse to the + /// eviction sweep. + /// + /// It FAILS on current code (silence + bound session) by design: it is + /// the reproduction to iterate fixes against. + #[compio::test] + async fn logout_rejected_by_closed_gate_must_still_reply_to_client() { + const CLIENT_A: u128 = 1; + const CLIENT_B: u128 = 2; + const SESSION: u64 = 1; + const ACTING_USER: u32 = 7; + const TRANSPORT_A: u128 = 77; + + let dir = tempfile::tempdir().unwrap(); + let journal = PrepareJournal::open(&dir.path().join("journal.wal"), 0) + .await + .unwrap(); + let bus = SpyBus::default(); + let consensus = VsrConsensus::new( + 1, + 0, + 1, + server_common::sharding::METADATA_CONSENSUS_NAMESPACE, + bus.clone(), + LocalPipeline::new(), + ); + consensus.init(); + let metadata: IggyMetadata<_, PrepareJournal, IggySnapshot, TestMux> = IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + None, + ); + let partitions = IggyPartitions::new( + ShardId::new(0), + PartitionsConfig { + messages_required_to_save: 1, + size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64), + enforce_fsync: false, + segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), + encryptor: None, + }, + ); + let shard = Rc::new(TestShard::without_inbox( + ShardIdentity::new(0, "logout-window-test".to_string()), + bus.clone(), + metadata, + partitions, + PapayaShardsTable::new(), + PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()), + )); + let md = shard.plane.metadata(); + let consensus = md.consensus.as_ref().unwrap(); + + // A and B hold committed sessions (as after their CLI logins). + for client in [CLIENT_A, CLIENT_B] { + md.client_table.borrow_mut().commit_register( + client, + ACTING_USER, + register_reply(client, SESSION), + |_| false, + ); + } + // A's transport connection, authenticated + bound — the state a + // CLI connection is in right after its create-stream reply. + let sessions = Rc::new(RefCell::new(SessionManager::new())); + sessions.borrow_mut().ensure_connection( + TRANSPORT_A, + "127.0.0.1:34567".parse().unwrap(), + ClientTransportKind::Tcp, + ); + sessions + .borrow_mut() + .login(TRANSPORT_A, ACTING_USER) + .unwrap(); + sessions + .borrow_mut() + .bind_session(TRANSPORT_A, CLIENT_A, SESSION) + .unwrap(); + + // Sibling B's op: prepared, journaled, self-acked through the real + // replicate path. (The public submit API cannot be used to open + // the window: `dispatch_prepare_and_await` pumps its own loopback + // inline, committing before it returns. Production's window is a + // sibling submit task parked INSIDE `on_ack`'s awaits — modeled + // below by driving `on_ack` by hand.) + let create_body = CreateStreamRequest { + name: iggy_binary_protocol::primitives::identifier::WireName::new("s1").unwrap(), + } + .to_bytes(); + let prepare = prepare_message(Operation::CreateStream, CLIENT_B, 1, &create_body); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + let mut loopback = Vec::new(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("one self-ack per replicated prepare") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + + // Open the window: first poll of `on_ack` advances commit_max at + // quorum, then parks at the journal read — commit_min unchanged. + // Every production NotCaughtUp logout was submitted exactly here. + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + let mut driver = Box::pin(md.on_ack(ack)); + assert!( + driver.as_mut().poll(&mut cx).is_pending(), + "driver must park mid-commit at the journal read" + ); + assert_eq!(consensus.commit_max(), 1); + assert_eq!(consensus.commit_min(), 0); + + // A's logout lands in the window, through the real dispatch path. + let logout = request_message(Operation::Logout, CLIENT_A, SESSION, 2, &[]); + handle_logout_request(&shard, &sessions, TRANSPORT_A, logout).await; + + // DESIRED CONTRACT (red on current code): the client must never be + // left in silence — that silence is what a one-shot CLI reports as + // "Problem with server logout / Disconnected". + assert!( + bus.client_replies.borrow().contains(&TRANSPORT_A), + "logout must produce a reply frame to the client even while the \ + catch-up gate is closed (silence = CLI 'Disconnected', exit 1)" + ); + assert_eq!( + sessions.borrow().get_session(TRANSPORT_A), + None, + "transport session must be unbound by a client-initiated logout; \ + the VSR slot may lapse to the eviction sweep" + ); + } +} From 6e1bd764f7ea62f8ff2759414035ff1c2b7ccbca Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 20 Jul 2026 16:47:59 +0200 Subject: [PATCH 6/7] update commetns --- core/metadata/src/impls/metadata.rs | 4 +--- core/server-ng/src/dispatch.rs | 3 +-- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index e8031a5e58..ef2f294703 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -3163,7 +3163,6 @@ mod tests { } /// Reproduces the single-node "metadata prepare queue is full" wedge - /// (laserdata cloud-core suite, 2026-07-18). /// /// `checkpoint_if_needed` runs inside `on_replicate`, once per submit. /// Under a concurrent login/create burst, several `on_replicate` futures @@ -3319,8 +3318,7 @@ mod tests { } /// The exact window behind the historical "logout/unregister failed - /// ... primary not yet caught up on `commit_journal`" reports - /// (laserdata harness, 2026-07-17..20): a logout submitted while + /// ... primary not yet caught up on `commit_journal`". /// ANOTHER client's op sits between quorum-ack (`commit_max` advanced /// inside `on_ack`) and apply (`commit_min` behind, driver parked at /// the journal read). diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index bdc6a486de..5e8e5bc871 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -2899,8 +2899,7 @@ mod tests { } /// Desired-contract test for the production failure chain "CLI stream - /// create succeeded, logout failed: Disconnected" (laserdata harness - /// reports, 2026-07-17..20). + /// create succeeded, logout failed: Disconnected". /// /// Why the logout of a CLI invocation fails during ITS OWN successful /// `stream create`: the catch-up gate is GLOBAL. The suite runs many From ae8fdb8d88447171fdebcd6bcdf19350fb3357a9 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 20 Jul 2026 20:03:17 +0200 Subject: [PATCH 7/7] caveman stronk --- core/metadata/src/impls/metadata.rs | 576 ++++++++++++++++++---------- core/server-ng/config.toml | 2 +- core/server-ng/src/dispatch.rs | 39 +- core/shard/src/lib.rs | 8 + 4 files changed, 409 insertions(+), 216 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index ef2f294703..061b2dfbf2 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -370,7 +370,11 @@ struct LocalGateGuard<'a> { impl Drop for LocalGateGuard<'_> { fn drop(&mut self) { self.gate.busy.set(false); - for waker in self.gate.waiters.borrow_mut().drain(..) { + // Move the waiters out before waking: `wake()` only schedules under + // compio today, but a waker that ever polled a waiter inline would + // re-enter `acquire`'s `waiters.borrow_mut()` and panic the RefCell. + let waiters = std::mem::take(&mut *self.gate.waiters.borrow_mut()); + for waker in waiters { waker.wake(); } } @@ -914,8 +918,6 @@ where let quorum = ack_quorum_reached(consensus, PlaneKind::Metadata, header); if quorum { - let journal = self.journal.as_ref().unwrap(); - debug!( target: "iggy.metadata.diag", plane = "metadata", @@ -924,195 +926,7 @@ where "ack quorum received" ); - // Commit loop: peek -> await journal read -> revalidate head -> - // sync {pop, apply, advance_commit_min}. - // - // The entry stays in the pipeline across the journal-read await, - // so a driver of this function that is dropped there — a hyper - // HTTP handler future canceled by peer disconnect, or a parked - // in-process submitter — strands nothing: the next driver - // (sibling submit, shard pump, repair tick) re-peeks the same - // head and commits it. Popping BEFORE the await loses the entry - // forever on cancellation (nothing re-applies a popped entry; - // `repair_primary_self_acks` is re-ack-only), pinning - // `commit_min` below `commit_max` and panicking the next commit - // with "commit_min must advance sequentially". - // - // Concurrent drivers are serialized by the head revalidation: - // only the driver that still finds its peeked header at the - // pipeline head after the await owns that op's commit; everyone - // else re-peeks and moves on to the next committable op. - let mut wire_replies: Vec<(CommitLogEvent, Message)> = Vec::new(); - while let Some(prepare_header) = peek_committable_head(consensus) { - // TODO(hubcio): should we replace this with graceful fallback (warn + return)? - // When journal compaction is implemented compaction could race - // with this lookup if it removes entries below the commit number. - let prepare = journal - .handle() - .entry(&prepare_header) - .await - .unwrap_or_else(|| { - panic!( - "on_ack: committed prepare op={} checksum={} must be in journal", - prepare_header.op, prepare_header.checksum - ) - }); - - // 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 - }); - if !head_is_ours { - continue; - } - - let mut entry = consensus - .pipeline() - .borrow_mut() - .pop() - .expect("on_ack: revalidated head exists"); - - let pipeline_depth = consensus.pipeline().borrow().len(); - let event = CommitLogEvent { - replica: ReplicaLogContext::from_consensus(consensus, PlaneKind::Metadata), - op: prepare_header.op, - client_id: prepare_header.client, - request_id: prepare_header.request, - operation: prepare_header.operation, - pipeline_depth, - }; - - // Apply SM + mutate client_table BEFORE advancing commit_min. - // `is_caught_up_primary` reads `commit_min == commit_max` as - // proof the table is caught up. Table first, counter last: - // panic mid-commit leaves the gate closed. - // - // Invariant: no .await or panic from the pop above through - // `advance_commit_min` and the subscriber fire below. - // Sync-only — this is what makes pop/apply/advance atomic on - // the single-threaded shard and keeps the head revalidation - // sound. - let reply = if prepare_header.operation == Operation::Register { - // Register: commit_register creates session, no SM. - let reply = build_reply_message(&prepare_header, &bytes::Bytes::new()); - let in_flight = - |c: u128| consensus.pipeline().borrow().has_message_from_client(c); - self.client_table.borrow_mut().commit_register( - prepare_header.client, - prepare_header.user_id, - reply.clone(), - in_flight, - ); - reply - } else if prepare_header.operation == Operation::Logout { - // Logout unregisters the VSR client session on every replica. - let reply = build_reply_message(&prepare_header, &bytes::Bytes::new()); - self.client_table - .borrow_mut() - .remove_client(prepare_header.client); - // Drop the disconnected client from every consumer group it - // joined and rebalance. Deterministic side-effect of the - // Logout commit, applied identically on every replica. - self.mux_stm.streams().remove_consumer_group_member( - prepare_header.client, - iggy_common::IggyTimestamp::from(prepare_header.timestamp), - ); - reply - } else { - // Normal op: apply SM, commit_reply. `Err` is decode/corruption - // only; a business rejection commits as a deterministic no-op - // whose `code` rides the reply body, replayed on retry. - let apply = gated_apply(&self.mux_stm, prepare).unwrap_or_else(|err| { - panic!( - "on_ack: committed metadata op={} failed to apply: {err}", - prepare_header.op - ); - }); - // Post-commit notifier (e.g. partition reconciler - // wake-up). Filtering by operation is the - // recipient's responsibility. - self.fire_commit_notifier(prepare_header.operation); - let reply = - build_reply_message_with(&prepare_header, apply.reply_body_len(), |dst| { - apply.write_reply_body(dst); - }); - // Cache only if session exists. Client evicted between - // prepare and commit: skip cache (`commit_reply` no-ops), - // wire reply still ships. - let session = self - .client_table - .borrow() - .get_session(prepare_header.client); - if let Some(session) = session { - self.client_table.borrow_mut().commit_reply( - prepare_header.client, - session, - reply.clone(), - ); - } else { - tracing::trace!( - client = prepare_header.client, - op = prepare_header.op, - "on_ack: client evicted while being prepared; emitting reply but skipping cache" - ); - } - reply - }; - consensus.advance_commit_min(prepare_header.op); - emit_sim_event(SimEventKind::OperationCommitted, &event); - - // Fire subscriber BEFORE wire send. Slot already updated - // (slot-first ordering, see take_reply_sender). Dropped - // receiver: ignored. Still inside the sync region, so an - // in-process awaiter is woken atomically with its commit. - let had_in_process_subscriber = entry.has_reply_sender(); - if let Some(sender) = entry.take_reply_sender() { - let _ = sender.send(reply.clone()); - } - - // Skip wire send when an in-process subscriber consumed the - // reply: the caller (e.g. `complete_login_register`, - // `handle_logout_request`) ships its own full-body reply on - // the same socket. Sending both desyncs the SDK -- it reads - // the first frame, fails to decode the typed body, and - // leaves the second frame stuck in the socket buffer. - if !had_in_process_subscriber { - wire_replies.push((event, reply)); - } - } - - // Wire replies AFTER the commit loop: this region may await, and - // a driver dropped here loses only reply frames — every commit - // above is applied and its reply cached in the client_table, so - // the SDK recovers it via request replay. - for (event, reply) in wire_replies { - let generic_reply = reply.into_generic(); - let reply_buffers = freeze_client_reply(generic_reply); - emit_sim_event(SimEventKind::ClientReplyEmitted, &event); - - if let Err(e) = consensus - .message_bus() - .send_to_client(event.client_id, reply_buffers) - .await - { - error!( - client = event.client_id, - op = event.op, - request_id = event.request_id, - operation = ?event.operation, - %e, - "client reply forward failed, no retransmit path; client will time out", - ); - } - } - - // Commits freed prepare slots and reopened the catch-up gate; - // promote buffered requests so the pipeline stays busy and - // absorbed submits (queued while this batch was mid-commit) - // dispatch immediately. - self.drain_request_queue_into_prepares().await; + self.commit_committable_prefix().await; } } } @@ -1419,6 +1233,11 @@ where .as_ref() .expect("submit_complete_revocation_in_process: consensus only exists on shard 0"); + // Deliberately bounce-based (no request-queue absorption, unlike the + // client submit paths above): the caller is the partition + // reconciler's completion loop, which retries on its own tick, and + // parking internal completions would tie up request slots that + // client submits compete for. if !is_caught_up_primary(consensus) { return Err( if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() { @@ -1498,6 +1317,9 @@ where "submit_delete_personal_access_token_in_process: consensus only exists on shard 0", ); + // Deliberately bounce-based (no request-queue absorption): the + // caller is the background PAT cleaner, which simply retries the + // deletion on its next sweep. if !is_caught_up_primary(consensus) { return Err( if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() { @@ -1862,8 +1684,248 @@ where true } - /// Promote up to `slots_freed` buffered requests into prepares after - /// `on_ack` commits a prefix. + /// Commit the committable prefix, ship the resulting wire replies, and + /// promote queued requests into the freed prepare slots. + /// + /// Runs at the tail of every quorum-advancing `on_ack` and from the + /// shard tick via [`Self::resume_stranded_commits`]. Safe under + /// concurrent drivers: ownership of each op is arbitrated by the head + /// revalidation inside the loop. + #[allow(clippy::too_many_lines)] + #[allow(clippy::future_not_send)] + async fn commit_committable_prefix(&self) { + let consensus = self.consensus.as_ref().unwrap(); + let journal = self.journal.as_ref().unwrap(); + + // Commit loop: peek -> await journal read -> revalidate head -> + // sync {pop, apply, advance_commit_min}. + // + // The entry stays in the pipeline across the journal-read await, + // so a driver of this function that is dropped there — a hyper + // HTTP handler future canceled by peer disconnect, or a parked + // in-process submitter — strands nothing: the next driver + // (sibling submit, shard pump, repair tick) re-peeks the same + // head and commits it. Popping BEFORE the await loses the entry + // forever on cancellation (nothing re-applies a popped entry; + // `repair_primary_self_acks` is re-ack-only), pinning + // `commit_min` below `commit_max` and panicking the next commit + // with "commit_min must advance sequentially". + // + // Concurrent drivers are serialized by the head revalidation: + // only the driver that still finds its peeked header at the + // pipeline head after the await owns that op's commit; everyone + // else re-peeks and moves on to the next committable op. + let mut wire_replies: Vec<(CommitLogEvent, Message)> = Vec::new(); + while let Some(prepare_header) = peek_committable_head(consensus) { + // TODO(hubcio): should we replace this with graceful fallback (warn + return)? + // When journal compaction is implemented compaction could race + // with this lookup if it removes entries below the commit number. + let prepare = journal + .handle() + .entry(&prepare_header) + .await + .unwrap_or_else(|| { + panic!( + "on_ack: committed prepare op={} checksum={} must be in journal", + prepare_header.op, prepare_header.checksum + ) + }); + + // 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 + }); + if !head_is_ours { + continue; + } + + let mut entry = consensus + .pipeline() + .borrow_mut() + .pop() + .expect("on_ack: revalidated head exists"); + + let pipeline_depth = consensus.pipeline().borrow().len(); + let event = CommitLogEvent { + replica: ReplicaLogContext::from_consensus(consensus, PlaneKind::Metadata), + op: prepare_header.op, + client_id: prepare_header.client, + request_id: prepare_header.request, + operation: prepare_header.operation, + pipeline_depth, + }; + + // Apply SM + mutate client_table BEFORE advancing commit_min. + // `is_caught_up_primary` reads `commit_min == commit_max` as + // proof the table is caught up. Table first, counter last: + // panic mid-commit leaves the gate closed. + // + // Invariant: no .await or panic from the pop above through + // `advance_commit_min` and the subscriber fire below. + // Sync-only — this is what makes pop/apply/advance atomic on + // the single-threaded shard and keeps the head revalidation + // sound. + let reply = if prepare_header.operation == Operation::Register { + // Register: commit_register creates session, no SM. + let reply = build_reply_message(&prepare_header, &bytes::Bytes::new()); + let in_flight = |c: u128| consensus.pipeline().borrow().has_message_from_client(c); + self.client_table.borrow_mut().commit_register( + prepare_header.client, + prepare_header.user_id, + reply.clone(), + in_flight, + ); + reply + } else if prepare_header.operation == Operation::Logout { + // Logout unregisters the VSR client session on every replica. + let reply = build_reply_message(&prepare_header, &bytes::Bytes::new()); + self.client_table + .borrow_mut() + .remove_client(prepare_header.client); + // Drop the disconnected client from every consumer group it + // joined and rebalance. Deterministic side-effect of the + // Logout commit, applied identically on every replica. + self.mux_stm.streams().remove_consumer_group_member( + prepare_header.client, + iggy_common::IggyTimestamp::from(prepare_header.timestamp), + ); + reply + } else { + // Normal op: apply SM, commit_reply. `Err` is decode/corruption + // only; a business rejection commits as a deterministic no-op + // whose `code` rides the reply body, replayed on retry. + let apply = gated_apply(&self.mux_stm, prepare).unwrap_or_else(|err| { + panic!( + "on_ack: committed metadata op={} failed to apply: {err}", + prepare_header.op + ); + }); + // Post-commit notifier (e.g. partition reconciler + // wake-up). Filtering by operation is the + // recipient's responsibility. + self.fire_commit_notifier(prepare_header.operation); + let reply = + build_reply_message_with(&prepare_header, apply.reply_body_len(), |dst| { + apply.write_reply_body(dst); + }); + // Cache only if session exists. Client evicted between + // prepare and commit: skip cache (`commit_reply` no-ops), + // wire reply still ships. + let session = self + .client_table + .borrow() + .get_session(prepare_header.client); + if let Some(session) = session { + self.client_table.borrow_mut().commit_reply( + prepare_header.client, + session, + reply.clone(), + ); + } else { + tracing::trace!( + client = prepare_header.client, + op = prepare_header.op, + "on_ack: client evicted while being prepared; emitting reply but skipping cache" + ); + } + reply + }; + consensus.advance_commit_min(prepare_header.op); + emit_sim_event(SimEventKind::OperationCommitted, &event); + + // Fire subscriber BEFORE wire send. Slot already updated + // (slot-first ordering, see take_reply_sender). Dropped + // receiver: ignored. Still inside the sync region, so an + // in-process awaiter is woken atomically with its commit. + let had_in_process_subscriber = entry.has_reply_sender(); + if let Some(sender) = entry.take_reply_sender() { + let _ = sender.send(reply.clone()); + } + + // Skip wire send when an in-process subscriber consumed the + // reply: the caller (e.g. `complete_login_register`, + // `handle_logout_request`) ships its own full-body reply on + // the same socket. Sending both desyncs the SDK -- it reads + // the first frame, fails to decode the typed body, and + // leaves the second frame stuck in the socket buffer. + if !had_in_process_subscriber { + wire_replies.push((event, reply)); + } + } + + // Wire replies AFTER the commit loop: this region may await, and + // a driver dropped here loses only reply frames — every commit + // above is applied and its reply cached in the client_table, so + // the SDK recovers it via request replay. + for (event, reply) in wire_replies { + let generic_reply = reply.into_generic(); + let reply_buffers = freeze_client_reply(generic_reply); + emit_sim_event(SimEventKind::ClientReplyEmitted, &event); + + if let Err(e) = consensus + .message_bus() + .send_to_client(event.client_id, reply_buffers) + .await + { + error!( + client = event.client_id, + op = event.op, + request_id = event.request_id, + operation = ?event.operation, + %e, + "client reply forward failed, no retransmit path; client will time out", + ); + } + } + + // Commits freed prepare slots and reopened the catch-up gate; + // promote buffered requests so the pipeline stays busy and + // absorbed submits (queued while this batch was mid-commit) + // dispatch immediately. + self.drain_request_queue_into_prepares().await; + } + + /// Timer-driven backstop (shard pump tick) for commit work stranded by + /// a canceled `on_ack` driver. + /// + /// The commit loop and the promotion of queued requests run at the tail + /// of the quorum-advancing `on_ack` — inside whichever future delivered + /// that ack, and that future can be dropped at any of its awaits + /// (journal read, wire-reply send). The pipeline is then left with + /// committed-but-unapplied entries (`commit_min < commit_max`) and/or + /// still-queued requests, and on an idle server nothing re-drives them: + /// `ack_quorum_reached` opens the commit path only when `commit_max` + /// advances, which duplicate and repair acks never do. This tick entry + /// re-runs the same commit path; a still-parked sibling driver loses + /// the head revalidation and exits clean. + /// + /// Ordering note: register promotion requires the catch-up gate open + /// (`register_preflight` drops the entry otherwise), and the gate can + /// only be closed here while stranded commits exist — which the commit + /// loop applies, reopening the gate, before promotion runs. + #[allow(clippy::future_not_send)] + pub async fn resume_stranded_commits(&self) { + let Some(consensus) = self.consensus.as_ref() else { + return; + }; + if !(consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing()) { + 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() + }; + if !stranded_commits && !promotable_requests { + return; + } + self.commit_committable_prefix().await; + } + + /// Promote buffered requests into free prepare slots after a commit + /// batch drains. /// /// # Safety /// Re-preflight per iteration: `commit_journal` may have advanced the @@ -3576,4 +3638,130 @@ mod tests { "session created by the promoted register" ); } + + /// The commit loop and the promotion of queued requests run at the tail + /// of `on_ack`, inside whichever future delivered the quorum ack. Drop + /// that future mid-commit and — on an idle server — nothing re-drives + /// the work: duplicate/repair acks do not re-open the commit path + /// (quorum already recorded, `commit_max` does not advance), so the + /// committed-but-unapplied op pins the catch-up gate closed and an + /// absorbed register parks in the request queue indefinitely. + /// + /// `resume_stranded_commits` (wired into the shard pump tick) is the + /// backstop: it re-enters the commit path, applies the stranded prefix, + /// and promotes the queued register, whose awaiter then resolves. + #[compio::test] + async fn tick_backstop_must_resume_stranded_commits_and_promotions() { + use std::future::Future; + + /// The client whose in-flight commit is stranded by the dropped driver. + const CLIENT_B: u128 = 2; + /// The client whose register parks in the request queue. + const CLIENT_C: u128 = 3; + 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_CONSENSUS_NAMESPACE, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + TestMux::default(), + None, + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table.borrow_mut().commit_register( + CLIENT_B, + ACTING_USER, + register_reply(CLIENT_B, SESSION), + |_| false, + ); + + // B's op journaled + self-acked; park its commit driver mid-window + // at the journal read. + let prepare = md + .prepare_request(create_stream_request(CLIENT_B, 1, "s1")) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + let mut loopback = Vec::new(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("one self-ack per prepare") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + let waker = std::task::Waker::noop(); + let mut cx = std::task::Context::from_waker(waker); + let mut driver = Box::pin(md.on_ack(ack)); + assert!(driver.as_mut().poll(&mut cx).is_pending()); + assert_eq!(consensus.commit_max(), 1); + assert_eq!(consensus.commit_min(), 0); + + // 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); + + // The committing driver dies at its await — the hyper-disconnect + // analogue. Commit and promotion are now stranded: op 1 is quorum'd + // (commit_max = 1) but unapplied (commit_min = 0), and no further + // ack will arrive to re-drive either. + drop(driver); + assert_eq!(consensus.commit_max(), 1); + assert_eq!(consensus.commit_min(), 0); + assert_eq!(consensus.pipeline().borrow().request_queue_len(), 1); + assert!( + register.as_mut().poll(&mut cx).is_pending(), + "queued register must still be parked with no driver alive" + ); + + // The pump tick backstop re-drives: commits op 1 (reopening the + // catch-up gate) and promotes the queued register into a prepare + // (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" + ); + + // Commit the promoted register (production: pump loopback drain) + // and the parked caller resolves with its session. + loopback.clear(); + consensus.drain_loopback_into(&mut loopback); + let ack = loopback + .pop() + .expect("promoted register must self-ack") + .try_into_typed::() + .expect("loopback holds self PrepareOks"); + md.on_ack(ack).await; + + let mut outcome = None; + for _ in 0..1_000 { + if let std::task::Poll::Ready(result) = register.as_mut().poll(&mut cx) { + outcome = Some(result); + break; + } + compio::time::sleep(std::time::Duration::from_millis(1)).await; + } + assert_eq!(outcome.expect("promoted register must resolve"), Ok(2)); + assert_eq!(md.client_table.borrow().get_session(CLIENT_C), Some(2)); + assert!(is_caught_up_primary(consensus)); + } } diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index f29ebbb6a1..6ed7195a1f 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -696,7 +696,6 @@ self_signed = true cert_file = "core/certs/iggy_cert.pem" key_file = "core/certs/iggy_key.pem" -# Message bus configuration. # Metadata consensus plane tunables (shard 0's VSR replica: users, # streams, topics, sessions). Size these together: a deeper prepare queue # admits more concurrent in-flight metadata ops (e.g. login storms), and @@ -716,6 +715,7 @@ prepare_queue_depth = 32 # rewrites per checkpoint. journal_slots = 1024 +# Message bus configuration. # Tunables for the inter-shard / inter-replica internal bus that ships # consensus traffic between replicas and SDK-client traffic between # shards. These knobs are consensus-liveness-critical (max_batch gates diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 5e8e5bc871..89160a63a1 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -2898,34 +2898,31 @@ mod tests { message } - /// Desired-contract test for the production failure chain "CLI stream + /// Regression test for the production failure chain "CLI stream /// create succeeded, logout failed: Disconnected". /// - /// Why the logout of a CLI invocation fails during ITS OWN successful - /// `stream create`: the catch-up gate is GLOBAL. The suite runs many - /// CLI invocations against one shared single-node server; each one is - /// three replicated ops (Register, work, Logout). When THIS client's - /// logout frame arrives, some SIBLING client's op is regularly sitting - /// between quorum-ack (`commit_max` advanced inside `on_ack`) and - /// apply (`commit_min` still behind, driver parked at the journal - /// read). `submit_logout_in_process` then rejects `NotCaughtUp`, and - /// `handle_logout_request` swallows the error: no reply frame, session - /// left bound. A one-shot CLI sees only a dead connection — "Problem - /// with server logout / Disconnected" — and exits non-zero although - /// its create committed; the harness retry then trips "already - /// exists". + /// Why the logout of a CLI invocation used to fail during ITS OWN + /// successful `stream create`: the catch-up gate was GLOBAL. The suite + /// runs many CLI invocations against one shared single-node server; + /// each one is three replicated ops (Register, work, Logout). When + /// THIS client's logout frame arrived, some SIBLING client's op was + /// regularly sitting between quorum-ack (`commit_max` advanced inside + /// `on_ack`) and apply (`commit_min` still behind, driver parked at + /// the journal read). `submit_logout_in_process` then rejected + /// `NotCaughtUp`, and `handle_logout_request` swallowed the error: no + /// reply frame, session left bound. A one-shot CLI saw only a dead + /// connection — "Problem with server logout / Disconnected" — and + /// exited non-zero although its create committed; the harness retry + /// then tripped "already exists". /// /// This test rebuilds that interleaving deterministically (client B = /// the sibling parked mid-commit; client A = the CLI logging out) and - /// asserts the DESIRED contract instead of today's behavior: + /// pins the contract that fixed it (non-register ops carry no + /// catch-up gate, see `submit_logout_in_process`): /// /// a client-initiated logout must always produce a reply frame and - /// unbind the transport session, even when the submit gate is - /// closed — teardown is best-effort, the VSR slot may lapse to the - /// eviction sweep. - /// - /// It FAILS on current code (silence + bound session) by design: it is - /// the reproduction to iterate fixes against. + /// unbind the transport session, even while a sibling's commit is + /// in flight — the logout simply pipelines behind it. #[compio::test] async fn logout_rejected_by_closed_gate_must_still_reply_to_client() { const CLIENT_A: u128 = 1; diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 2baaefb5de..60803f5b57 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -2683,6 +2683,14 @@ where // `IggyMetadata::repair_primary_self_acks`. metadata.repair_primary_self_acks().await; + // Backstop for commit work stranded by a canceled `on_ack` driver + // (a future dropped at its journal-read or wire-reply await): no + // further ack re-drives an already-advanced `commit_max`, so on an + // idle primary committed-but-unapplied ops and queued requests + // would otherwise wait for unrelated traffic. Quiet no-op when + // nothing is stranded. + metadata.resume_stranded_commits().await; + // Stall retry, mirroring `tick_partitions`: a lost repair frame must // not wedge the session forever. let stalled = {