| // 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. |
| |
| //! Per-shard request dispatch: queue plumbing and the request funnel. |
| //! |
| //! The tree: [`host`] (the shard host: per-client queues, session manager, |
| //! connection-lost hook), [`session_ops`] (login/register/logout and their |
| //! replica forwards), [`partition`] (the partition data plane, both mesh ends), |
| //! [`reads`] (the non-replicated read router), [`submit`] (the shard-0 |
| //! metadata-submit RPC), `authz` (the wire-path authorization gates), |
| //! `failure` (the wire failure channels and the one send exit for |
| //! host-built frames). |
| //! |
| //! Deliberate asymmetry (the two authz gates): replicated metadata ops are |
| //! authorized in-apply by the STM, in committed order on every replica; |
| //! partition and non-replicated ops never enter the metadata log, so `authz` |
| //! gates them pre-dispatch against this shard's applied permissioner. The |
| //! HTTP spine keeps its own equivalent gates (see `crate::http`) because its |
| //! error contract (404-before-403) is pinned client-visible behavior. |
| |
| mod authz; |
| mod failure; |
| pub mod host; |
| pub mod login_error; |
| pub mod partition; |
| pub mod reads; |
| pub mod session_ops; |
| pub mod submit; |
| #[cfg(test)] |
| pub mod test_support; |
| |
| use crate::consumer_group::maybe_rewrite_consumer_group_request; |
| use crate::dispatch::failure::{ |
| FrameChannel, send_deny_reply, send_eviction, send_host_frame, send_pre_consensus_deny, |
| send_unbound_deny_reply, |
| }; |
| use crate::dispatch::partition::{dispatch_partition_request, handle_delete_segments_request}; |
| use crate::dispatch::reads::handle_non_replicated_request; |
| use crate::dispatch::session_ops::{handle_login_register_request, handle_logout_request}; |
| use crate::dispatch::submit::{committed_reply_commit, submit_client_request_on_owner}; |
| use crate::reply_frame::build_raw_pat_reply; |
| use crate::rewrite::{RewriteDeny, RewriteStage, tcp_chain}; |
| use crate::session_manager::{ConnectionContext, SessionManager}; |
| use crate::shell::{ShellBus, ShellShard, ShellShardHandle}; |
| use crate::wire::verify_request_checksum; |
| use ahash::{AHashMap, AHashSet}; |
| use configs::server::ServerConfig; |
| use consensus::MetadataHandle; |
| use futures::future::{Either, select}; |
| use iggy_binary_protocol::PrepareHeader; |
| use iggy_binary_protocol::codes::{ |
| LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, |
| }; |
| use iggy_binary_protocol::requests::system::SessionIdentity; |
| use iggy_binary_protocol::requests::users::login_register::BindSecret; |
| use iggy_binary_protocol::{ |
| EvictionReason, GenericHeader, Operation, RequestHeader, RoutedRequestHeader, |
| }; |
| use iggy_common::{IggyError, UserStatus}; |
| use journal::superblock::SuperblockStore; |
| use journal::{Journal, JournalHandle}; |
| use metadata::impls::metadata::StreamsFrontend; |
| use secrecy::ExposeSecret; |
| use server_common::Message; |
| use std::cell::RefCell; |
| use std::collections::VecDeque; |
| use std::pin::pin; |
| use std::rc::Rc; |
| use std::sync::Arc; |
| use tracing::{debug, warn}; |
| |
| type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<Message<GenericHeader>>>>>; |
| |
| /// Requests one client may have queued behind a request this shard has not |
| /// answered yet. |
| /// |
| /// The drain loop below serves one frame per client at a time, and a frame can |
| /// legitimately hold it for a while: a metadata write awaits consensus, and a |
| /// read can be HELD for the read-your-writes budget (see |
| /// `crate::dispatch::reads`). Without a cap, a client that keeps pipelining |
| /// through such a stall grows its queue - and this node's memory - unbounded. |
| /// |
| /// Overflow is ANSWERED, not dropped: `TransientNotAccepted` is the honest |
| /// code, since the frame provably never entered any pipeline, so the SDK may |
| /// re-issue it anywhere, including here once the queue drains. Sized far above |
| /// any SDK's in-flight window, so it only ever fires under a genuine stall. |
| const MAX_QUEUED_CLIENT_REQUESTS: usize = 1024; |
| type ActiveClientRequests = Rc<RefCell<AHashSet<u128>>>; |
| |
| // BindSession authenticates with the registered secret, never with the |
| // public client id and epoch alone. It retains the original retry identity; |
| // unbound replicated frames still fail with Eviction(NoSession). |
| |
| #[allow(clippy::too_many_arguments)] |
| fn enqueue_client_request<B, MJ, S, SB>( |
| bus: &B, |
| shard_handle: &ShellShardHandle<B, MJ, S, SB>, |
| sessions: &Rc<RefCell<SessionManager>>, |
| server_config: &Arc<ServerConfig>, |
| max_tokens_per_user: u32, |
| queues: &ClientRequestQueues, |
| active: &ActiveClientRequests, |
| client_id: u128, |
| message: Message<GenericHeader>, |
| ) where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| { |
| let mut queues = queues.borrow_mut(); |
| let queue = queues.entry(client_id).or_default(); |
| if queue.len() >= MAX_QUEUED_CLIENT_REQUESTS { |
| // Borrow released before the deny, which spawns onto this same |
| // task and would otherwise re-enter the table. |
| drop(queues); |
| deny_overflowing_client_request(shard_handle, client_id, message); |
| return; |
| } |
| queue.push_back(message); |
| } |
| if !active.borrow_mut().insert(client_id) { |
| return; |
| } |
| |
| let shard_handle = Rc::clone(shard_handle); |
| let sessions = Rc::clone(sessions); |
| let server_config = Arc::clone(server_config); |
| let queues = Rc::clone(queues); |
| let active = Rc::clone(active); |
| bus.spawn(async move { |
| let _slot = ActiveDrainSlot { active, client_id }; |
| // The handle is set once the shard is built. A frame that beats it |
| // stays queued and the client's next frame drains both, which needs |
| // the slot released on this path too - the guard above does it. |
| let Some(shard) = upgrade_shard_handle(&shard_handle) else { |
| return; |
| }; |
| drain_client_requests( |
| shard, |
| sessions, |
| server_config, |
| max_tokens_per_user, |
| queues, |
| client_id, |
| ) |
| .await; |
| }); |
| } |
| |
| /// Answer a request that arrived with this client's queue already at |
| /// [`MAX_QUEUED_CLIENT_REQUESTS`] with the retryable transient denial. |
| /// |
| /// Spawned rather than awaited: the enqueue path is sync (it runs straight off |
| /// frame arrival) and the reply goes out on the bus. Two shapes are dropped |
| /// rather than answered: a frame whose header will not even cast, exactly as |
| /// the drain loop drops it, and a frame that arrives before the shard is |
| /// built, which leaves nothing to render a reply from. |
| fn deny_overflowing_client_request<B, MJ, S, SB>( |
| shard_handle: &ShellShardHandle<B, MJ, S, SB>, |
| transport_client_id: u128, |
| message: Message<GenericHeader>, |
| ) where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| // Unreachable in practice: a listener binds only after the build backfills |
| // the weak self-reference, which `crate::boot` pins with a `debug_assert`. |
| // Dropped rather than admitted past the cap -- the cap exists to refuse |
| // this frame, and the deny itself needs the shard for its metric and reply. |
| let Some(shard) = upgrade_shard_handle(shard_handle) else { |
| warn!( |
| transport_client_id, |
| "dropping over-queue client request received before the shard was built" |
| ); |
| return; |
| }; |
| shard.metrics().record_client_request_denied_queue_full(); |
| let Ok(request) = message.try_into_typed::<RequestHeader>() else { |
| warn!( |
| transport_client_id, |
| "dropping over-queue client request with invalid header" |
| ); |
| return; |
| }; |
| let request = request.into_routed(); |
| debug!( |
| transport_client_id, |
| operation = ?request.header().operation, |
| queued = MAX_QUEUED_CLIENT_REQUESTS, |
| "denying client request retryable: this connection's request queue is full" |
| ); |
| let bus = shard.bus.clone(); |
| bus.spawn(async move { |
| send_deny_reply( |
| &shard, |
| transport_client_id, |
| request.header(), |
| IggyError::TransientNotAccepted.as_code(), |
| ) |
| .await; |
| }); |
| } |
| |
| /// Holds a client's one drain slot for as long as its drain task lives. |
| /// |
| /// Releasing from the task's own `Drop` covers every exit, the panic compio |
| /// catches included: a slot left taken with no live drain queues every later |
| /// frame for that client forever. It also keeps the release out of the |
| /// connection-lost hook, which fires from the transport task and could |
| /// otherwise clear the slot under a drain suspended at an `.await`. |
| struct ActiveDrainSlot { |
| active: ActiveClientRequests, |
| client_id: u128, |
| } |
| |
| impl Drop for ActiveDrainSlot { |
| fn drop(&mut self) { |
| self.active.borrow_mut().remove(&self.client_id); |
| } |
| } |
| |
| #[allow(clippy::future_not_send)] |
| async fn drain_client_requests<B, MJ, S, SB>( |
| shard: Rc<ShellShard<B, MJ, S, SB>>, |
| sessions: Rc<RefCell<SessionManager>>, |
| server_config: Arc<ServerConfig>, |
| max_tokens_per_user: u32, |
| queues: ClientRequestQueues, |
| client_id: u128, |
| ) where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| loop { |
| let Some(message) = pop_next_client_request(&queues, client_id) else { |
| return; |
| }; |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| max_tokens_per_user, |
| client_id, |
| message, |
| ) |
| .await; |
| } |
| } |
| |
| fn pop_next_client_request( |
| queues: &ClientRequestQueues, |
| client_id: u128, |
| ) -> Option<Message<GenericHeader>> { |
| // Freed as soon as it drains empty. Keeping it would cost the connection |
| // one `VecDeque` (at its peak depth, since `pop_front` never gives |
| // capacity back), and the connection-lost hook cannot be the sole owner: |
| // the dispatch task can still deliver a buffered frame after the hook |
| // ran, and `client_id` is never reused, so that entry would live until |
| // process exit. |
| let mut queues = queues.borrow_mut(); |
| let queue = queues.get_mut(&client_id)?; |
| let message = queue.pop_front(); |
| if queue.is_empty() { |
| queues.remove(&client_id); |
| } |
| message |
| } |
| |
| /// Where the funnel routes a client request. Derived by [`classify`], which |
| /// IS the routing: [`handle_client_request`] matches on its result. The |
| /// variant ORDER mirrors the order of the checks inside [`classify`], and |
| /// that order is semantics (documented there). |
| #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| pub(in crate::dispatch) enum RequestClass { |
| /// Legacy pre-register login code: rejected with a typed |
| /// `MalformedLogin` eviction before the session gate. |
| LegacyLogin, |
| /// Non-replicated code other than PING on an unbound transport: |
| /// denied `Unauthenticated` (plain deny reply, never an eviction). |
| UnauthenticatedRead, |
| /// Non-replicated read for the reads router. |
| NonReplicatedRead, |
| /// The register handshake (`session == 0 && request == 0`). |
| LoginRegister, |
| Logout, |
| /// Replicated operation on an unbound transport: `Eviction(NoSession)`. |
| UnboundReplicated, |
| /// Neither a partition nor a metadata consensus op: resolved to a |
| /// replicated `TruncatePartition` by the owning shard. |
| DeleteSegments, |
| /// Partition-plane operation. |
| Partition, |
| /// Everything else: replicated metadata consensus. |
| ReplicatedMetadata, |
| } |
| |
| /// Route a client request: [`handle_client_request`] matches on the result, |
| /// so this function IS the routing and the order of the checks below is the |
| /// semantics. Pure: `bound` stands for |
| /// `sessions.get_session(transport_client_id).is_some()`, the only session |
| /// fact the checks consult. |
| /// |
| /// The pins, in check order: |
| /// - the legacy-login rejection precedes the session gate: a legacy code |
| /// must get the typed `MalformedLogin` eviction, not the generic |
| /// unauthenticated deny; |
| /// - the pre-auth allowlist is PING only; `GET_CLUSTER_METADATA` is |
| /// deliberately NOT pre-auth (see the funnel's auth-bypass guard); |
| /// - poll-messages and consumer-offset reads are non-replicated CODES, not |
| /// partition operations: they classify |
| /// [`RequestClass::NonReplicatedRead`] and route inside the reads router; |
| /// - a `Register` with `session != 0` or `request != 0` falls through to |
| /// the default. Those rows are classify-only: `RequestHeader::validate` |
| /// rejects such a header before the funnel sees it, so no mid-session |
| /// register ever reaches consensus; |
| /// - `DeleteSegments` is neither a partition nor a metadata op, so its |
| /// check sits before `is_partition`; |
| /// - the checksum and heartbeat pre-gates run BEFORE classification in the |
| /// funnel. |
| pub(in crate::dispatch) fn classify(header: &RoutedRequestHeader, bound: bool) -> RequestClass { |
| if header.operation == Operation::NonReplicated { |
| let nr_code = non_replicated_code(header); |
| if matches!( |
| nr_code, |
| LOGIN_USER_CODE | LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE |
| ) { |
| return RequestClass::LegacyLogin; |
| } |
| if !matches!( |
| nr_code, |
| PING_CODE | iggy_binary_protocol::codes::BIND_SESSION_CODE |
| ) && !bound |
| { |
| return RequestClass::UnauthenticatedRead; |
| } |
| return RequestClass::NonReplicatedRead; |
| } |
| if header.operation == Operation::Register && header.session == 0 && header.request == 0 { |
| return RequestClass::LoginRegister; |
| } |
| if header.operation == Operation::Logout { |
| return RequestClass::Logout; |
| } |
| if !bound { |
| return RequestClass::UnboundReplicated; |
| } |
| if header.operation == Operation::DeleteSegments { |
| return RequestClass::DeleteSegments; |
| } |
| if header.operation.is_partition() { |
| return RequestClass::Partition; |
| } |
| RequestClass::ReplicatedMetadata |
| } |
| |
| /// The command code a `NonReplicated` header carries in its first four |
| /// reserved bytes. |
| fn non_replicated_code(header: &RoutedRequestHeader) -> u32 { |
| u32::from_le_bytes(header.reserved[..4].try_into().unwrap()) |
| } |
| |
| #[allow(clippy::future_not_send, clippy::too_many_lines)] |
| async fn handle_client_request<B, MJ, S, SB>( |
| shard: &Rc<ShellShard<B, MJ, S, SB>>, |
| sessions: &Rc<RefCell<SessionManager>>, |
| server_config: &Arc<ServerConfig>, |
| max_tokens_per_user: u32, |
| transport_client_id: u128, |
| message: Message<iggy_binary_protocol::GenericHeader>, |
| ) where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| let request = match message.try_into_typed::<RequestHeader>() { |
| Ok(request) => request, |
| Err(error) => { |
| warn!( |
| transport_client_id, |
| error = %error, |
| "dropping client request with invalid header" |
| ); |
| return; |
| } |
| }; |
| // Promote to the server-internal routed shape at the boundary: the |
| // client wire carries no group (it is derived -- plane from `operation`, |
| // partition target from the payload), so it starts unset here and the |
| // resolution sites below stamp it before anything routes on it. |
| let request = request.into_routed(); |
| |
| // The last point that still sees the body the CLIENT sent; every rewrite below |
| // substitutes server-chosen bytes and carries the stamp through unchanged. |
| if let Err(error) = verify_request_checksum(&request) { |
| warn!( |
| transport_client_id, |
| operation = ?request.header().operation, |
| request = request.header().request, |
| "dropping client request whose body does not match its own checksum" |
| ); |
| send_deny_reply( |
| shard, |
| transport_client_id, |
| request.header(), |
| error.as_code(), |
| ) |
| .await; |
| return; |
| } |
| |
| // ONE `connections` walk for the whole prologue. Any request is liveness |
| // proof, not just PING: an idle-but-active client (e.g. an admin issuing |
| // reads between long sleeps) must not be evicted by the heartbeat |
| // verifier. A genuinely dead connection sends nothing, so the intended |
| // stale-client eviction still fires. |
| // Bound, not matched: a `match` on `borrow_mut()` would hold the guard |
| // across the arm that borrows again. |
| let mut touched = sessions.borrow_mut().touch_connection(transport_client_id); |
| if touched.is_none() { |
| // A transport's first frame: the peer address and transport kind live |
| // on the bus, so only this path pays that lookup. |
| ensure_transport_connection(shard, sessions, transport_client_id); |
| touched = sessions.borrow_mut().touch_connection(transport_client_id); |
| } |
| let ConnectionContext { |
| bound, |
| user_id, |
| address: client_address, |
| metadata_watermark, |
| } = touched.unwrap_or_default(); |
| |
| // Borrowed, not copied: the 256-byte header is only worth a by-value |
| // snapshot where an arm rewrites it (`ReplicatedMetadata`) and still has |
| // to echo the client's original fields on a deny. |
| let header = request.header(); |
| match classify(header, bound.is_some()) { |
| RequestClass::LegacyLogin => { |
| // Legacy (pre-register) login codes. The server authenticates only via |
| // the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT, |
| // Operation::Register); the vsr SDK funnels both logins there and never |
| // emits these. Reject them uniformly with a typed MalformedLogin (the |
| // SDK maps it to InvalidFormat) before the session gate, so a legacy or |
| // foreign client fails fast instead of getting the generic |
| // Unauthenticated deny the pre-auth guard would send unbound, or the |
| // silent empty-ok Reply the bound non-replicated path would send. |
| let nr_code = non_replicated_code(header); |
| warn!( |
| transport_client_id, |
| code = nr_code, |
| "rejecting legacy login code; server requires the register handshake" |
| ); |
| send_eviction( |
| shard, |
| transport_client_id, |
| header.client, |
| EvictionReason::MalformedLogin, |
| "legacy_login_rejection", |
| ) |
| .await; |
| } |
| RequestClass::UnauthenticatedRead => { |
| let nr_code = non_replicated_code(header); |
| // No per-code exemption: every in-tree SDK reads the roster only |
| // after login, so an unauthenticated roster read is a real event |
| // and not something to hide at debug. |
| warn!( |
| transport_client_id, |
| code = nr_code, |
| "denying pre-auth non-replicated read with Unauthenticated" |
| ); |
| // A plain deny Reply, not an Eviction: there is no session to |
| // evict, and an Eviction is session-terminal by wire contract, |
| // so SDKs would tear down the very connection their login is |
| // about to use. The status channel carries the error the same |
| // way the request-checksum denial above does. |
| send_unbound_deny_reply( |
| shard, |
| transport_client_id, |
| request.header(), |
| IggyError::Unauthenticated.as_code(), |
| ) |
| .await; |
| } |
| RequestClass::NonReplicatedRead => { |
| // The auth-bypass guard is `classify`'s `UnauthenticatedRead` class: |
| // `PING`, the liveness probe, is the only pre-auth code, on every |
| // roster shape. `GET_CLUSTER_METADATA` describes the private replica |
| // network and is not something an unauthenticated caller gets to |
| // read; a client that dialed a backup no longer needs it to find the |
| // leader, because the backup authenticates the login locally and |
| // forwards only the consensus proposal |
| // (`submit_register_local_or_forward`). Every other non-replicated |
| // code MUST go through Register first, which binds the acting user |
| // the per-op authz gates resolve. |
| handle_non_replicated_request( |
| shard, |
| sessions, |
| server_config, |
| transport_client_id, |
| request, |
| ConnectionContext { |
| bound, |
| user_id, |
| address: client_address, |
| metadata_watermark, |
| }, |
| ) |
| .await; |
| } |
| RequestClass::LoginRegister => { |
| handle_login_register_request(shard, sessions, transport_client_id, request).await; |
| } |
| RequestClass::Logout => { |
| handle_logout_request(shard, sessions, transport_client_id, request).await; |
| } |
| RequestClass::UnboundReplicated => { |
| // Replicated request on an unbound transport. Without this short- |
| // circuit, the rewrite below overwrites `header.client` with |
| // `transport_client_id` and dispatches; the request_preflight then |
| // rejects with `NoSession`/`Fenced` and the failure disappears |
| // silently, wedging the SDK until the socket timeout. A typed |
| // `Eviction(NoSession)` is right here, unlike the plain deny of |
| // `UnauthenticatedRead`: a replicated request implies the client |
| // believes it has a session, and that session is gone, so it must |
| // register again. An empty status-0 Reply is not safe here, because |
| // SendMessages is the one replicated operation without a result |
| // section, and its decoder would read the empty body as a |
| // successful send. |
| warn!( |
| transport_client_id, |
| operation = ?header.operation, |
| "rejecting replicated request from unbound transport with Eviction(NoSession)" |
| ); |
| // The eviction context is best-effort off the metadata consensus |
| // (peer shards have none; zeroes are cosmetic -- the SDK only |
| // reads the reason), and the evicted id is the transport id: no |
| // VSR session exists to name. |
| send_eviction( |
| shard, |
| transport_client_id, |
| transport_client_id, |
| EvictionReason::NoSession, |
| "unbound_replicated_request", |
| ) |
| .await; |
| } |
| RequestClass::DeleteSegments => { |
| // DeleteSegments is neither a partition nor a metadata consensus op: the |
| // owning shard resolves the requested count to a concrete offset, then a |
| // `TruncatePartition` is replicated through metadata (Option A). Each |
| // replica's reconciler trims to the committed watermark. `classify` |
| // names it ahead of the partition and metadata classes. |
| handle_delete_segments_request(shard, transport_client_id, bound, &request).await; |
| } |
| RequestClass::Partition => { |
| // `bound` is Some here: `classify` sends unbound transports to |
| // `UnboundReplicated`. |
| let (vsr_client_id, bound_session) = bound.unwrap_or((0, 0)); |
| let consumer_session = { |
| let attachment = sessions |
| .borrow() |
| .attached_consumer_session(transport_client_id); |
| match attachment { |
| Ok(attachment) => attachment, |
| Err(error) => { |
| send_deny_reply( |
| shard, |
| transport_client_id, |
| request.header(), |
| error.as_code(), |
| ) |
| .await; |
| return; |
| } |
| } |
| }; |
| // The acting user comes from the prologue's lookup. A bound |
| // transport always has one, but the gate below fails closed on |
| // `None` rather than trust that. |
| dispatch_partition_request( |
| shard, |
| request, |
| vsr_client_id, |
| bound_session, |
| transport_client_id, |
| user_id, |
| consumer_session, |
| None, |
| ) |
| .await; |
| } |
| RequestClass::ReplicatedMetadata => { |
| // The one arm that needs the by-value copy: the rewrite below |
| // stamps the consensus client / session / group over the header, |
| // and a pre-consensus deny still has to echo what the client sent. |
| let header = *header; |
| let request = |
| request.transmute_header(|header, new_header: &mut RoutedRequestHeader| { |
| *new_header = header; |
| // Metadata-plane ops route by operation: stamp the sentinel group. |
| new_header.group = server_common::sharding::METADATA_GROUP; |
| // `bound` is always Some here (`classify` sends unbound transports to |
| // `UnboundReplicated`); this sets the consensus client id + session |
| // for the replicated op. |
| if let Some((bound_client_id, bound_session)) = bound { |
| new_header.client = bound_client_id; |
| new_header.session = bound_session; |
| } |
| }); |
| let (request, raw_pat_token) = match tcp_chain( |
| shard, |
| sessions, |
| transport_client_id, |
| max_tokens_per_user, |
| request, |
| ) { |
| Ok(rewritten) => rewritten, |
| Err(RewriteDeny { stage, error }) => { |
| send_pre_consensus_deny(shard, transport_client_id, &header, &error, stage) |
| .await; |
| return; |
| } |
| }; |
| // Enrich consumer-group Join/Leave with the client's VSR id (+ topic |
| // partition count for Join) before replication; see `crate::consumer_group`. |
| let request = match maybe_rewrite_consumer_group_request(shard, request).await { |
| Ok(rewritten) => rewritten, |
| Err(error) => { |
| // Preserve transient recovery rejection so the client can |
| // retry the join once partition state is available. |
| send_pre_consensus_deny( |
| shard, |
| transport_client_id, |
| &header, |
| &error, |
| RewriteStage::ConsumerGroup, |
| ) |
| .await; |
| return; |
| } |
| }; |
| let request_header = *request.header(); |
| // Replicated request: run consensus on the metadata owner (shard 0) and |
| // bring the committed reply back here. This shard owns the connection, |
| // so it writes the reply to the socket via the transport client id -- |
| // shard 0 can't route by the consensus client id (no home-shard bits). |
| match submit_client_request_on_owner(shard, request).await { |
| Some(reply) => { |
| // Recorded before the reply reaches the socket, so a read the client |
| // sends the instant it decodes this frame already sees the mark. |
| if let Some(commit) = committed_reply_commit(&reply) { |
| sessions |
| .borrow_mut() |
| .record_metadata_watermark(transport_client_id, commit); |
| } |
| // The raw PAT token never enters consensus (it is non-deterministic |
| // and secret), so the committed reply body is empty. Substitute the |
| // raw-token response here, on the minting client's home shard, using |
| // the confirmed commit position from the committed reply. |
| let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) { |
| Ok(reply) => reply, |
| Err(error) => { |
| warn!( |
| transport_client_id, |
| error = %error, |
| "failed to build raw PAT reply" |
| ); |
| // The op COMMITTED; only the reply could not be |
| // rendered. A typed deny is still the right frame: |
| // silence wedges the lockstep connection on a |
| // request that succeeded server-side. The raw |
| // token is unrecoverable either way -- it lives in |
| // that one reply and a retry dedups to the empty |
| // committed body -- so the caller must delete the |
| // token and mint a new one. |
| send_deny_reply(shard, transport_client_id, &header, error.as_code()) |
| .await; |
| return; |
| } |
| }; |
| send_host_frame( |
| &shard.bus, |
| transport_client_id, |
| reply.into_frozen(), |
| FrameChannel::Reply, |
| "committed_reply", |
| ) |
| .await; |
| } |
| None => { |
| // Transient submit failure (not primary / not caught up / dedup |
| // absorbed). Stay silent; the SDK read-timeout replays. |
| warn!( |
| transport_client_id, |
| operation = ?header.operation, |
| "replicated request not committed (transient); client will replay" |
| ); |
| } |
| } |
| } |
| } |
| } |
| |
| fn ensure_transport_connection<B, MJ, S, SB>( |
| shard: &Rc<ShellShard<B, MJ, S, SB>>, |
| sessions: &Rc<RefCell<SessionManager>>, |
| transport_client_id: u128, |
| ) where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| let Some(meta) = shard.bus.client_meta(transport_client_id) else { |
| return; |
| }; |
| sessions |
| .borrow_mut() |
| .ensure_connection(transport_client_id, meta.peer_addr, meta.transport); |
| } |
| |
| pub(in crate::dispatch) fn upgrade_shard_handle<B, MJ, S, SB>( |
| shard_handle: &ShellShardHandle<B, MJ, S, SB>, |
| ) -> Option<Rc<ShellShard<B, MJ, S, SB>>> |
| where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| shard_handle |
| .borrow() |
| .as_ref() |
| .and_then(std::rc::Weak::upgrade) |
| } |
| |
| #[allow(clippy::future_not_send)] |
| async fn complete_session_binding<B, MJ, S, SB>( |
| shard: &Rc<ShellShard<B, MJ, S, SB>>, |
| sessions: &Rc<RefCell<SessionManager>>, |
| transport_client_id: u128, |
| identity: SessionIdentity, |
| secret: BindSecret, |
| ) -> Result<(u32, u64), IggyError> |
| where |
| B: ShellBus, |
| MJ: JournalHandle + 'static, |
| MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, |
| S: 'static, |
| SB: SuperblockStore + 'static, |
| { |
| let watermark = identity.metadata_watermark.max(identity.session); |
| reads::await_metadata_read_frontier(shard, watermark).await?; |
| let (user_id, session, attachment) = if shard.id == 0 { |
| shard |
| .plane |
| .metadata() |
| .client_table |
| .borrow_mut() |
| .bind_session(identity.client_id, identity.session, secret.expose_secret())? |
| } else { |
| let (reply, receiver) = shard::channel(1); |
| shard.forward_metadata_submit(shard::MetadataSubmit::BindSession { |
| vsr_client_id: identity.client_id, |
| session: identity.session, |
| secret, |
| reply, |
| }); |
| let outcome = pin!(receiver.recv()); |
| let deadline = pin!( |
| shard |
| .bus |
| .sleep(shard.plane.metadata().applied_frontier().read_budget()) |
| ); |
| match select(outcome, deadline).await { |
| Either::Left((result, _)) => result.map_err(|_| IggyError::TransientNotAccepted)??, |
| Either::Right(_) => return Err(IggyError::TransientNotAccepted), |
| } |
| }; |
| let active_user = shard.plane.metadata().mux_stm.users().read(|users| { |
| users |
| .items |
| .get(user_id as usize) |
| .is_some_and(|user| user.status == UserStatus::Active) |
| }); |
| if !active_user { |
| return Err(IggyError::Unauthenticated); |
| } |
| sessions.borrow_mut().bind_authenticated_connection( |
| transport_client_id, |
| identity.client_id, |
| session, |
| user_id, |
| attachment, |
| watermark, |
| )?; |
| Ok((user_id, session)) |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| use crate::cluster_meta::ClusterRoster; |
| use crate::dispatch::host::ServerHost; |
| use crate::dispatch::test_support::{ |
| FIRST_BOOT, SpyBus, TestMux, TestShard, prepare_message, register_reply, request_message, |
| test_shard, |
| }; |
| use consensus::Sequencer; |
| use consensus::client_table::bind_verifier; |
| use iggy_binary_protocol::Command; |
| use iggy_binary_protocol::codes::{ |
| GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, POLL_MESSAGES_CODE, |
| }; |
| use iggy_binary_protocol::requests::users::{ |
| CreateUserRequest, DeleteUserRequest, UpdateUserRequest, |
| }; |
| use iggy_binary_protocol::{EvictionHeader, ReplyHeader}; |
| use iggy_binary_protocol::{WireEncode, WireIdentifier, WireName, WireOptions}; |
| use journal::prepare_journal::PrepareJournal; |
| use message_bus::installer::conn_info::ClientTransportKind; |
| use metadata::IggyMetadata; |
| use metadata::impls::metadata::IggySnapshot; |
| use metadata::stm::StateMachine; |
| use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig}; |
| use server_common::MESSAGE_ALIGN; |
| use server_common::sharding::ShardId; |
| use shard::metrics::ShardMetrics; |
| use shard::shards_table::PapayaShardsTable; |
| use shard::{ |
| ConnectedClientInfo, LifecycleFrame, ListClientsReply, MetadataSubmit, NoopHost, |
| PartitionConsensusConfig, ReplicaTopology, ShardFrame, ShardHost, ShardIdentity, |
| shard_channel, |
| }; |
| use std::mem::size_of; |
| use std::sync::atomic::AtomicBool; |
| |
| #[compio::test] |
| async fn bind_session_rechecks_the_owner_exists_and_is_active() { |
| const CLIENT: u128 = 1; |
| const TRANSPORT: u128 = 2; |
| const SESSION: u64 = 1; |
| const SECRET: [u8; 32] = [0x5a; 32]; |
| for status in [None, Some(UserStatus::Inactive), Some(UserStatus::Active)] { |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let metadata = shard.plane.metadata(); |
| if let Some(status) = status { |
| metadata.mux_stm.users().ensure_root_user("iggy", "hash"); |
| metadata |
| .mux_stm |
| .update(prepare_message( |
| Operation::UpdateUser, |
| CLIENT, |
| SESSION, |
| &UpdateUserRequest { |
| user_id: WireIdentifier::numeric(0), |
| username: None, |
| status: Some(status.as_code()), |
| options: WireOptions::empty(), |
| } |
| .to_bytes(), |
| )) |
| .unwrap(); |
| } |
| metadata |
| .client_table |
| .borrow_mut() |
| .commit_register( |
| CLIENT, |
| 0, |
| bind_verifier(CLIENT, 0, &SECRET), |
| register_reply(CLIENT, SESSION), |
| ) |
| .unwrap(); |
| metadata.applied_frontier().advance(SESSION); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| sessions.borrow_mut().ensure_connection( |
| TRANSPORT, |
| "127.0.0.1:34567".parse().unwrap(), |
| ClientTransportKind::Tcp, |
| ); |
| let outcome = complete_session_binding( |
| &shard, |
| &sessions, |
| TRANSPORT, |
| SessionIdentity { |
| client_id: CLIENT, |
| session: SESSION, |
| metadata_watermark: SESSION, |
| }, |
| BindSecret::new(Box::new(SECRET)), |
| ) |
| .await; |
| if status == Some(UserStatus::Active) { |
| assert_eq!(outcome.unwrap(), (0, SESSION)); |
| assert_eq!( |
| sessions.borrow().get_session(TRANSPORT), |
| Some((CLIENT, SESSION)) |
| ); |
| } else { |
| assert!( |
| matches!(outcome, Err(IggyError::Unauthenticated)), |
| "{status:?}: {outcome:?}" |
| ); |
| assert!(sessions.borrow().get_session(TRANSPORT).is_none()); |
| } |
| } |
| } |
| |
| #[compio::test] |
| async fn given_deleted_user_when_id_reused_should_reject_retained_session() { |
| for primary in [false, true] { |
| assert_revoked_user_cannot_resume(Operation::DeleteUser, primary).await; |
| } |
| } |
| |
| #[compio::test] |
| async fn given_deactivated_user_when_reactivated_should_reject_retained_session() { |
| for primary in [false, true] { |
| assert_revoked_user_cannot_resume(Operation::UpdateUser, primary).await; |
| } |
| } |
| |
| #[allow(clippy::too_many_lines)] |
| async fn assert_revoked_user_cannot_resume(operation: Operation, primary: bool) { |
| const CLIENT: u128 = 10; |
| const TRANSPORT: u128 = 20; |
| const SESSION: u64 = 2; |
| const USER: u32 = 1; |
| const ROOT_CLIENT: u128 = CLIENT + 1; |
| const SECRET: [u8; 32] = [0x5a; 32]; |
| let bus = SpyBus::default(); |
| let directory = tempfile::tempdir().unwrap(); |
| let mut shard = test_shard(&bus, 0, 1, FIRST_BOOT); |
| shard.plane.metadata_mut().journal = Some( |
| PrepareJournal::open(&directory.path().join("journal.wal"), 0) |
| .await |
| .unwrap(), |
| ); |
| let shard = Rc::new(shard); |
| let metadata = shard.plane.metadata(); |
| metadata.mux_stm.users().ensure_root_user("iggy", "hash"); |
| let create = CreateUserRequest { |
| username: WireName::new("original").unwrap(), |
| password: "hash".to_owned(), |
| status: UserStatus::Active.as_code(), |
| permissions: None, |
| options: WireOptions::empty(), |
| }; |
| let created = metadata |
| .mux_stm |
| .update(prepare_message( |
| Operation::CreateUser, |
| 0, |
| 1, |
| &create.to_bytes(), |
| )) |
| .unwrap(); |
| assert_eq!(created.code, 0); |
| metadata |
| .client_table |
| .borrow_mut() |
| .commit_register( |
| CLIENT, |
| USER, |
| bind_verifier(CLIENT, USER, &SECRET), |
| register_reply(CLIENT, SESSION), |
| ) |
| .unwrap(); |
| metadata.applied_frontier().advance(SESSION); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| sessions.borrow_mut().ensure_connection( |
| TRANSPORT, |
| "127.0.0.1:34567".parse().unwrap(), |
| ClientTransportKind::Tcp, |
| ); |
| let identity = SessionIdentity { |
| client_id: CLIENT, |
| session: SESSION, |
| metadata_watermark: SESSION, |
| }; |
| complete_session_binding( |
| &shard, |
| &sessions, |
| TRANSPORT, |
| identity, |
| BindSecret::new(Box::new(SECRET)), |
| ) |
| .await |
| .unwrap(); |
| let body = if operation == Operation::DeleteUser { |
| DeleteUserRequest { |
| user_id: WireIdentifier::numeric(USER), |
| } |
| .to_bytes() |
| } else { |
| UpdateUserRequest { |
| user_id: WireIdentifier::numeric(USER), |
| username: None, |
| status: Some(UserStatus::Inactive.as_code()), |
| options: WireOptions::empty(), |
| } |
| .to_bytes() |
| }; |
| if primary { |
| metadata |
| .client_table |
| .borrow_mut() |
| .commit_register(ROOT_CLIENT, 0, SECRET, register_reply(ROOT_CLIENT, 1)) |
| .unwrap(); |
| let consensus = metadata.consensus.as_ref().unwrap(); |
| consensus.restore_commit_state(SESSION, SESSION); |
| consensus.sequencer().set_sequence(SESSION); |
| let request = request_message(operation, ROOT_CLIENT, 1, 1, &body).transmute_header( |
| |old, header: &mut RoutedRequestHeader| { |
| *header = old; |
| header.metadata_watermark = SESSION; |
| }, |
| ); |
| metadata.submit_request_in_process(request).await.unwrap(); |
| } else { |
| let revocation = prepare_message(operation, 0, 3, &body).transmute_header( |
| |old, header: &mut iggy_binary_protocol::PrepareHeader| { |
| *header = old; |
| header.op = 3; |
| }, |
| ); |
| metadata::apply_committed_prepare( |
| &*metadata.mux_stm, |
| &metadata.client_table, |
| true, |
| |_| {}, |
| consensus::seal_prepare_checksum(revocation), |
| ); |
| } |
| let old_binding = sessions.borrow().get_session(TRANSPORT); |
| let restore = if operation == Operation::DeleteUser { |
| prepare_message( |
| Operation::CreateUser, |
| 0, |
| 4, |
| &CreateUserRequest { |
| username: WireName::new("replacement").unwrap(), |
| ..create |
| } |
| .to_bytes(), |
| ) |
| } else { |
| prepare_message( |
| Operation::UpdateUser, |
| 0, |
| 4, |
| &UpdateUserRequest { |
| user_id: WireIdentifier::numeric(USER), |
| username: None, |
| status: Some(UserStatus::Active.as_code()), |
| options: WireOptions::empty(), |
| } |
| .to_bytes(), |
| ) |
| }; |
| assert_eq!(metadata.mux_stm.update(restore).unwrap().code, 0); |
| assert!(metadata.mux_stm.users().read(|users| { |
| users |
| .items |
| .get(USER as usize) |
| .is_some_and(|user| user.status == UserStatus::Active) |
| })); |
| let outcome = complete_session_binding( |
| &shard, |
| &sessions, |
| TRANSPORT, |
| identity, |
| BindSecret::new(Box::new(SECRET)), |
| ) |
| .await; |
| assert!( |
| matches!(outcome, Err(IggyError::Unauthenticated)), |
| "committed {operation:?} must reject the old secret after restoring user id {USER}: {outcome:?}" |
| ); |
| assert!( |
| old_binding.is_none(), |
| "committed {operation:?} must invalidate an already authenticated connection" |
| ); |
| assert_eq!(metadata.client_table.borrow().ended_sessions().count(), 1); |
| } |
| |
| /// A test shard wired to its own lanes (the held sender feeds them), |
| /// for the reply-lane pump tests below. |
| fn reply_lane_test_shard(name: &str) -> (SpyBus, shard::TaggedSender, Rc<TestShard>) { |
| // These tests do not dispatch disk polls, so keep that lane minimal. |
| const POLL_COMPLETION_CAPACITY: usize = 1; |
| |
| let bus = SpyBus::default(); |
| let metadata = IggyMetadata::new(None, None, None, 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), |
| validate_checksum: true, |
| segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), |
| preallocate_segments: false, |
| encryptor: None, |
| path_layout: PartitionPathLayout::default(), |
| }, |
| ); |
| let (sender, inbox_rx, reply_inbox_rx) = shard_channel(0, 16, 16); |
| let lane_sender = sender.clone(); |
| let shard = TestShard::new( |
| ShardIdentity::new(0, name.to_string()), |
| bus.clone(), |
| Rc::new(NoopHost), |
| metadata, |
| partitions, |
| vec![sender], |
| inbox_rx, |
| reply_inbox_rx, |
| POLL_COMPLETION_CAPACITY, |
| None, |
| PapayaShardsTable::new(), |
| PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1), bus.clone()), |
| None, |
| ShardMetrics::for_shard(), |
| ) |
| .expect("single-sender ring is canonically ordered"); |
| (bus, lane_sender, Rc::new(shard)) |
| } |
| |
| fn reply_lane_forward(client_id: u128) -> ShardFrame { |
| ShardFrame::lifecycle(LifecycleFrame::ForwardClientSend { |
| client_id, |
| msg: server_common::iobuf::Frozen::from( |
| server_common::iobuf::Owned::<MESSAGE_ALIGN>::zeroed(64), |
| ) |
| .into(), |
| }) |
| } |
| |
| /// A frame on the reply lane must reach the client through the RUNNING |
| /// pump's reply arm: the lane split moved `ForwardClientSend` off the |
| /// main inbox, so a pump that forgot to service the new lane would |
| /// strand every cross-shard reply while the send sites happily report |
| /// success. |
| #[compio::test] |
| async fn pump_live_arm_delivers_reply_lane_forwards() { |
| const TRANSPORT: u128 = 92; |
| let (bus, lane_sender, shard) = reply_lane_test_shard("reply-lane-live-arm-test"); |
| |
| let (stop_tx, stop_rx) = shard::channel::<()>(1); |
| let pump_shard = Rc::clone(&shard); |
| let pump = compio::runtime::spawn(async move { |
| pump_shard |
| .run_message_pump(stop_rx, Arc::new(AtomicBool::new(false))) |
| .await; |
| }); |
| |
| lane_sender |
| .reply_sender() |
| .try_send(reply_lane_forward(TRANSPORT)) |
| .expect("reply lane has capacity"); |
| |
| // The pump is idle on the main lane, so its bottom reply arm must |
| // serve the frame without any main-lane traffic or shutdown drain. |
| let mut delivered = false; |
| for _ in 0..500 { |
| if !bus.client_replies.borrow().is_empty() { |
| delivered = true; |
| break; |
| } |
| compio::time::sleep(std::time::Duration::from_millis(1)).await; |
| } |
| stop_tx.try_send(()).expect("stop channel has capacity"); |
| let _ = pump.await; |
| |
| assert!( |
| delivered, |
| "the live reply arm must deliver a forward while the pump runs" |
| ); |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies[0].0, TRANSPORT, "forward must reach its client"); |
| } |
| |
| /// The shutdown path must ALSO deliver reply-lane frames: a forward |
| /// already accepted by the lane when the stop signal wins the biased |
| /// select would otherwise be silently destroyed at teardown. |
| #[compio::test] |
| async fn pump_shutdown_drain_delivers_reply_lane_forwards() { |
| const TRANSPORT: u128 = 93; |
| let (bus, lane_sender, shard) = reply_lane_test_shard("reply-lane-drain-test"); |
| |
| lane_sender |
| .reply_sender() |
| .try_send(reply_lane_forward(TRANSPORT)) |
| .expect("reply lane has capacity"); |
| |
| // Pre-armed stop: the pump exits through the biased stop arm and the |
| // post-loop drain must still deliver the reply-lane frame. |
| let (stop_tx, stop_rx) = shard::channel::<()>(1); |
| stop_tx.try_send(()).expect("stop channel has capacity"); |
| shard |
| .run_message_pump(stop_rx, Arc::new(AtomicBool::new(false))) |
| .await; |
| |
| let replies = bus.client_replies.borrow(); |
| assert_eq!( |
| replies.len(), |
| 1, |
| "the pump's reply-lane drain must deliver the forwarded reply" |
| ); |
| assert_eq!( |
| replies[0].0, TRANSPORT, |
| "the forward must reach the client it was addressed to" |
| ); |
| } |
| |
| /// A shard built on [`NoopHost`] routes the host-bound lifecycle frames |
| /// like any other: the host drops the carried reply sender, so the |
| /// waiting side sees a disconnect instead of hanging on the pump, and |
| /// the pump goes on serving frames afterwards. |
| #[compio::test] |
| async fn noop_host_drops_host_bound_replies_and_keeps_the_pump_alive() { |
| const TRANSPORT: u128 = 94; |
| const BUDGET: std::time::Duration = std::time::Duration::from_secs(5); |
| let (bus, lane_sender, shard) = reply_lane_test_shard("noop-host-test"); |
| let (list_reply, list_rx) = shard::channel::<Vec<ConnectedClientInfo>>(1); |
| let (submit_reply, submit_rx) = shard::channel::<Option<Message<GenericHeader>>>(1); |
| lane_sender |
| .try_send(ShardFrame::lifecycle(LifecycleFrame::ListClients { |
| reply: ListClientsReply::Clients(list_reply), |
| })) |
| .expect("inbox has capacity"); |
| lane_sender |
| .try_send(ShardFrame::lifecycle(LifecycleFrame::MetadataSubmit( |
| MetadataSubmit::ClientRequest { |
| request: Message::<GenericHeader>::new(size_of::<GenericHeader>()), |
| reply: submit_reply, |
| }, |
| ))) |
| .expect("inbox has capacity"); |
| |
| let (stop_tx, stop_rx) = shard::channel::<()>(1); |
| let pump_shard = Rc::clone(&shard); |
| let pump = compio::runtime::spawn(async move { |
| pump_shard |
| .run_message_pump(stop_rx, Arc::new(AtomicBool::new(false))) |
| .await; |
| }); |
| |
| let list = compio::time::timeout(BUDGET, list_rx.recv()) |
| .await |
| .expect("the pump must process the list-clients frame"); |
| let submit = compio::time::timeout(BUDGET, submit_rx.recv()) |
| .await |
| .expect("the pump must process the metadata-submit frame"); |
| assert!( |
| list.is_err(), |
| "a noop host answers list-clients by dropping the reply sender" |
| ); |
| assert!( |
| submit.is_err(), |
| "a noop host answers metadata-submit by dropping the reply sender" |
| ); |
| |
| // Sent only now, so its delivery proves the pump outlived both |
| // host-bound frames rather than being served between them. |
| lane_sender |
| .reply_sender() |
| .try_send(reply_lane_forward(TRANSPORT)) |
| .expect("reply lane has capacity"); |
| compio::time::timeout(BUDGET, async { |
| while bus.client_replies.borrow().is_empty() { |
| compio::time::sleep(std::time::Duration::from_millis(1)).await; |
| } |
| }) |
| .await |
| .expect("the pump must keep serving after the host-bound frames"); |
| stop_tx.try_send(()).expect("stop channel has capacity"); |
| let _ = pump.await; |
| |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies[0].0, TRANSPORT, "forward must reach its client"); |
| } |
| |
| /// The `GET_CLUSTER_METADATA` auth gate holds on every roster shape: it |
| /// describes the private replica network, and a client that dialed a |
| /// backup reaches the cluster by logging in there (the backup forwards |
| /// the register), not by reading the topology first. |
| /// |
| /// The denial must be a plain Reply on the status channel, not an |
| /// Eviction: no session exists yet, and a session-terminal frame makes |
| /// SDKs drop the connection their login is about to use. |
| #[compio::test] |
| async fn pre_auth_cluster_metadata_denied_on_every_roster() { |
| use configs::cluster::{ClusterNodeConfig, TransportPorts}; |
| use iggy_binary_protocol::{GenericHeader, ReplyHeader}; |
| |
| const TRANSPORT: u128 = 91; |
| const COMMAND_OFFSET: usize = std::mem::offset_of!(GenericHeader, command); |
| const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status); |
| const OP_OFFSET: usize = std::mem::offset_of!(ReplyHeader, op); |
| const COMMIT_OFFSET: usize = std::mem::offset_of!(ReplyHeader, commit); |
| |
| fn metadata_read() -> Message<GenericHeader> { |
| let header_size = size_of::<RequestHeader>(); |
| let mut message = Message::<RequestHeader>::new(header_size); |
| { |
| let header = bytemuck::checked::from_bytes_mut::<RequestHeader>( |
| &mut message.as_mut_slice()[..header_size], |
| ); |
| *header = RequestHeader { |
| command: Command::Request, |
| operation: Operation::NonReplicated, |
| size: u32::try_from(header_size).expect("header fits u32"), |
| client: TRANSPORT, |
| ..Default::default() |
| }; |
| header.reserved[..4].copy_from_slice(&GET_CLUSTER_METADATA_CODE.to_le_bytes()); |
| } |
| message.into_generic() |
| } |
| |
| fn roster_node(name: &str) -> ClusterNodeConfig { |
| ClusterNodeConfig { |
| name: name.to_owned(), |
| ip: "127.0.0.1".to_owned(), |
| advertised_address: None, |
| advertised_addresses: Vec::new(), |
| replica_id: 0, |
| ports: TransportPorts::default(), |
| } |
| } |
| |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| let server_config = Arc::new(ServerConfig::default()); |
| |
| let multi_node = Rc::new(ClusterRoster { |
| enabled: true, |
| name: "test-cluster".to_owned(), |
| nodes: ["node-0", "node-1"] |
| .map(|name| { |
| configs::cluster::ResolvedClusterNode::try_from(roster_node(name)) |
| .expect("valid roster node") |
| }) |
| .to_vec(), |
| self_advertised: "127.0.0.1".to_owned(), |
| configured_ports: TransportPorts::default(), |
| bound_ports: Arc::default(), |
| metadata_view: Arc::new(std::sync::atomic::AtomicU64::new( |
| crate::cluster_meta::METADATA_VIEW_UNKNOWN, |
| )), |
| }); |
| // Default roster is disabled / single node; the installed one is a |
| // real cluster. Neither serves an unbound caller. |
| for roster in [None, Some(multi_node)] { |
| if let Some(roster) = roster { |
| sessions.borrow_mut().set_cluster_roster(roster); |
| } |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| 1, |
| TRANSPORT, |
| metadata_read(), |
| ) |
| .await; |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies.len(), 1, "gated read must still produce a frame"); |
| let (client, frame) = &replies[0]; |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!( |
| frame[COMMAND_OFFSET], |
| Command::Reply as u8, |
| "an unbound cluster-metadata read must be denied with a Reply, not evicted" |
| ); |
| let status = |
| u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap()); |
| assert_eq!( |
| status, |
| IggyError::Unauthenticated.as_code(), |
| "deny reply status must be Unauthenticated" |
| ); |
| let op = u64::from_le_bytes(frame[OP_OFFSET..OP_OFFSET + 8].try_into().unwrap()); |
| assert_eq!(op, 0, "pre-auth deny carries no session, so op must be 0"); |
| let commit = |
| u64::from_le_bytes(frame[COMMIT_OFFSET..COMMIT_OFFSET + 8].try_into().unwrap()); |
| assert_eq!(commit, 0, "pre-auth deny must not disclose commit activity"); |
| drop(replies); |
| bus.client_replies.borrow_mut().clear(); |
| } |
| } |
| |
| /// Client-wire frame for the funnel tests, shaped like the SDK sends it |
| /// (the funnel promotes it to the routed shape itself). |
| fn wire_request( |
| operation: Operation, |
| client: u128, |
| session: u64, |
| request: u64, |
| body: &[u8], |
| ) -> Message<RequestHeader> { |
| let header_size = size_of::<RequestHeader>(); |
| let total = header_size + body.len(); |
| let mut message = Message::<RequestHeader>::new(total); |
| { |
| let slice = message.as_mut_slice(); |
| slice[header_size..total].copy_from_slice(body); |
| let header = |
| bytemuck::checked::from_bytes_mut::<RequestHeader>(&mut slice[..header_size]); |
| *header = RequestHeader { |
| command: Command::Request, |
| operation, |
| size: u32::try_from(total).expect("test request fits u32"), |
| client, |
| session, |
| request, |
| ..Default::default() |
| }; |
| } |
| message |
| } |
| |
| fn non_replicated_request(client: u128, nr_code: u32) -> Message<GenericHeader> { |
| let mut message = wire_request(Operation::NonReplicated, client, 0, 0, &[]); |
| { |
| let header_size = size_of::<RequestHeader>(); |
| let header = bytemuck::checked::from_bytes_mut::<RequestHeader>( |
| &mut message.as_mut_slice()[..header_size], |
| ); |
| header.reserved[..4].copy_from_slice(&nr_code.to_le_bytes()); |
| } |
| message.into_generic() |
| } |
| |
| const fn frame_command(frame: &[u8]) -> u8 { |
| frame[std::mem::offset_of!(GenericHeader, command)] |
| } |
| |
| const fn eviction_reason_byte(frame: &[u8]) -> u8 { |
| frame[std::mem::offset_of!(EvictionHeader, reason)] |
| } |
| |
| fn reply_status(frame: &[u8]) -> u32 { |
| const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status); |
| u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap()) |
| } |
| |
| /// The classifier is the funnel's probe chain as data: every row pins |
| /// one routing decision, and the labeled rows pin the probe ORDERINGS |
| /// the funnel relies on. |
| #[allow(clippy::too_many_lines)] |
| #[test] |
| fn classify_pins_probe_order() { |
| fn routed(operation: Operation, session: u64, request: u64) -> RoutedRequestHeader { |
| RoutedRequestHeader { |
| command: Command::Request, |
| operation, |
| client: 7, |
| session, |
| request, |
| ..Default::default() |
| } |
| } |
| fn nr(nr_code: u32) -> RoutedRequestHeader { |
| let mut header = routed(Operation::NonReplicated, 0, 0); |
| header.reserved[..4].copy_from_slice(&nr_code.to_le_bytes()); |
| header |
| } |
| let table = [ |
| ( |
| "legacy login beats the session gate (unbound)", |
| nr(LOGIN_USER_CODE), |
| false, |
| RequestClass::LegacyLogin, |
| ), |
| ( |
| "legacy login beats the bound reads route", |
| nr(LOGIN_USER_CODE), |
| true, |
| RequestClass::LegacyLogin, |
| ), |
| ( |
| "legacy PAT login beats the session gate (unbound)", |
| nr(LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE), |
| false, |
| RequestClass::LegacyLogin, |
| ), |
| ( |
| "legacy PAT login beats the bound reads route", |
| nr(LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE), |
| true, |
| RequestClass::LegacyLogin, |
| ), |
| ( |
| "ping is the only pre-auth code", |
| nr(PING_CODE), |
| false, |
| RequestClass::NonReplicatedRead, |
| ), |
| ( |
| "cluster metadata is not pre-auth", |
| nr(GET_CLUSTER_METADATA_CODE), |
| false, |
| RequestClass::UnauthenticatedRead, |
| ), |
| ( |
| "poll-messages is a non-replicated code, not a partition op", |
| nr(POLL_MESSAGES_CODE), |
| true, |
| RequestClass::NonReplicatedRead, |
| ), |
| ( |
| "consumer-offset read is a non-replicated code, not a partition op", |
| nr(GET_CONSUMER_OFFSET_CODE), |
| true, |
| RequestClass::NonReplicatedRead, |
| ), |
| ( |
| "the register handshake", |
| routed(Operation::Register, 0, 0), |
| false, |
| RequestClass::LoginRegister, |
| ), |
| ( |
| "register with a session falls to the default", |
| routed(Operation::Register, 1, 0), |
| true, |
| RequestClass::ReplicatedMetadata, |
| ), |
| ( |
| "register with a request number falls to the default", |
| routed(Operation::Register, 0, 1), |
| true, |
| RequestClass::ReplicatedMetadata, |
| ), |
| ( |
| "logout, bound", |
| routed(Operation::Logout, 1, 1), |
| true, |
| RequestClass::Logout, |
| ), |
| ( |
| "logout beats the session gate", |
| routed(Operation::Logout, 1, 1), |
| false, |
| RequestClass::Logout, |
| ), |
| ( |
| "unbound metadata op", |
| routed(Operation::CreateStream, 1, 1), |
| false, |
| RequestClass::UnboundReplicated, |
| ), |
| ( |
| "the session gate beats the delete-segments probe", |
| routed(Operation::DeleteSegments, 1, 1), |
| false, |
| RequestClass::UnboundReplicated, |
| ), |
| ( |
| "the session gate beats the partition probe", |
| routed(Operation::SendMessages, 1, 1), |
| false, |
| RequestClass::UnboundReplicated, |
| ), |
| ( |
| "delete-segments is neither partition nor metadata", |
| routed(Operation::DeleteSegments, 1, 1), |
| true, |
| RequestClass::DeleteSegments, |
| ), |
| ( |
| "send-messages is partition-plane", |
| routed(Operation::SendMessages, 1, 1), |
| true, |
| RequestClass::Partition, |
| ), |
| ( |
| "store-consumer-offset is partition-plane", |
| routed(Operation::StoreConsumerOffset, 1, 1), |
| true, |
| RequestClass::Partition, |
| ), |
| ( |
| "create-topic is replicated metadata", |
| routed(Operation::CreateTopic, 1, 1), |
| true, |
| RequestClass::ReplicatedMetadata, |
| ), |
| ]; |
| for (label, header, bound, expected) in table { |
| assert_eq!(classify(&header, bound), expected, "{label}"); |
| } |
| } |
| |
| /// A legacy login code must get the typed `MalformedLogin` eviction |
| /// BEFORE the session gate runs: an unbound sender must not fall into |
| /// the generic Unauthenticated deny the pre-auth guard sends for other |
| /// codes. |
| #[compio::test] |
| async fn legacy_login_codes_evicted_before_session_gate() { |
| const TRANSPORT: u128 = 94; |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| let server_config = Arc::new(ServerConfig::default()); |
| |
| for code in [LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE] { |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| 1, |
| TRANSPORT, |
| non_replicated_request(TRANSPORT, code), |
| ) |
| .await; |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies.len(), 1, "code {code} must produce one frame"); |
| let (client, frame) = &replies[0]; |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!( |
| frame_command(frame), |
| Command::Eviction as u8, |
| "legacy code {code} must be evicted, not denied" |
| ); |
| assert_eq!( |
| eviction_reason_byte(frame), |
| EvictionReason::MalformedLogin as u8, |
| "legacy code {code} must carry MalformedLogin" |
| ); |
| drop(replies); |
| bus.client_replies.borrow_mut().clear(); |
| } |
| } |
| |
| /// A replicated op from an unbound transport must get the typed |
| /// `Eviction(NoSession)`: the client believes it has a session and that |
| /// session is gone, so a silent drop or an empty Reply would wedge or |
| /// mislead it. |
| #[compio::test] |
| async fn unbound_replicated_request_gets_no_session_eviction() { |
| const TRANSPORT: u128 = 95; |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| let server_config = Arc::new(ServerConfig::default()); |
| |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| 1, |
| TRANSPORT, |
| wire_request(Operation::CreateStream, TRANSPORT, 1, 1, &[]).into_generic(), |
| ) |
| .await; |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies.len(), 1, "unbound replicated op must be answered"); |
| let (client, frame) = &replies[0]; |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!( |
| frame_command(frame), |
| Command::Eviction as u8, |
| "unbound replicated op must be evicted, not denied or dropped" |
| ); |
| assert_eq!( |
| eviction_reason_byte(frame), |
| EvictionReason::NoSession as u8, |
| "the eviction must carry NoSession" |
| ); |
| } |
| |
| /// PING is the one pre-auth code: an unbound transport's ping must get a |
| /// normal status-0 Reply, not an eviction and not a deny. |
| #[compio::test] |
| async fn pre_auth_ping_allowed() { |
| const TRANSPORT: u128 = 96; |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| let server_config = Arc::new(ServerConfig::default()); |
| |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| 1, |
| TRANSPORT, |
| non_replicated_request(TRANSPORT, PING_CODE), |
| ) |
| .await; |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies.len(), 1, "pre-auth ping must be answered"); |
| let (client, frame) = &replies[0]; |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!( |
| frame_command(frame), |
| Command::Reply as u8, |
| "pre-auth ping must get a Reply, not an eviction" |
| ); |
| assert_eq!(reply_status(frame), 0, "pre-auth ping must succeed"); |
| } |
| |
| /// The request-checksum gate runs before every probe: a replicated op |
| /// from an UNBOUND transport with a bad stamp must get the checksum deny |
| /// Reply, not the `Eviction(NoSession)` the session gate would send. |
| #[compio::test] |
| async fn checksum_mismatch_denies_before_everything() { |
| const TRANSPORT: u128 = 97; |
| const BODY: &[u8] = b"stream-body"; |
| let bus = SpyBus::default(); |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| let sessions = Rc::new(RefCell::new(SessionManager::new())); |
| let server_config = Arc::new(ServerConfig::default()); |
| |
| let mut message = wire_request(Operation::CreateStream, TRANSPORT, 1, 1, BODY); |
| { |
| let header_size = size_of::<RequestHeader>(); |
| let header = bytemuck::checked::from_bytes_mut::<RequestHeader>( |
| &mut message.as_mut_slice()[..header_size], |
| ); |
| // Nonzero (zero means unstamped and skips the check) and never |
| // the body's real checksum. |
| header.request_checksum = u128::from(iggy_common::calculate_checksum(BODY)) + 1; |
| } |
| handle_client_request( |
| &shard, |
| &sessions, |
| &server_config, |
| 1, |
| TRANSPORT, |
| message.into_generic(), |
| ) |
| .await; |
| let replies = bus.client_replies.borrow(); |
| assert_eq!(replies.len(), 1, "bad stamp must be answered"); |
| let (client, frame) = &replies[0]; |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!( |
| frame_command(frame), |
| Command::Reply as u8, |
| "a bad stamp must get the checksum deny, not the NoSession eviction" |
| ); |
| assert_eq!( |
| reply_status(frame), |
| IggyError::InvalidFormat.as_code(), |
| "the deny status must carry the checksum error" |
| ); |
| } |
| |
| /// The late-bound self-reference as the shard build creates it: unset |
| /// until the shard exists. |
| fn unset_shard_handle() -> ShellShardHandle<SpyBus, PrepareJournal, IggySnapshot> { |
| Rc::new(RefCell::new(None)) |
| } |
| |
| /// Yield so the drain task the enqueue spawned on the bus can run. |
| async fn run_spawned_tasks() { |
| compio::time::sleep(std::time::Duration::from_millis(1)).await; |
| } |
| |
| /// One host per shard means one connection-lost hook on its bus: a |
| /// second install would overwrite the first and orphan the sessions |
| /// the first one bound. |
| #[test] |
| fn server_host_installs_one_connection_lost_hook() { |
| let bus = SpyBus::default(); |
| let _host = ServerHost::new( |
| &bus, |
| &unset_shard_handle(), |
| Arc::new(ServerConfig::default()), |
| 1, |
| ); |
| assert_eq!( |
| bus.connection_lost_hooks.get(), |
| 1, |
| "one host must install exactly one connection-lost hook" |
| ); |
| } |
| |
| /// The host is built before the shard it serves. A frame that |
| /// arrives while the self-reference is still unset stays queued, and |
| /// the enqueue must release the client's active slot: otherwise every |
| /// later frame for that client finds the slot taken and nothing is |
| /// ever drained, the stranded frame included. |
| #[compio::test] |
| async fn server_host_drains_after_the_shard_handle_is_set() { |
| const TRANSPORT: u128 = 98; |
| let bus = SpyBus::default(); |
| let shard_handle = unset_shard_handle(); |
| let host = ServerHost::new(&bus, &shard_handle, Arc::new(ServerConfig::default()), 1); |
| |
| host.on_client_request(TRANSPORT, non_replicated_request(TRANSPORT, PING_CODE)); |
| run_spawned_tasks().await; |
| assert!( |
| bus.client_replies.borrow().is_empty(), |
| "nothing can be served before the shard exists" |
| ); |
| |
| let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT)); |
| *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard)); |
| host.on_client_request(TRANSPORT, non_replicated_request(TRANSPORT, PING_CODE)); |
| for _ in 0..500 { |
| if bus.client_replies.borrow().len() == 2 { |
| break; |
| } |
| run_spawned_tasks().await; |
| } |
| |
| let replies = bus.client_replies.borrow(); |
| assert_eq!( |
| replies.len(), |
| 2, |
| "the stranded ping and the new one must both be served once the shard exists" |
| ); |
| for (client, frame) in replies.iter() { |
| assert_eq!(*client, TRANSPORT); |
| assert_eq!(frame_command(frame), Command::Reply as u8); |
| assert_eq!(reply_status(frame), 0, "a served ping succeeds"); |
| } |
| } |
| |
| /// The per-client queue entry goes with its last frame. The |
| /// connection-lost hook cannot be its only owner: the dispatch task still |
| /// delivers frames it had buffered when the socket closed, which re-create |
| /// the entry after the hook ran, and `client_id` is monotonic - so an |
| /// entry nothing frees again lives until process exit. |
| #[test] |
| fn draining_a_client_queue_frees_its_entry() { |
| const CLIENT: u128 = 7; |
| let queues: ClientRequestQueues = Rc::new(RefCell::new(AHashMap::new())); |
| queues |
| .borrow_mut() |
| .entry(CLIENT) |
| .or_default() |
| .push_back(non_replicated_request(CLIENT, PING_CODE)); |
| |
| assert!(pop_next_client_request(&queues, CLIENT).is_some()); |
| assert!( |
| queues.borrow().is_empty(), |
| "the entry must be freed as soon as it drains empty" |
| ); |
| assert!( |
| pop_next_client_request(&queues, CLIENT).is_none(), |
| "a drained client has nothing left to pop" |
| ); |
| } |
| } |
