hubcio commented on code in PR #4506: URL: https://github.com/apache/iggy/pull/4506#discussion_r4261451684
########## core/sdk/src/links/mux.rs: ########## @@ -0,0 +1,815 @@ +// 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. + +//! One TCP or WebSocket connection that carries many requests at once. +//! +//! A writer task and a reader task own the two halves of the connection, and +//! each reply goes to the request with its id. So a caller that goes away +//! never leaves a frame half written or half read on the connection. + +use crate::links::{Frame, Reply}; +use crate::vsr::{decode_eviction, peek_command, reply_request_id, response_size}; +use bytes::{BufMut, BytesMut}; +use futures_util::stream::{SplitSink, SplitStream}; +use futures_util::{SinkExt, StreamExt}; +use iggy_binary_protocol::consensus::{Command, HEADER_SIZE}; +use iggy_common::IggyError; +use std::collections::HashMap; +use std::collections::hash_map::Entry; +use std::future::Future; +use std::io; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex as StdMutex, MutexGuard, PoisonError}; +use std::time::Duration; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}; +use tokio::sync::{Mutex, Semaphore, mpsc, oneshot, watch}; +use tokio::task::JoinHandle; +use tokio::time::{Instant, timeout, timeout_at}; +use tokio_tungstenite::WebSocketStream; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::protocol::CloseFrame; +use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode; +use tracing::{debug, error, trace}; + +/// Frames queued for the writer task before a sender waits for room. +const LINK_WRITE_QUEUE: usize = 32; +/// Requests that wait for a reply on one link at once. It matches +/// `message_bus.client_queue_capacity`, the most requests the server runs at +/// once for one connection. +const MAX_PENDING_PER_LINK: usize = 256; +/// A peer that stops reading cannot hold up a shutdown for longer than this. +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); +const WEBSOCKET_CLOSE_REASON: &str = "Client requested shutdown"; + +type Waiter = oneshot::Sender<Result<Reply, IggyError>>; + +/// A connection that carries many requests at once. Dropping the link stops +/// its tasks, which closes the connection. +#[derive(Debug)] +pub(crate) struct MuxLink { + shared: Arc<MuxShared>, + queue: mpsc::Sender<Frame>, + tasks: LinkTasks, +} + +#[derive(Debug)] +struct MuxShared { + peer: SocketAddr, + pending: StdMutex<Pending>, + permits: Semaphore, + /// Before bind, request ids repeat, so one such exchange runs at a time. + unbound: Mutex<()>, + /// Turns true once, when the link closes. + closed: watch::Sender<bool>, +} + +#[derive(Debug)] +struct Pending { + /// Cleared when the link closes. A closed link never opens again. + open: bool, + waiters: HashMap<u64, Waiter>, +} + +#[derive(Debug)] +struct LinkTasks { + writer: JoinHandle<()>, + reader: JoinHandle<()>, +} + +impl Drop for LinkTasks { + fn drop(&mut self) { + self.writer.abort(); + self.reader.abort(); + } +} + +/// A request that waits for its reply. Dropping it ends the wait. +struct PendingEntry<'a> { + shared: &'a MuxShared, + request: u64, + /// Set for a repeated id that reached the queue: its late reply would + /// answer the next request with the same id. + close_if_abandoned: bool, +} + +impl MuxLink { + /// A link over the two halves of a TCP connection, plain or TLS. + pub(crate) fn tcp<R, W>(reader: R, writer: W, peer: SocketAddr) -> Self + where + R: AsyncRead + Unpin + Send + 'static, + W: AsyncWrite + Unpin + Send + 'static, + { + Self::start( + peer, + |frames, shared| write_tcp(BufWriter::new(writer), frames, shared), + |shared| read_tcp(BufReader::new(reader), shared), + ) + } + + /// A link over a WebSocket connection, plain or TLS. + pub(crate) fn websocket<S>(stream: WebSocketStream<S>, peer: SocketAddr) -> Self + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + let (sink, stream) = stream.split(); + Self::start( + peer, + |frames, shared| write_websocket(sink, frames, shared), + |shared| read_websocket(stream, shared), + ) + } + + /// Sends `frame` and waits for its reply until `deadline`. `sent` turns + /// true when the frame reaches the queue, so a caller that goes away + /// before the reply still knows that the frame left. + /// + /// `NotConnected` means the frame never reached the queue, so the request + /// can go again. `Disconnected` or an eviction reason means it did, and no + /// reply came. + pub(crate) async fn send( + &self, + frame: Frame, + deadline: Instant, + sent: &mut bool, + ) -> Result<Reply, IggyError> { + let request = frame.header.request; + let unbound = frame.header.session == 0; + let _unbound_turn = if unbound { + let turn = timeout_at(deadline, self.shared.unbound.lock()).await; + Some(turn.map_err(|_| IggyError::NotConnected)?) + } else { + None + }; + let _permit = timeout_at(deadline, self.shared.permits.acquire()) + .await + .map_err(|_| IggyError::NotConnected)? + .map_err(|_| IggyError::NotConnected)?; + let (waiter, reply) = oneshot::channel(); + let mut entry = self.shared.wait_for(request, waiter)?; + if !matches!( + timeout_at(deadline, self.queue.send(frame)).await, + Ok(Ok(())) + ) { + return Err(IggyError::NotConnected); + } + *sent = true; + entry.close_if_abandoned = unbound; + match timeout_at(deadline, reply).await { + Ok(Ok(result)) => result, + Ok(Err(_)) => Err(IggyError::Disconnected), Review Comment: Deferred to multi-leader PR5 with the transport/driver error-contract audit. Separate exchange failures normalized to Disconnected from establishment paths that still use transport-specific errors, then remove only unreachable arms and document downstream matching changes. ########## core/sdk/src/links/replicas.rs: ########## @@ -0,0 +1,1571 @@ +// 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. + +//! One link per replica of the roster, and the sweep that signs in. +//! +//! `connect()` sweeps the known endpoints. The first link that answers +//! becomes the bootstrap link, and the session signs in there. A read of the +//! roster then gives one slot per replica, keyed by its replica id. Each slot +//! runs one task that dials its replica, binds the session there and redials +//! after a loss. A request never dials: it takes a ready link or fails with +//! `NotConnected`. + +use super::connect::{ + ConnectCoordinator, is_same_address, is_same_spelling, node_address, transport_port, +}; +use super::{ + Dialer, Link, OPERATION_TIMEOUT, backoff, event_channel, exchange, lost_link, + pause_before_retry, +}; +use crate::SDK_VERSION; +use crate::session_owner::{SessionEpoch, SessionOwner}; +use async_broadcast::{Receiver, Sender}; +use futures::future::join_all; +use iggy_binary_protocol::codec::{WireDecode, WireEncode}; +use iggy_binary_protocol::codes::{BIND_SESSION_CODE, GET_CLUSTER_METADATA_CODE, PING_CODE}; +use iggy_binary_protocol::requests::system::{ + BindSessionRequest, GetClusterMetadataRequest, PingRequest, +}; +use iggy_binary_protocol::responses::system::ClusterMetadataResponse; +use iggy_binary_protocol::responses::users::LoginRegisterResponse; +use iggy_common::{ + ClientState, ClusterMetadata, ClusterNodeRole, Credentials, DiagnosticEvent, IggyError, + QuicClientReconnectionConfig, TcpClientReconnectionConfig, WebSocketClientReconnectionConfig, + decode_session_binding, rust_sdk_version_info, sign_in_request, +}; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use std::time::Duration; +use tokio::sync::Semaphore; +use tokio::task::AbortHandle; +use tokio::time::{Instant, sleep}; +use tracing::{info, warn}; + +/// Dials that run at once, for one client. +const MAX_CONCURRENT_DIALS: usize = 4; + +/// The longest pause before a slot or a sweep tries again. +const RECONNECT_BACKOFF_MAX: Duration = Duration::from_secs(10); + +/// The reconnection configuration, the same for every transport. +#[derive(Debug, Clone, Copy)] +pub(crate) struct ReconnectPolicy { + /// Without it, a lost link stays down, and a sweep makes one pass. + pub(super) enabled: bool, + /// The attempts of a slot after its first, and the passes of a sweep + /// after its first. `None` means no limit. + pub(super) max_retries: Option<u32>, + /// The base of the backoff. + pub(super) interval: Duration, + /// A link that stayed up this long redials at once after a loss. + pub(super) reestablish_after: Duration, +} + +impl ReconnectPolicy { + fn allows(&self, retry: u32) -> bool { + self.enabled && self.max_retries.is_none_or(|max| retry <= max) + } + + fn pause(&self, retry: u32) -> Duration { + backoff(retry, self.interval, RECONNECT_BACKOFF_MAX, rand::random()) + } +} + +macro_rules! reconnect_policy_from { + ($($config:ty),+) => {$( + impl From<&$config> for ReconnectPolicy { + fn from(config: &$config) -> Self { + Self { + enabled: config.enabled, + max_retries: config.max_retries, + interval: config.interval.get_duration(), + reestablish_after: config.reestablish_after.get_duration(), + } + } + } + )+}; +} + +reconnect_policy_from!( + TcpClientReconnectionConfig, + QuicClientReconnectionConfig, + WebSocketClientReconnectionConfig +); + +/// Where a request goes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Target { + /// The metadata primary of the last roster, else the bootstrap link. + MetadataPrimary, + Replica(u8), +} + +/// The links of one client. Dropping it aborts every task and drops every +/// link, without a runtime. +#[derive(Debug)] +pub(crate) struct ReplicaLinks { + shared: Arc<LinksShared>, +} + +/// The part that the tasks hold. No task holds [`ReplicaLinks`], so its +/// `Drop` always runs. +#[derive(Debug)] +struct LinksShared { + dialer: Dialer, + configured_endpoint: String, + policy: ReconnectPolicy, + session: Arc<SessionOwner>, + /// Never held across an await. The epoch of the session is read before + /// this lock, never under it. + table: Mutex<Table>, + dial_permits: Semaphore, + coordinator: ConnectCoordinator, + events: Sender<DiagnosticEvent>, + /// Never read. Each subscriber is a clone, so it replays the backlog. + subscribers: Receiver<DiagnosticEvent>, +} + +#[derive(Debug)] +struct Table { + state: ClientState, + /// One slot per replica of the last roster. A replica id is a `u8`, so + /// the table holds at most 256 slots. + slots: BTreeMap<u8, Slot>, + bootstrap: Option<Bootstrap>, + metadata_primary: Option<u8>, + /// The endpoints of the last roster. A sweep dials them after the + /// configured endpoint. + roster: Vec<String>, +} + +#[derive(Debug)] +struct Slot { + endpoint: String, + link: Option<ReadyLink>, + /// `None` after the task stopped. Only the task of this handle changes + /// the slot. + task: Option<AbortHandle>, +} + +/// The link that the sweep signed in on, until a slot takes it over or the +/// metadata primary slot is ready. +#[derive(Debug)] +struct Bootstrap { + endpoint: String, + link: ReadyLink, + /// `None` while the sweep still uses the link. + watcher: Option<AbortHandle>, +} + +#[derive(Debug)] +struct ReadyLink { + link: Arc<Link>, + /// The fence of the session that the link carries, `None` before a + /// sign-in. + bound_fence: Option<u64>, +} + +/// How one endpoint of a sweep ended. +enum Attempt { + /// Another endpoint can still work. + EndpointFailed, + /// No retry can fix it, but the endpoints behind this one can still work. + ConfigFault(IggyError), + /// The sign-in lost its link. Another endpoint can still work. + SignInLost(IggyError), + /// The sweep ends with this error. + Failed(IggyError), +} + +#[derive(Debug)] +struct SignInFailure { + error: IggyError, + link_lost: bool, +} + +impl From<IggyError> for SignInFailure { + fn from(error: IggyError) -> Self { + Self { + link_lost: lost_link(&error), + error, + } + } +} + +impl ReplicaLinks { + pub(crate) fn new( + dialer: Dialer, + configured_endpoint: String, + policy: ReconnectPolicy, + session: Arc<SessionOwner>, + ) -> Self { + let (events, subscribers) = event_channel(); + let table = Table { + state: ClientState::Disconnected, + slots: BTreeMap::new(), + bootstrap: None, + metadata_primary: None, + roster: Vec::new(), + }; + Self { + shared: Arc::new(LinksShared { + dialer, + configured_endpoint, + policy, + session, + table: Mutex::new(table), + dial_permits: Semaphore::new(MAX_CONCURRENT_DIALS), + coordinator: ConnectCoordinator::new(), + events, + subscribers, + }), + } + } + + /// Sweeps the known endpoints until one of them signs in. Concurrent + /// callers share one result. On a connected client, every slot without a + /// link dials again at once. + pub(crate) async fn connect(&self) -> Result<(), IggyError> { + let shared = &self.shared; + shared + .coordinator + .run(|abandoned, _| shared.sweep(abandoned)) + .await + } + + /// Aborts every task and closes every link, which fails each pending + /// request. The roster stays for the next sweep. + pub(crate) fn disconnect(&self) { + self.shared.disconnect(); + } + + /// Ends every link the way its protocol expects. It is final. + pub(crate) async fn shutdown(&self) { + let links = { + let mut table = self.shared.table(); + if table.state == ClientState::Shutdown { + return; + } + table.state = ClientState::Shutdown; + table.clear() + }; + join_all(links.into_iter().map(Link::shutdown)).await; + self.shared.dialer.shutdown().await; + self.shared.publish(DiagnosticEvent::Shutdown); + } + + /// Pings every link that serves the session, all at once. A failed ping + /// only logs, because a link that lost its connection redials on its own. + pub(crate) async fn keepalive(&self) -> Result<(), IggyError> { + let links: Vec<Arc<Link>> = { + let epoch = self.shared.session.epoch(); + let table = self.shared.table(); + if table.state == ClientState::Shutdown { + return Err(IggyError::ClientShutdown); + } + table + .links() + .filter(|ready| serves(ready, epoch)) + .map(|ready| Arc::clone(&ready.link)) + .collect() + }; + let deadline = Instant::now() + OPERATION_TIMEOUT; + let pings = links.iter().map(|link| { + exchange( + &self.shared.session, + link, + PING_CODE, + PingRequest.to_bytes(), + deadline, + ) + }); + for result in join_all(pings).await { + if let Err(error) = result { + warn!("A keepalive ping failed: {error}"); + } + } + Ok(()) + } + + /// The link for `target`. A request never dials, so a target without a + /// ready link fails with `NotConnected`. + pub(crate) fn link(&self, target: Target) -> Result<Arc<Link>, IggyError> { + let epoch = self.shared.session.epoch(); + let table = self.shared.table(); + let slot = |id: u8| table.slots.get(&id).and_then(|slot| slot.link.as_ref()); + let ready = match target { + Target::Replica(id) => slot(id), + Target::MetadataPrimary => table + .metadata_primary + .and_then(slot) + .filter(|ready| serves(ready, epoch)) + .or(table.bootstrap.as_ref().map(|bootstrap| &bootstrap.link)), + }; + ready + .filter(|ready| serves(ready, epoch)) + .map(|ready| Arc::clone(&ready.link)) + .ok_or(IggyError::NotConnected) + } + + /// The replicas with a link that serves the session, in id order. + pub(crate) fn ready_replicas(&self) -> Vec<u8> { + let epoch = self.shared.session.epoch(); + self.shared + .table() + .slots + .iter() + .filter(|(_, slot)| slot.link.as_ref().is_some_and(|ready| serves(ready, epoch))) + .map(|(id, _)| *id) + .collect() + } + + pub(crate) fn metadata_primary(&self) -> Option<u8> { + self.shared.table().metadata_primary + } + + pub(crate) fn session(&self) -> &Arc<SessionOwner> { + &self.shared.session + } + + /// Records that `link` carries the session of `fence`. A sign-in calls it + /// before it binds the session, so the slots keep this link when they + /// close the links of the old session. + pub(crate) fn mark_bound(&self, link: &Arc<Link>, fence: u64) { + self.shared.mark_bound(link, fence); + } + + /// Reads the roster on `link` and rebuilds the slots from it. + pub(crate) async fn refresh_roster(&self, link: &Arc<Link>) -> Result<(), IggyError> { + self.shared.refresh_roster(link).await + } + + pub(crate) fn subscribe(&self) -> Receiver<DiagnosticEvent> { + self.shared.subscribers.clone() + } + + pub(crate) fn publish(&self, event: DiagnosticEvent) { + self.shared.publish(event); + } + + pub(crate) fn state(&self) -> ClientState { + self.shared.table().state + } + + /// Sets the state, except after a shutdown, which is final. + pub(crate) fn set_state(&self, state: ClientState) { + let mut table = self.shared.table(); + if table.state != ClientState::Shutdown { + table.state = state; + } + } + + /// The endpoint of the metadata primary, else of the bootstrap link, else + /// the configured one. + pub(crate) fn connection_address(&self) -> String { + let table = self.shared.table(); + table + .metadata_primary + .and_then(|id| table.slots.get(&id)) + .map(|slot| &slot.endpoint) + .or(table + .bootstrap + .as_ref() + .map(|bootstrap| &bootstrap.endpoint)) + .unwrap_or(&self.shared.configured_endpoint) + .clone() + } +} + +impl Drop for ReplicaLinks { + fn drop(&mut self) { + // An abort needs no runtime, and each dropped link stops its own + // tasks. + drop(self.shared.table().clear()); + } +} + +impl LinksShared { + fn table(&self) -> MutexGuard<'_, Table> { + self.table.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// The channel runs in overflow mode, so a publish never waits. + fn publish(&self, event: DiagnosticEvent) { + let _ = self.events.try_broadcast(event); + } + + fn disconnect(&self) { + let (links, previous) = { + let mut table = self.table(); + if table.state == ClientState::Shutdown { + return; + } + let links = table.clear(); + ( + links, + std::mem::replace(&mut table.state, ClientState::Disconnected), + ) + }; + links.iter().for_each(|link| link.close()); + if previous != ClientState::Disconnected { + self.publish(DiagnosticEvent::Disconnected); + } + } + + async fn sweep(self: &Arc<Self>, abandoned: bool) -> Result<(), IggyError> { + if abandoned { + // The last sweep lost its caller halfway. + self.disconnect(); + } + let (candidates, stopped) = { + let mut table = self.table(); + match table.state { + ClientState::Shutdown => return Err(IggyError::ClientShutdown), + ClientState::Connected + | ClientState::Authenticating + | ClientState::Authenticated => { + self.restart_down_slots(&mut table); + return Ok(()); + } + ClientState::Disconnected | ClientState::Connecting => {} + } + table.state = ClientState::Connecting; + (self.candidates(&table.roster), table.clear()) + }; + stopped.iter().for_each(|link| link.close()); + + let mut pass = 0; + loop { + let mut config_fault = None; + let mut sign_in_failure = None; + for endpoint in &candidates { + match self.try_endpoint(endpoint).await { + Ok(()) => return Ok(()), + Err(Attempt::EndpointFailed) => {} + Err(Attempt::ConfigFault(error)) => config_fault = Some(error), + Err(Attempt::SignInLost(error)) => sign_in_failure = Some(error), + Err(Attempt::Failed(error)) => return Err(error), + } + } + pass += 1; + let failure = match config_fault { + Some(error) => Some(error), + None if !self.policy.allows(pass) => { + Some(sign_in_failure.unwrap_or(IggyError::CannotEstablishConnection)) + } + None => None, + }; + if let Some(error) = failure { + self.fail_sweep(); + return Err(error); + } + let pause = self.policy.pause(pass); + info!("No endpoint signed in. Pass {pass} starts in {pause:?}."); + sleep(pause).await; + } + } + + /// The configured endpoint, then the endpoints of the last roster, each + /// spelling once. Spellings need no name lookup, which can stall a sweep. + fn candidates(&self, roster: &[String]) -> Vec<String> { + let mut candidates = vec![self.configured_endpoint.clone()]; + for endpoint in roster { + if !candidates + .iter() + .any(|known| is_same_spelling(known, endpoint)) + { + candidates.push(endpoint.clone()); + } + } + candidates + } + + async fn try_endpoint(self: &Arc<Self>, endpoint: &str) -> Result<(), Attempt> { + sweeping(self.table().state).map_err(Attempt::Failed)?; + info!("Connecting to {endpoint}..."); + let dialed = match self.dial_permits.acquire().await { + Ok(_permit) => self.dialer.dial(endpoint).await, + Err(_) => Err(IggyError::CannotEstablishConnection), + }; + let link = match dialed { + Ok(link) => Arc::new(link), + Err(IggyError::CannotEstablishConnection) => return Err(Attempt::EndpointFailed), + Err(error) => return Err(Attempt::ConfigFault(error)), + }; + if let Err(error) = self.install_bootstrap(endpoint, &link) { + link.close(); + return Err(Attempt::Failed(error)); + } + self.publish(DiagnosticEvent::Connected); + info!("Connected to {endpoint}."); + + if let Err(failure) = self.resume_or_sign_in(&link).await { + if failure.link_lost { + warn!( + "The sign-in on {endpoint} did not complete: {}", + failure.error + ); + self.lose_bootstrap(&link); + return Err(Attempt::SignInLost(failure.error)); + } + // The link stands, and only the session is missing. Every + // endpoint gives the same answer to the same credentials. + self.watch_bootstrap(&link); + return Err(Attempt::Failed(failure.error)); + } + if self.session.epoch().bound + && let Err(error) = self.refresh_roster(&link).await + { + warn!("Failed to read the roster from {endpoint}: {error}"); + } + self.watch_bootstrap(&link); + Ok(()) + } + + fn install_bootstrap(&self, endpoint: &str, link: &Arc<Link>) -> Result<(), IggyError> { + let mut table = self.table(); + sweeping(table.state)?; + table.bootstrap = Some(Bootstrap { + endpoint: endpoint.to_owned(), + link: ReadyLink { + link: Arc::clone(link), + bound_fence: None, + }, + watcher: None, + }); + table.state = ClientState::Connected; + Ok(()) + } + + /// Drops a bootstrap link that its sign-in lost. The sweep goes on. + fn lose_bootstrap(&self, link: &Arc<Link>) { + link.close(); + let mut table = self.table(); + if !table.is_bootstrap(link) { + return; + } + table.bootstrap = None; + table.state = ClientState::Connecting; + drop(table); + self.publish(DiagnosticEvent::Disconnected); + } + + fn fail_sweep(&self) { + let mut table = self.table(); + if table.state == ClientState::Connecting { + table.state = ClientState::Disconnected; + drop(table); + self.publish(DiagnosticEvent::Disconnected); + } + } + + /// Watches a bootstrap link that the sweep is done with, until it closes. + fn watch_bootstrap(self: &Arc<Self>, link: &Arc<Link>) { + let mut table = self.table(); + let Some(bootstrap) = table + .bootstrap + .as_mut() + .filter(|bootstrap| Arc::ptr_eq(&bootstrap.link.link, link)) + else { + return; + }; + let shared = Arc::clone(self); + let link = Arc::clone(link); + let watcher = tokio::spawn(async move { + shared.hold(&link).await; + shared.bootstrap_down(&link); + }); + bootstrap.watcher = Some(watcher.abort_handle()); + } + + fn bootstrap_down(&self, link: &Arc<Link>) { + let mut table = self.table(); + if !table.is_bootstrap(link) { + return; + } + table.bootstrap = None; + let last = table.ready_count() == 0; + table.settle(); + drop(table); + if last { + self.publish(DiagnosticEvent::Disconnected); + } + } + + /// Resumes a bound session on the bootstrap link, else signs in there. + async fn resume_or_sign_in(&self, link: &Arc<Link>) -> Result<(), SignInFailure> { + let epoch = self.session.epoch(); + if epoch.bound { + match self.bind_session(link).await { + Ok(()) => { + self.mark_bound(link, epoch.fence); + self.set_state_on(link, ClientState::Authenticated); + self.publish(DiagnosticEvent::SignedIn); + info!("Resumed the session."); + return Ok(()); + } + Err(error) if session_lost(&error) => { + info!("The session ended ({error}). Signing in again."); + self.session.reset_if_fence(epoch.fence)?; + } + // A replica that cannot bind inside its read budget is behind + // the session, and another one can be further ahead. + Err(error) => { + return Err(SignInFailure { + link_lost: lost_link(&error) + || matches!(error, IggyError::TransientNotAccepted), + error, + }); + } + } + } + self.sign_in(link).await + } + + /// Signs in on `link` with the known credentials, if any. A failure that + /// leaves the link up leaves the client connected without a session. A + /// rejected credential is forgotten, because it stays rejected, and each + /// replay costs the server a password hash. + async fn sign_in(&self, link: &Arc<Link>) -> Result<(), SignInFailure> { + let Some(credentials) = self.session.sign_in_credentials() else { + info!("No credentials to sign in with."); + return Ok(()); + }; + self.set_state_on(link, ClientState::Authenticating); + let error = match self.register(link, &credentials).await { + Ok(user_id) => { + self.set_state_on(link, ClientState::Authenticated); + self.session.remember(credentials, user_id); + self.publish(DiagnosticEvent::SignedIn); + info!("Signed in."); + return Ok(()); + } + Err(error) => error, + }; + let failure = SignInFailure::from(error); + if !failure.link_lost { + self.set_state_on(link, ClientState::Connected); + } + if matches!( + failure.error, + IggyError::InvalidCredentials + | IggyError::InvalidUsername + | IggyError::InvalidPassword + | IggyError::Unauthenticated + ) { + self.session.forget(); + } + Err(failure) + } + + /// Registers a new session on `link` and binds it. A Register that the + /// server did not commit or did not admit goes again with the same client + /// id, so it registers one session. + async fn register( + &self, + link: &Arc<Link>, + credentials: &Credentials, + ) -> Result<u32, IggyError> { + let fence = self.session.fence(); + let (code, body) = sign_in_request( + credentials, + rust_sdk_version_info(SDK_VERSION)?, + self.session.bind_secret()?, + )?; + let deadline = Instant::now() + OPERATION_TIMEOUT; + let mut retry = 0_u32; + let reply = loop { + let error = match exchange(&self.session, link, code, body.clone(), deadline).await { + Ok(reply) => break reply, + Err(error) => error, + }; + if !matches!( + error, + IggyError::TransientNotCommitted | IggyError::TransientNotAccepted + ) || !pause_before_retry(retry, deadline, rand::random()).await + { + return Err(error); + } + retry = retry.saturating_add(1); + }; + let registered = + LoginRegisterResponse::decode_from(&reply).map_err(|_| IggyError::InvalidFormat)?; + self.mark_bound(link, fence); + self.session.bind_at(fence, registered.session)?; + Ok(registered.user_id) + } + + /// Binds the bound session on `link`. The reply must name the same + /// session. + async fn bind_session(&self, link: &Link) -> Result<(), IggyError> { + let identity = self.session.identity()?; + let request = BindSessionRequest { + version_info: rust_sdk_version_info(SDK_VERSION)?, + identity, + bind_secret: self.session.bind_secret()?, + }; + let reply = exchange( + &self.session, + link, + BIND_SESSION_CODE, + request.to_bytes(), + Instant::now() + OPERATION_TIMEOUT, + ) + .await?; + decode_session_binding(&identity, &reply).map(drop) + } + + async fn refresh_roster(self: &Arc<Self>, link: &Arc<Link>) -> Result<(), IggyError> { + let reply = exchange( + &self.session, + link, + GET_CLUSTER_METADATA_CODE, + GetClusterMetadataRequest.to_bytes(), + Instant::now() + OPERATION_TIMEOUT, + ) + .await?; + let roster = + ClusterMetadataResponse::decode_from(&reply).map_err(|_| IggyError::InvalidFormat)?; + self.apply_roster(&ClusterMetadata::try_from(roster)?, link) + .await + } + + /// Rebuilds the slots from the roster that `via` carried. A removed + /// replica or a moved endpoint retires its link, so the requests that + /// hold it finish first. + async fn apply_roster( + self: &Arc<Self>, + roster: &ClusterMetadata, + via: &Arc<Link>, + ) -> Result<(), IggyError> { + let protocol = self.dialer.protocol(); + let mut ids = BTreeSet::new(); + let mut wanted = BTreeMap::new(); + for node in &roster.nodes { + if !ids.insert(node.replica_id) { + warn!( + "The roster names replica {} twice, so the links stay as they are.", + node.replica_id + ); + return Err(IggyError::InvalidFormat); + } + // A node without a port for this transport gets no slot, so the + // client never dials `ip:0`. + let port = transport_port(node, protocol); + if port != 0 { + wanted.insert(node.replica_id, node_address(node, port)); + } + } + let primary = roster + .nodes + .iter() + .find(|node| node.role == ClusterNodeRole::Leader) + .map(|node| node.replica_id) + .filter(|id| wanted.contains_key(id)); + let adopt = self.bootstrap_replica(&wanted).await; + + let mut retired = Vec::new(); + let mut table = self.table(); + if table.find(via).is_none() { + // A disconnect ran during the read. + return Err(IggyError::Disconnected); + } + table.slots.retain(|id, slot| { + let kept = wanted.get(id) == Some(&slot.endpoint); + if !kept { + if let Some(task) = slot.task.take() { + task.abort(); + } + retired.extend(slot.link.take()); + } + kept + }); + for (id, endpoint) in &wanted { + if table.slots.contains_key(id) { + continue; + } + let adopted = match &adopt { + Some((link, replica)) if replica == id && table.is_bootstrap(link) => { + table.retire_bootstrap() + } + _ => None, + }; + let task = self.spawn_slot( + *id, + endpoint.clone(), + adopted.as_ref().map(|ready| Arc::clone(&ready.link)), + ); + table.slots.insert( + *id, + Slot { + endpoint: endpoint.clone(), + link: adopted, + task: Some(task), + }, + ); + } + table.metadata_primary = primary; + table.roster = wanted.into_values().collect(); + if primary.is_some_and(|id| table.slots.get(&id).is_some_and(|slot| slot.link.is_some())) { + retired.extend(table.retire_bootstrap()); + } + drop(table); + drop(retired); + Ok(()) + } + + /// The replica that the bootstrap link reaches. With one replica in the + /// roster, that replica: a single node advertises its bind address, and + /// that can be `0.0.0.0`. + async fn bootstrap_replica(&self, wanted: &BTreeMap<u8, String>) -> Option<(Arc<Link>, u8)> { + let (link, endpoint) = { + let table = self.table(); + let bootstrap = table.bootstrap.as_ref()?; + (Arc::clone(&bootstrap.link.link), bootstrap.endpoint.clone()) + }; + if wanted.len() == 1 { + return wanted.keys().next().map(|id| (link, *id)); + } + for (id, replica) in wanted { + if is_same_address(&endpoint, replica).await { + return Some((link, *id)); + } + } + None + } + + fn spawn_slot( + self: &Arc<Self>, + id: u8, + endpoint: String, + adopted: Option<Arc<Link>>, + ) -> AbortHandle { + tokio::spawn(Arc::clone(self).run_slot(id, endpoint, adopted)).abort_handle() + } + + /// Starts every slot without a link again, at once and without its + /// backoff. + fn restart_down_slots(self: &Arc<Self>, table: &mut Table) { + for (id, slot) in table + .slots + .iter_mut() + .filter(|(_, slot)| slot.link.is_none()) + { + if let Some(task) = slot.task.take() { + task.abort(); + } + slot.task = Some(self.spawn_slot(*id, slot.endpoint.clone(), None)); + } + } + + /// Keeps one link to replica `id` up. Each retry waits its backoff, and a + /// link that stayed up for `reestablish_after` redials at once. + async fn run_slot(self: Arc<Self>, id: u8, endpoint: String, mut adopted: Option<Arc<Link>>) { + let mut signs_in_at = None; + let mut retry = 0_u32; + loop { + let link = match adopted.take() { + Some(link) => link, + None => { + if retry > 0 { + if !self.policy.allows(retry) { + break; + } + sleep(self.policy.pause(retry)).await; + } + match self.open(id, &endpoint, &mut signs_in_at).await { + Some(link) => link, + None => { + retry = retry.saturating_add(1); + continue; + } + } + } + }; + let up_since = Instant::now(); + self.hold(&link).await; + self.slot_down(id, &link); + if !self.policy.enabled { + break; + } + retry = if up_since.elapsed() >= self.policy.reestablish_after { Review Comment: Deferred to multi-leader PR5 connection/failover work. Apply jitter to the first redial after a live link drops and test simultaneous reconnects while preserving the operation deadline and bounded replica-link lifecycle. ########## core/sdk/src/driver.rs: ########## @@ -0,0 +1,1309 @@ +// 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. + +//! The operation driver sends every request of a binary client. One deadline +//! bounds the lane wait, the lookups, the attempts and the pauses of a call. +//! A frame that can have taken effect is never frozen again, so a retry never +//! applies a request twice. + +// Unused until the clients send through the driver. +#![allow(dead_code)] + +use crate::links::{ + Frame, Link, OPERATION_TIMEOUT, ReplicaLinks, Target, lost_link, pause_before_retry, read_reply, +}; +use crate::session_owner::{LaneKey, LaneTicket, RetainedOperation, SessionOwner}; +use bytes::Bytes; +use iggy_binary_protocol::codes::{ + BIND_SESSION_CODE, GET_PARTITION_ROUTE_CODE, LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, + SEND_MESSAGES_CODE, +}; +use iggy_binary_protocol::consensus::{Operation, operation_for_code}; +use iggy_binary_protocol::primitives::partition_history::PartitionContext; +use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest; +use iggy_binary_protocol::responses::messages::PartitionRouteResponse; +use iggy_binary_protocol::responses::users::LoginRegisterResponse; +use iggy_binary_protocol::{WireDecode, WireEncode}; +use iggy_common::{ClientState, IggyError, RequestTarget}; +use route_cache::RouteCache; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; +use tokio::time::{Instant, timeout_at}; +use tracing::debug; + +mod poll_router; +mod route_cache; + +pub(crate) use poll_router::{ + PollRouter, PollTransport, ROSTER_READ_TIMEOUT, is_poll_routing_code, +}; +pub(crate) use route_cache::PartitionKey; + +/// Sends the requests of one client over its replica links. +pub(crate) struct OperationDriver { + session: Arc<SessionOwner>, + links: ReplicaLinks, + routes: Mutex<RouteCache>, +} + +impl OperationDriver { + /// The driver shares the session owner of `links`. + pub(crate) fn new(links: ReplicaLinks) -> Self { + Self { + session: Arc::clone(links.session()), + links, + routes: Mutex::default(), + } + } + + /// Sends one request and returns the body of its reply. + pub(crate) async fn send( + &self, + code: u32, + payload: Bytes, + target: RequestTarget<'_>, + ) -> Result<Bytes, IggyError> { + let deadline = Instant::now() + OPERATION_TIMEOUT; + self.drive(Call::new(code, payload, target, deadline)).await + } + + /// Replays the retained operation `op` until it has an outcome or + /// `deadline` passes. Its success frees the lane. + pub(crate) async fn resolve(&self, op: u64, deadline: Instant) -> Result<(), IggyError> { + let (ticket, retained) = self.session.resume(op, deadline).await?; + self.drive(Call::resumed(ticket, retained, deadline)) + .await + .map(|_| ()) + } + + /// Frees the lane of the retained operation `op`. Its outcome stays + /// unknown. + pub(crate) fn abandon(&self, op: u64) -> Result<(), IggyError> { + self.session.abandon(op) + } + + async fn drive(&self, mut call: Call) -> Result<Bytes, IggyError> { + loop { + let next = match self.attempt(&mut call).await { + Ok(body) => { + call.complete(); + return Ok(body); + } + Err(next) => next, + }; + call.settle_sent(); + match next { + Next::Again if Instant::now() < call.deadline => {} + Next::Again => return Err(call.expire()), + Next::Pause => { + if !call.pause().await { + return Err(call.expire()); + } + } + Next::Fail(error) => return Err(call.end(error)), + } + } + } + + /// One attempt of `call`, from the choice of its replica to its reply. + async fn attempt(&self, call: &mut Call) -> Result<Bytes, Next> { + if let Some(error) = self.stopped() { + return Err(Next::Fail(error)); + } + if call.frame.is_some() && call.fence() != Some(self.session.fence()) { + // A replaced session can never apply the frame. An uncertain call + // ends with its uncertainty, any other call freezes again. + if call.uncertain { + return Err(Next::Fail(IggyError::Unauthenticated)); + } + call.frame = None; + } + if call.frame.is_none() { + self.wait_for_session(call).await?; + } + let route = self.route_for(call).await?; + let context = self.context(call, route.as_ref()); + // The lane comes first, so a blocked lane ends the call at once, also + // while no link is ready. + let ticket = self.ticket(call, &context).await?; + let hop = match self.hop(call, route).await { + Ok(hop) => hop, + Err(next) => { + call.ticket = Some(ticket); + return Err(next); + } + }; + let deadline = call.deadline; + let (frame, ticket) = call.frame(ticket, context)?; + let result = hop.link.send(frame, deadline, ticket.sent_flag()).await; + let sent = *ticket.sent_flag(); + let result = result.and_then(|reply| read_reply(&self.session, reply)); + self.settle(call, &hop, result, sent).await + } + + /// The error of a call on a client that nothing brings back. + fn stopped(&self) -> Option<IggyError> { + match self.links.state() { + ClientState::Shutdown => Some(IggyError::ClientShutdown), + ClientState::Disconnected => Some(IggyError::NotConnected), + _ => None, + } + } + + /// Waits for a bound session while the links sign in again. Without + /// credentials nothing signs in, so the call goes on unbound. + async fn wait_for_session(&self, call: &Call) -> Result<(), Next> { + if is_sign_in(call.code) + || self.session.epoch().bound + || self.session.sign_in_credentials().is_none() + { + return Ok(()); + } + let mut epoch = self.session.subscribe_epoch(); + match timeout_at(call.deadline, epoch.wait_for(|epoch| epoch.bound)).await { + Ok(Ok(_)) => Ok(()), + Ok(Err(_)) | Err(_) => Err(Next::Fail(IggyError::Unauthenticated)), + } + } + + /// The route of a partition call, `None` for any other call. A partition + /// call looks up its route unless the cached one is usable. + async fn route_for(&self, call: &mut Call) -> Result<Option<PartitionRouteResponse>, Next> { + let Some(partition) = &call.partition else { + return Ok(None); + }; + match self.route(&partition.key, call.deadline).await { + Ok(route) => Ok(Some(route)), + Err(error) if retries_lookup(&error) => { + call.refusal = error; + Err(Next::Pause) + } + Err(error) => Err(Next::Fail(error)), + } + } + + /// The context that a new frame of `call` carries. Every attempt and + /// every replay keeps the context of the first frame. + fn context(&self, call: &Call, route: Option<&PartitionRouteResponse>) -> PartitionContext { + match (&call.partition, route) { + (Some(partition), Some(route)) => partition.context.unwrap_or_else(|| route.context()), + _ if operation_for_code(call.code) == Operation::NonReplicated => self.read_context(), + _ => PartitionContext::default(), + } + } + + /// A read without a partition context waits until its replica applied + /// the metadata that the session saw. + fn read_context(&self) -> PartitionContext { + PartitionContext { + metadata_op: self.session.metadata_watermark(), + ..PartitionContext::default() + } + } + + /// The lane ticket of `call`. The call enters its lane on its first + /// attempt and keeps the ticket until it ends. + async fn ticket( + &self, + call: &mut Call, + context: &PartitionContext, + ) -> Result<LaneTicket, Next> { + match call.ticket.take() { + Some(ticket) => Ok(ticket), + None => { + let key = LaneKey::of(call.code, context, call.partition_id()); + self.session + .enter_lane(key, call.deadline) + .await + .map_err(Next::Fail) + } + } + } + + /// The link of the next attempt: to the primary of the route, else to + /// the metadata primary. + async fn hop( + &self, + call: &mut Call, + route: Option<PartitionRouteResponse>, + ) -> Result<Hop, Next> { + let target = route.map_or(Target::MetadataPrimary, |route| { + Target::Replica(route.primary_replica) + }); + let error = match self.links.link(target) { + Ok(link) => return Ok(Hop { link, route }), + Err(error) => error, + }; + call.refusal = error; + match (&call.partition, route) { + (Some(partition), Some(route)) => { + self.routes() + .failed(&route, partition.key.partition_id, route.primary_replica); + } + _ => self.find_metadata_primary(call.deadline).await, + } + Err(Next::Pause) + } + + /// Reads the outcome of one attempt. `sent` tells that this or an + /// earlier attempt can have taken effect. + async fn settle( + &self, + call: &mut Call, + hop: &Hop, + result: Result<Bytes, IggyError>, + sent: bool, + ) -> Result<Bytes, Next> { + let error = match result { + Ok(body) => return self.signed_in(call, hop, body), + Err(error) => error, + }; + let repeated = std::mem::replace( + &mut call.not_primary, + matches!(error, IggyError::NotPrimary), + ); + match error { + IggyError::TransientNotCommitted => { + call.uncertain = true; + Err(Next::Pause) + } + IggyError::TransientNotAccepted => { + call.refusal = error; + self.refresh(call, hop, Failure::Refused).await; + Err(Next::Pause) + } + IggyError::NotPrimary => { + // Neither was admitted, so the deadline reports both as 58. + call.refusal = IggyError::TransientNotAccepted; + self.refresh(call, hop, Failure::NotPrimary).await; + Err(if repeated { Next::Pause } else { Next::Again }) + } + error if lost_link(&error) => { + call.uncertain |= sent; + call.refusal = error; + self.refresh(call, hop, Failure::LostLink).await; + Err(Next::Pause) + } + // The refusal spent the request id, so the call starts a new + // operation in the same lane. + IggyError::LifecycleBusy if !call.uncertain => { + call.refusal = error; + call.frame = None; + Err(Next::Pause) + } + IggyError::HistoryUnavailable => { + self.invalidate(call, hop); + if call.restarts_on_new_history() { + call.restart(); + return Err(Next::Again); + } + Err(Next::Fail(error)) + } + IggyError::ConsumerGroupPartitionNotOwned(..) | IggyError::PartitionNotFound(..) => { + self.invalidate(call, hop); + Err(Next::Fail(error)) + } + error => Err(Next::Fail(error)), Review Comment: Tracked as a blocking prerequisite for multi-leader PR5 driver activation. The wire must distinguish a committed business rejection from a pre-consensus denial. Before activation, test an uncertain request replayed to a committed rejection: resolve must terminate and release its lane without treating an uncommitted denial as a settled outcome. ########## core/sdk/src/session_owner/lanes.rs: ########## @@ -0,0 +1,605 @@ +// 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. + +//! A lane admits one mutation at a time, in arrival order. There is one lane +//! for metadata and one for each partition history. A mutation whose outcome +//! is unknown stays retained in its lane, and the lane admits nothing else +//! until a resolve or an abandon frees it. So a retained operation is always +//! the newest operation of its lane, and its replay can never age out. + +use super::SessionOwner; +use crate::driver::PartitionKey; +use crate::links::Frame; +use crate::vsr::encode_request_header; +use bytes::Bytes; +use iggy_binary_protocol::consensus::{Operation, operation_for_code}; +use iggy_binary_protocol::primitives::partition_history::PartitionContext; +use iggy_common::IggyError; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, OwnedMutexGuard}; +use tokio::time::{Instant, timeout_at}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum LaneKey { + Metadata, + /// The incarnation names one history of the partition for every spelling + /// of its stream and topic. + Partition { + incarnation: u64, + partition_id: u32, + }, +} + +impl LaneKey { + /// The lane of `code`. A write to a partition takes the lane of the + /// incarnation that its context carries. Reads, Register and Logout take + /// no lane. Review Comment: PR4 documents the auto-commit exemption in e247f1ef3 and fixes server-side consumer-progress ordering in 5a2e85e54. Client-side lane semantics for auto-commit polls are deferred to multi-leader PR5 driver activation, with poll/store/delete ordering and cancellation tests required. ########## core/sdk/src/session_owner/lanes.rs: ########## @@ -0,0 +1,605 @@ +// 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. + +//! A lane admits one mutation at a time, in arrival order. There is one lane +//! for metadata and one for each partition history. A mutation whose outcome +//! is unknown stays retained in its lane, and the lane admits nothing else +//! until a resolve or an abandon frees it. So a retained operation is always +//! the newest operation of its lane, and its replay can never age out. + +use super::SessionOwner; +use crate::driver::PartitionKey; +use crate::links::Frame; +use crate::vsr::encode_request_header; +use bytes::Bytes; +use iggy_binary_protocol::consensus::{Operation, operation_for_code}; +use iggy_binary_protocol::primitives::partition_history::PartitionContext; +use iggy_common::IggyError; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, OwnedMutexGuard}; +use tokio::time::{Instant, timeout_at}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum LaneKey { + Metadata, + /// The incarnation names one history of the partition for every spelling + /// of its stream and topic. + Partition { + incarnation: u64, + partition_id: u32, + }, +} + +impl LaneKey { + /// The lane of `code`. A write to a partition takes the lane of the + /// incarnation that its context carries. Reads, Register and Logout take + /// no lane. + pub(crate) fn of(code: u32, context: &PartitionContext, partition_id: u32) -> Option<Self> { + match operation_for_code(code) { + Operation::NonReplicated | Operation::Register | Operation::Logout => None, + operation if operation.is_partition() => Some(Self::Partition { + incarnation: context.incarnation, + partition_id, + }), + _ => Some(Self::Metadata), + } + } +} + +pub(super) type Lanes = HashMap<LaneKey, Lane>; + +#[derive(Debug, Default)] +pub(super) struct Lane { + gate: Arc<Mutex<()>>, + /// The tickets that wait for the gate, hold it or resume. The lane leaves + /// the table when none is left and nothing is retained. + tickets: usize, + retained: Option<RetainedOperation>, +} + +impl Lane { + fn holds(&self, op: u64) -> bool { + self.retained + .as_ref() + .is_some_and(|retained| retained.op() == op) + } +} + +/// A frozen mutation. Every attempt and every replay sends the same frame. +#[derive(Debug, Clone)] +pub(crate) struct RetainedOperation { + pub(crate) code: u32, + pub(crate) frame: Frame, + /// The partition of a partition write, `None` for any other request. A + /// replay looks up its route by this key. + pub(crate) partition: Option<PartitionKey>, +} + +impl RetainedOperation { + pub(crate) fn op(&self) -> u64 { + self.frame.header.request + } +} + +/// The right of one call to send in its lane. Dropping the ticket frees the +/// lane, except that a frame which left the client without a definitive +/// outcome stays retained. +#[derive(Debug)] +pub(crate) struct LaneTicket { + owner: Arc<SessionOwner>, + /// `None` for a call that takes no lane. + key: Option<LaneKey>, + guard: Option<OwnedMutexGuard<()>>, + operation: Option<RetainedOperation>, + fence: u64, + /// A resume replays an operation that its lane still retains. + resumed: bool, + sent: bool, + done: bool, +} + +impl SessionOwner { + /// Waits for the lane of `key` in arrival order. A lane that retains an + /// operation fails at once with `LaneBlocked`, and a lane still busy at + /// `deadline` fails with `TransientNotAccepted`. Neither failure takes a + /// request id. + pub(crate) async fn enter_lane( + self: &Arc<Self>, + key: Option<LaneKey>, + deadline: Instant, + ) -> Result<LaneTicket, IggyError> { + let Some(key) = key else { + return Ok(LaneTicket::new(Arc::clone(self), None, false)); + }; + let (mut ticket, gate) = { + let mut lanes = self.lanes(); + let lane = lanes.entry(key).or_default(); Review Comment: Deferred to multi-leader PR6 after driver activation. Profile uncontended lane allocations and route-cache insertion at capacity before choosing a change; retain bounded cache growth and correct lane cleanup on cancellation. ########## core/sdk/src/driver/route_cache.rs: ########## @@ -0,0 +1,379 @@ +// 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. + +//! The routes of one client. A route has no age limit. It goes stale after +//! a refusal or a failure, and a stale route keeps its view as the lowest +//! view that its next lookup accepts. So a lookup never goes back to an +//! older primary of the same partition history. + +use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest; +use iggy_binary_protocol::responses::messages::PartitionRouteResponse; +use iggy_common::wire_conversions::identifier_to_wire; +use iggy_common::{Identifier, IggyError, PartitionTarget}; +use std::collections::HashMap; + +/// The most routes that one client keeps. A new route beyond it evicts the +/// oldest lookup. +pub(crate) const MAX_CACHED_ROUTES: usize = 4096; + +/// A partition as its caller spelled it. A stream or topic named by number +/// and the same one named by name are two keys. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct PartitionKey { + pub(crate) stream_id: Identifier, + pub(crate) topic_id: Identifier, + pub(crate) partition_id: u32, +} + +impl From<&PartitionTarget<'_>> for PartitionKey { + fn from(target: &PartitionTarget<'_>) -> Self { + Self { + stream_id: target.stream_id.clone(), + topic_id: target.topic_id.clone(), + partition_id: target.partition_id, + } + } +} + +impl TryFrom<&PartitionKey> for GetPartitionRouteRequest { + type Error = IggyError; + + fn try_from(key: &PartitionKey) -> Result<Self, Self::Error> { + Ok(Self { + stream_id: identifier_to_wire(&key.stream_id)?, + topic_id: identifier_to_wire(&key.topic_id)?, + partition_id: key.partition_id, + }) + } +} + +#[derive(Debug)] +struct Entry { + route: PartitionRouteResponse, + /// The session fence at the lookup. A route serves only that session. + fence: u64, + stale: bool, + /// The lowest view that the next lookup accepts. + floor: u32, + /// The replica that the next lookup asks first. The ids after it follow. + ask_first: Option<u8>, + /// Orders the entries for eviction. + lookup: u64, +} + +#[derive(Debug, Default)] +pub(crate) struct RouteCache { + entries: HashMap<PartitionKey, Entry>, + lookups: u64, +} + +impl RouteCache { + /// The route of `key`, unless it is stale, from another session, or + /// older than the metadata `watermark` that the session saw. + pub(crate) fn usable( + &self, + key: &PartitionKey, + fence: u64, + watermark: u64, + ) -> Option<PartitionRouteResponse> { + self.entries Review Comment: Deferred to multi-leader PR6. The global watermark remains the safety fence in PR4. Topic-specific invalidation requires trustworthy metadata-change provenance and tests for deletion/recreation and missed updates before it can replace that fence. ########## core/shard/src/lib.rs: ########## @@ -4511,17 +4581,38 @@ where } } + fn partition_incarnation_matches( + &self, + partition: &IggyPartition<B, SB>, + partition_incarnation: u64, + frame: &'static str, + ) -> bool { + if partition_incarnation == partition.created_revision() { + return true; + } + tracing::debug!( Review Comment: Deferred to multi-leader PR6 operational diagnostics: add rate-limited incarnation-mismatch visibility, including incarnation zero, with bounded metric cardinality and log volume. ########## core/consensus/src/plane_helpers.rs: ########## @@ -903,6 +903,62 @@ where .await } +/// Find a held entry below a view's suffix that the new history does not prove. +/// +/// `canonical` is ordered by descending op. Entries above the local commit floor +/// survive only if their parent chain reaches its lowest header, or a prepare's +/// commit stamp proves them through the uninterrupted local chain. This runs once +/// per installed log, before repair starts filling any resulting gap. +#[must_use] +pub fn first_unproven_op( + commit_floor: u64, + canonical: &[PrepareHeader], + op_head: u64, + last_op: Option<u64>, + header_at: impl Fn(u64) -> Option<PrepareHeader>, +) -> Option<u64> { + let last_op = last_op?; + let proven = locally_proven_commit(commit_floor, &header_at); + let (mut op, mut parent) = canonical.last().map_or((op_head, None), |lowest| { + (lowest.op.saturating_sub(1), Some(lowest.parent)) + }); + while op > proven { + match header_at(op) { + Some(local) + if local.checksum != CHECKSUM_UNSEALED && Some(local.checksum) == parent => + { + parent = Some(local.parent); + op -= 1; + } + _ => break, + } + } + (proven.checked_add(1)?..=op.min(last_op)).find(|&held| header_at(held).is_some()) Review Comment: Deferred to multi-leader PR6. Measure an ordered resident-index range scan during lagging-replica view installation, preserving the existing gap and repair-anchor checks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
