This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch integration_tests_basics in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b8eb2ca5ba054a2ee4769eb50e8e0cb4926e38a3 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed May 27 13:53:19 2026 +0200 debug prints and fix register cmd --- core/metadata/src/impls/metadata.rs | 25 +++++ core/server-ng/src/bootstrap.rs | 177 ++++++++++++++++++++++++++++++------ core/shard/src/builder.rs | 8 +- core/shard/src/lib.rs | 64 +++++++++++++ core/shard/src/router.rs | 12 +++ 5 files changed, 258 insertions(+), 28 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index a820cf1bf..32a33380f 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -622,6 +622,15 @@ where async fn on_ack(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareOkHeader>) { let consensus = self.consensus.as_ref().unwrap(); let header = message.header(); + eprintln!( + "DBG on_ack op={} from_replica={} self={} view={} cmin={} cmax={}", + header.op, + header.replica, + consensus.replica(), + consensus.view(), + consensus.commit_min(), + consensus.commit_max(), + ); if let Err(reason) = ack_preflight(consensus) { warn!( @@ -875,6 +884,15 @@ where return Ok(session); } + eprintln!( + "DBG submit_register entry client={client_id} primary={} normal={} syncing={} cmin={} cmax={} view={}", + consensus.is_primary(), + consensus.is_normal(), + consensus.is_syncing(), + consensus.commit_min(), + consensus.commit_max(), + consensus.view(), + ); // 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) { @@ -937,6 +955,7 @@ where is_caught_up_primary(consensus), "submit_register_in_process: gate flipped between check and dispatch" ); + eprintln!("DBG submit_register dispatched prepare client={client_id}, awaiting commit"); self.on_replicate(prepare).await; debug_assert!( consensus.view() == view_snapshot && consensus.commit_min() == commit_min_snapshot, @@ -944,6 +963,12 @@ where ); let mut loopback = Vec::new(); consensus.drain_loopback_into(&mut loopback); + eprintln!( + "DBG submit_register after on_replicate client={client_id} loopback={} cmin={} cmax={}", + loopback.len(), + consensus.commit_min(), + consensus.commit_max(), + ); for message in loopback { match message.header().command { Command2::PrepareOk => match message.try_into_typed::<PrepareOkHeader>() { diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index e8c68bd9b..0df29dcb0 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -60,8 +60,8 @@ use iggy_binary_protocol::responses::system::get_stats::StatsResponse; use iggy_binary_protocol::responses::topics::get_topic::{GetTopicResponse, PartitionResponse}; use iggy_binary_protocol::responses::topics::get_topics::GetTopicsResponse; use iggy_binary_protocol::{ - Command2, GenericHeader, Operation, ReplyHeader, RequestHeader, WireDecode, WireEncode, - WireIdentifier, WireName, WirePartitioning, + Command2, EvictionReason, GenericHeader, Operation, ReplyHeader, RequestHeader, WireDecode, + WireEncode, WireIdentifier, WireName, WirePartitioning, }; use iggy_common::{ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, IggyError, IggyTimestamp, PartitionStats, @@ -84,7 +84,7 @@ use message_bus::{ }; use metadata::IggyMetadata; use metadata::MuxStateMachine; -use metadata::impls::metadata::{IggySnapshot, StreamsFrontend}; +use metadata::impls::metadata::{IggySnapshot, RegisterSubmitError, StreamsFrontend}; use metadata::impls::recovery::recover; use metadata::stm::consumer_group::ConsumerGroups; use metadata::stm::mux::WithFactory; @@ -1116,12 +1116,14 @@ async fn build_shard_for_thread( let shard_handle = Rc::new(RefCell::new(None)); let on_replica_message = make_deferred_replica_message_handler(&shard_handle); let on_client_request = make_deferred_client_request_handler(&bus, &shard_handle); + let on_metadata_submit = make_metadata_submit_handler(&shard_handle); let shard_name = format!("server-ng-shard-{shard_id}"); let built = IggyShardBuilder::new( ShardIdentity::new(shard_id, shard_name), Rc::clone(&bus), on_replica_message, on_client_request, + on_metadata_submit, metadata, partitions, senders, @@ -2186,6 +2188,53 @@ fn make_deferred_client_request_handler( }) } +/// Handler shard 0 runs for an inbound [`shard::MetadataSubmit`]: a peer +/// shard has verified credentials and owns the session locally, and asks +/// shard 0 (the metadata consensus owner) to run only the consensus +/// proposal. Spawns a task so the awaiting peer is woken once the op +/// commits; replies `None` on transient submit failure so the peer never +/// blocks forever. +fn make_metadata_submit_handler(shard_handle: &ServerNgShardHandle) -> shard::MetadataSubmitHandler { + let shard_handle = Rc::clone(shard_handle); + Rc::new(move |submit| { + let shard_handle = Rc::clone(&shard_handle); + compio::runtime::spawn(async move { + let Some(shard) = upgrade_shard_handle(&shard_handle) else { + return; + }; + match submit { + shard::MetadataSubmit::Register { + vsr_client_id, + reply, + } => { + let session = shard + .plane + .metadata() + .submit_register_in_process(vsr_client_id) + .await + .ok(); + let _ = reply.try_send(session); + } + shard::MetadataSubmit::Logout { + vsr_client_id, + session, + request, + reply, + } => { + let commit = shard + .plane + .metadata() + .submit_logout_in_process(vsr_client_id, session, request) + .await + .ok(); + let _ = reply.try_send(commit); + } + } + }) + .detach(); + }) +} + fn enqueue_client_request( shard: Rc<ServerNgShard>, sessions: Rc<RefCell<SessionManager>>, @@ -2441,6 +2490,66 @@ async fn handle_non_replicated_request( } } +/// Run the consensus `Register` proposal on the metadata owner (shard 0) +/// and return the committed session. +/// +/// Credential verification and session binding stay on the calling (home) +/// shard -- only this consensus step must execute where the metadata +/// consensus group lives. On shard 0 it calls in-process directly; on a +/// peer it forwards a [`shard::MetadataSubmit`] to shard 0 and awaits the +/// committed op. A dropped reply (shard-0 inbox full / shutdown) maps to a +/// transient `Canceled`, which the caller wraps so the SDK replays. +#[allow(clippy::future_not_send)] +async fn submit_register_on_owner( + shard: &Rc<ServerNgShard>, + vsr_client_id: u128, +) -> Result<u64, RegisterSubmitError> { + if shard.id == 0 { + return shard + .plane + .metadata() + .submit_register_in_process(vsr_client_id) + .await; + } + let (reply, rx) = shard::channel::<Option<u64>>(1); + shard.forward_metadata_submit(shard::MetadataSubmit::Register { + vsr_client_id, + reply, + }); + match rx.recv().await { + Ok(Some(session)) => Ok(session), + _ => Err(RegisterSubmitError::Canceled), + } +} + +/// Logout counterpart of [`submit_register_on_owner`]. +#[allow(clippy::future_not_send)] +async fn submit_logout_on_owner( + shard: &Rc<ServerNgShard>, + vsr_client_id: u128, + session: u64, + request: u64, +) -> Result<u64, RegisterSubmitError> { + if shard.id == 0 { + return shard + .plane + .metadata() + .submit_logout_in_process(vsr_client_id, session, request) + .await; + } + let (reply, rx) = shard::channel::<Option<u64>>(1); + shard.forward_metadata_submit(shard::MetadataSubmit::Logout { + vsr_client_id, + session, + request, + reply, + }); + match rx.recv().await { + Ok(Some(commit)) => Ok(commit), + _ => Err(RegisterSubmitError::Canceled), + } +} + #[allow(clippy::future_not_send)] async fn handle_logout_request( shard: &Rc<ServerNgShard>, @@ -2457,12 +2566,7 @@ async fn handle_logout_request( }; let request_id = request.header().request; - let commit = match shard - .plane - .metadata() - .submit_logout_in_process(vsr_client_id, session, request_id) - .await - { + let commit = match submit_logout_on_owner(shard, vsr_client_id, session, request_id).await { Ok(commit) => commit, Err(error) => { warn!(transport_client_id, error = %error, "logout/unregister failed"); @@ -2611,7 +2715,8 @@ async fn handle_login_register_request( .await { warn!(transport_client_id, error = %error, "login/register failed"); - send_login_failure_reply(shard, transport_client_id, request.header()).await; + surface_login_failure(shard, transport_client_id, request.header(), &error) + .await; } return; } @@ -2623,7 +2728,7 @@ async fn handle_login_register_request( } Err(error) => { warn!(transport_client_id, error = %error, "login/register failed"); - send_login_failure_reply(shard, transport_client_id, request.header()).await; + surface_login_failure(shard, transport_client_id, request.header(), &error).await; return; } } @@ -2647,7 +2752,8 @@ async fn handle_login_register_request( error = %error, "login/register with PAT failed" ); - send_login_failure_reply(shard, transport_client_id, request.header()).await; + surface_login_failure(shard, transport_client_id, request.header(), &error) + .await; } return; } @@ -2657,7 +2763,7 @@ async fn handle_login_register_request( error = %error, "login/register with PAT failed" ); - send_login_failure_reply(shard, transport_client_id, request.header()).await; + surface_login_failure(shard, transport_client_id, request.header(), &error).await; return; } } @@ -2670,13 +2776,37 @@ async fn handle_login_register_request( send_login_failure_reply(shard, transport_client_id, request.header()).await; } -/// Empty Reply on a failed Register. Without it the SDK -- which only -/// decodes `Command2::Reply` -- blocks until the socket read timeout fires -/// for what is really a typed failure. An empty body fails downstream -/// `LoginRegisterResponse` decoding with `InvalidCommand`, surfacing the -/// failure to the caller immediately. A future change can switch this to -/// an Eviction frame with a typed `EvictionReason` once the SDK eviction -/// decoder lands at every transport. +/// Decide whether a failed login/register gets a terminal reply or silence. +/// +/// `Transient` / `InvalidClientId` are `NotEvictable` (see +/// [`LoginRegisterError`]'s `TryFrom` for `EvictionReason`): the cluster +/// could not commit *right now* (e.g. a freshly booted primary still +/// catching up, or a cross-shard submit canceled). Staying silent lets the +/// SDK read-timeout replay -- a later attempt lands once the primary is +/// caught up. Replying empty here would instead surface as a hard +/// `InvalidFormat` decode failure and break the replay. +/// +/// Terminal auth errors (`InvalidCredentials` / `InvalidToken` / +/// `UserInactive` / `Session`) map to an `EvictionReason`, so we fast-fail +/// with an empty reply rather than make the client wait for a timeout. +/// (TODO: ship a typed `Eviction` frame once the SDK eviction decoder lands +/// on every transport.) +#[allow(clippy::future_not_send)] +async fn surface_login_failure( + shard: &Rc<ServerNgShard>, + transport_client_id: u128, + request_header: &RequestHeader, + error: &LoginRegisterError, +) { + if EvictionReason::try_from(error).is_ok() { + send_login_failure_reply(shard, transport_client_id, request_header).await; + } +} + +/// Empty Reply on a terminal failed Register. The SDK only decodes +/// `Command2::Reply`; an empty body fails `LoginRegisterResponse` decoding +/// fast instead of hanging until the socket read timeout. Only call for +/// terminal errors -- see [`surface_login_failure`]. #[allow(clippy::future_not_send)] async fn send_login_failure_reply( shard: &Rc<ServerNgShard>, @@ -3198,12 +3328,7 @@ async fn complete_login_register( .map_err(LoginRegisterError::Session)?; } - let session = match shard - .plane - .metadata() - .submit_register_in_process(vsr_client_id) - .await - { + let session = match submit_register_on_owner(shard, vsr_client_id).await { Ok(session) => session, Err(error) => { let _ = sessions diff --git a/core/shard/src/builder.rs b/core/shard/src/builder.rs index 00cb92a9a..52203db46 100644 --- a/core/shard/src/builder.rs +++ b/core/shard/src/builder.rs @@ -30,8 +30,8 @@ use crate::coordinator::{ShardZeroCoordinator, classify_try_send_err}; use crate::metrics::{ShardMetrics, frame_drop_variant}; use crate::{ - CoordinatorConfig, IggyShard, LifecycleFrame, PartitionConsensusConfig, Receiver, - ShardCtorError, ShardFrame, ShardIdentity, TaggedSender, + CoordinatorConfig, IggyShard, LifecycleFrame, MetadataSubmitHandler, PartitionConsensusConfig, + Receiver, ShardCtorError, ShardFrame, ShardIdentity, TaggedSender, }; use consensus::VsrConsensus; use journal::JournalHandle; @@ -64,6 +64,7 @@ where bus: B, on_replica_message: MessageHandler, on_client_request: RequestHandler, + on_metadata_submit: MetadataSubmitHandler, metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>, partitions: IggyPartitions<B>, senders: Vec<TaggedSender>, @@ -90,6 +91,7 @@ where bus: B, on_replica_message: MessageHandler, on_client_request: RequestHandler, + on_metadata_submit: MetadataSubmitHandler, metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>, partitions: IggyPartitions<B>, senders: Vec<TaggedSender>, @@ -104,6 +106,7 @@ where bus, on_replica_message, on_client_request, + on_metadata_submit, metadata, partitions, senders, @@ -208,6 +211,7 @@ where self.bus, self.on_replica_message, self.on_client_request, + self.on_metadata_submit, self.metadata, self.partitions, self.senders, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 8b10833ec..72c90f21d 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -100,6 +100,34 @@ pub fn channel<T: Send + 'static>(capacity: usize) -> (Sender<T>, Receiver<T>) { crossfire::mpsc::bounded_blocking_async(capacity) } +/// Cross-shard metadata consensus submit. +/// +/// The metadata consensus group lives only on shard 0. When a client +/// connection homes on a peer shard, that shard verifies credentials and +/// owns the session locally, but the consensus proposal (`Register` / +/// `Logout`) must execute on shard 0. The peer hands just that step here +/// and awaits the committed op number over `reply` (`None` = transient +/// submit failure; all `RegisterSubmitError` variants are transient by +/// contract, so the caller retries rather than distinguishing them). +pub enum MetadataSubmit { + Register { + vsr_client_id: u128, + reply: Sender<Option<u64>>, + }, + Logout { + vsr_client_id: u128, + session: u64, + request: u64, + reply: Sender<Option<u64>>, + }, +} + +/// Handler shard 0 runs for an inbound [`MetadataSubmit`]. server-ng wires +/// it to `submit_register_in_process` / `submit_logout_in_process` and +/// sends the result back over the frame's `reply` sender. `None` on a peer +/// shard (no consensus): a peer must never receive this frame. +pub type MetadataSubmitHandler = Rc<dyn Fn(MetadataSubmit)>; + /// Create a bounded inter-shard channel whose sender is tagged with the /// owning shard. /// @@ -291,6 +319,11 @@ pub enum LifecycleFrame { client_id: u128, msg: Frozen<MESSAGE_ALIGN>, }, + /// A peer shard hands a metadata consensus submit (login/logout) to + /// shard 0, the metadata consensus owner. The committed op returns over + /// the `reply` sender carried in [`MetadataSubmit`]. Always addressed to + /// shard 0; processing it on a peer is a routing bug. + MetadataSubmit(MetadataSubmit), } /// Inter-shard channel envelope. @@ -364,6 +397,12 @@ where /// this shard. Invoked for each inbound `Request` frame. on_client_request: RequestHandler, + /// Handler for inbound [`MetadataSubmit`] frames. Only shard 0 receives + /// these (it owns the metadata consensus group); peers send them here + /// via [`Self::forward_metadata_submit`]. Defaults to a no-op for the + /// simulator stub ctor. + on_metadata_submit: MetadataSubmitHandler, + /// Channel senders to every shard, indexed by shard id. /// Includes a sender to self so that local routing goes through the /// same channel path as remote routing. @@ -434,6 +473,7 @@ where bus: B, on_replica_message: MessageHandler, on_client_request: RequestHandler, + on_metadata_submit: MetadataSubmitHandler, metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>, partitions: IggyPartitions<B>, senders: Vec<TaggedSender>, @@ -457,6 +497,7 @@ where bus, on_replica_message, on_client_request, + on_metadata_submit, senders, shard_count, inbox, @@ -467,6 +508,28 @@ where }) } + /// Hand a metadata consensus submit (login/logout) to shard 0. + /// + /// Sends a [`LifecycleFrame::MetadataSubmit`] into shard 0's inbox. The + /// caller owns the matching [`Receiver`] (paired with the `reply` sender + /// inside `submit`) and awaits the committed op there. On a full / + /// disconnected shard-0 inbox the frame is dropped; the dropped `reply` + /// sender then surfaces as a recv error the caller maps to a transient + /// failure. + pub fn forward_metadata_submit(&self, submit: MetadataSubmit) { + let frame = ShardFrame::lifecycle(LifecycleFrame::MetadataSubmit(submit)); + if let Err(error) = self.senders[0].try_send(frame) { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::CONSENSUS, + crate::coordinator::classify_try_send_err(&error), + ); + tracing::warn!( + shard = self.id, + "forward_metadata_submit: shard-0 inbox rejected frame: {error:?}" + ); + } + } + /// Return a clone of the shard-0 coordinator handle, if attached. /// Bootstrap uses this to wire the listener accept callbacks /// (replica + client) to coordinator-driven fd-delegation instead @@ -503,6 +566,7 @@ where bus, on_replica_message: std::rc::Rc::new(|_, _| {}), on_client_request: std::rc::Rc::new(|_, _| {}), + on_metadata_submit: std::rc::Rc::new(|_| {}), plane, coordinator: None, senders: Vec::new(), diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 5060f713c..f964a7c50 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -399,6 +399,18 @@ where ); } } + LifecycleFrame::MetadataSubmit(submit) => { + // Only shard 0 owns the metadata consensus group, and + // `forward_metadata_submit` always addresses shard 0, so a + // non-zero shard here is a routing bug. The handler (wired + // by server-ng) replies `None` on the carried sender if it + // cannot submit, so the awaiting peer never blocks forever. + debug_assert_eq!( + self.id, 0, + "MetadataSubmit must only be processed on shard 0" + ); + (self.on_metadata_submit)(submit); + } } } }
