numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4260687487
##########
core/server/src/http/forward.rs:
##########
@@ -332,92 +537,61 @@ fn record_relayed_floor(state: &HttpInner, user_id: u32,
response: &Response) {
state.metadata_watermarks.record(user_id, applied);
}
-async fn forward_partition_or_pass(state: HttpState, request: Request, next:
Next) -> Response {
- if !state.forward.active ||
request.headers().contains_key(FORWARDED_HEADER) {
- return next.run(request).await;
- }
- // A read that does not move consumer progress is answered by whichever
- // replica the caller reached, which is the point of reading from one. Only
- // an automatic commit needs a replica that can originate the offset
- // operation, so only that poll pays for the buffering and the credential
- // resolution below.
- if request.method() == Method::GET &&
!wants_auto_commit(request.uri().query()) {
- return next.run(request).await;
- }
+async fn attach_partition_request(state: HttpState, request: Request, next:
Next) -> Response {
+ let received = Instant::now();
+ let hop = match partition_hop(request.headers()) {
+ Ok(hop) => hop,
+ Err(error) => return CustomError::from(error).into_response(),
+ };
+ // Verify-only, like the metadata relay: an unauthenticated caller gets no
+ // buffered body and no relay. The handler authenticates again.
let bearer = match bearer_token(request.headers()) {
Ok(bearer) => bearer,
Err(error) => return CustomError::from(error).into_response(),
};
if let Err(rejection) = resolve_credential(&state, bearer).await {
return rejection.into_response();
}
-
- let (parts, request_body) = request.into_parts();
+ let (mut parts, request_body) = request.into_parts();
let Ok(body) = to_bytes(request_body, state.forward.body_limit).await else
{
Review Comment:
nit: every partition request decodes the credential and buffers the whole
body here before the handler, which authenticates again. For an unauthenticated
caller an oversized body becomes 413 instead of 401 and is buffered first. A
cheap bearer-presence check before buffering, and skipping this when the
request is served locally, would help.
##########
core/server/src/http/forward.rs:
##########
@@ -42,24 +42,32 @@
//! - The primary re-authenticates and re-authorizes the forwarded request
//! through its ordinary extractor stack; the forward marker header is a loop
//! guard only and never a trust input.
-//! - A forwarded request is retried only when it provably never entered the
-//! primary's pipeline: a connect-phase failure, a 503 whose body carries the
-//! `TransientNotAccepted` code, or a 307 from a stale target. Every other
-//! outcome - including a 503 carrying `TransientNotCommitted`, whose op may
-//! still commit - is relayed as-is, because re-issuing it under a fresh
+//! - The route header, like the marker, carries only values a caller can send
+//! itself, so the hop target authenticates and authorizes the request again.
+//! - A forwarded write is retried only when it provably never entered the
+//! primary's pipeline: a connect-phase failure, and for a metadata relay a
+//! 503 whose body carries the `TransientNotAccepted` or `NotPrimary` code or
+//! a 307 from a stale target, for a partition relay a 503 carrying
+//! `NotPrimary`. Every
Review Comment:
nit: broken wrap, "`NotPrimary`. Every" sits on its own line.
##########
core/server/src/http/forward.rs:
##########
@@ -494,18 +693,15 @@ async fn attempt(
return AttemptOutcome::Relay(bad_gateway());
}
};
- let request = builder
- .headers(forwarded_headers(request_headers))
- .body(body.clone())
- .build();
+ let request = builder.headers(headers).body(body.clone()).build();
let attempt = async {
let response = match state.forward.client.execute(request).await {
Ok(response) => response,
- Err(error) => return classify_transport_error(&error),
+ Err(error) => return classify_transport_error(&error, kind),
};
- classify_forwarded_reply(response, method, retry_redirect).await
+ classify_forwarded_reply(response, method, kind).await
};
- match compio::time::timeout(FORWARD_ATTEMPT_TIMEOUT, attempt).await {
+ match compio::time::timeout(timeout, attempt).await {
Review Comment:
nit: an elapsed attempt answers 504 "outcome unknown" for every relay kind,
including `PartitionEffect::Read`, which changes nothing. A timed-out read
relay can answer a retryable 503 instead.
##########
core/server/src/http/handlers.rs:
##########
@@ -1256,25 +1253,27 @@ pub(in crate::http) async fn delete_segments(
/// [`poll_wire_request`]) since HTTP has no persistent connection and cannot
observe
/// group's partition assignment (that can be subject to change).
///
-/// A non-replicated read served in band: the same resolution the TCP dispatch
-/// runs ([`resolve_poll_request`]), then a mesh read on the owning shard and a
-/// re-encode of the stored batches into the legacy wire body, decoded here by
-/// the SDK's own [`PolledMessages::from_bytes`] so the JSON is field-identical
-/// to a TCP poll. `auto_commit` rides [`resolve_poll_request`]'s args and is
-/// honored by the owning shard's poll plan; no extra HTTP work.
+/// A non-replicated read served by the partition primary: the same resolution
+/// the TCP dispatch runs ([`resolve_poll_request`]), then a mesh read on the
+/// owning shard and a re-encode of the stored batches into the legacy wire
+/// body, decoded here by the SDK's own [`PolledMessages::from_bytes`] so the
+/// JSON is field-identical to a TCP poll. `auto_commit` rides
+/// [`resolve_poll_request`]'s args and is honored by the owning shard's poll
+/// plan. Another node relays the poll to the primary ([`route_partition`]).
pub(in crate::http) async fn poll_messages(
State(state): State<HttpState>,
identity: Identity,
Path((stream_id, topic_id)): Path<(String, String)>,
Query(query): Query<PollMessages>,
Query(consistency): Query<ConsistencyQuery>,
-) -> Result<Json<PolledMessages>, ReadError> {
+ Extension(request): Extension<Arc<PartitionRequest>>,
+) -> Result<Response, ReadError> {
let stream_id =
Identifier::from_str_value(&stream_id).map_err(ReadError::Rejected)?;
let topic_id =
Identifier::from_str_value(&topic_id).map_err(ReadError::Rejected)?;
SendWrapper::new(gate_local_read(
Review Comment:
nit: with `?consistency=linearizable`, a node that is neither the metadata
nor the partition primary first 307s to the metadata primary, which then relays
to the partition primary again: two hops where the partition relay alone would
do.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1707,115 +1589,65 @@ impl TcpClient {
Err(error) => return (preencoded.header, Err(error),
outcome_so_far),
}
};
- let mut frame_complete = false;
- let outcome = async {
- loop {
- let header_bytes = bytemuck::bytes_of(&request_header);
- frame_complete = false;
- stream.write(header_bytes).await?;
- if !payload.is_empty() {
- stream.write(&payload).await?;
+ let mut link_failed = false;
+ let mut replay = false;
+ let outcome = loop {
+ let frame = Frame {
+ header: request_header,
+ body: payload.clone(),
+ };
+ trace!("Sent a TCP request with code: {code}, waiting for a
response...");
+ // The deadline guards against server-side reply loss (e.g. a
+ // stalled replication quorum that never commits the op), which
+ // would hold the link lock forever and wedge every later
+ // request on this client.
+ let reply = match link.send(frame, read_deadline, &mut
false).await {
+ Ok(reply) => reply,
+ Err(error) => {
+ link_failed = true;
+ break Err(outcome_so_far.observe_link_error(error,
replay));
}
- stream.flush().await?;
- trace!("Sent a TCP request with code: {code}, waiting for
a response...");
-
- let mut response_header = [0u8;
iggy_binary_protocol::HEADER_SIZE];
- // `stream.read` delegates to `read_exact`; on success it
- // always returns the requested length, so no short-read
- // guard is needed here.
- //
- // Deadline guards against server-side reply loss (e.g. a
- // stalled replication quorum that never commits the op):
- // the connection is lockstep, so an unanswered read would
- // hold the stream lock forever and wedge every later
- // request on this client. On expiry drop the stream --
- // a late reply would desync framing for the next request.
- //
- // One deadline spans BOTH the header and body reads: a
- // reply that delivers a header then stalls must not get a
- // fresh full timeout for the body.
- let header_read =
- tokio::time::timeout_at(read_deadline,
stream.read(&mut response_header))
- .await;
- let Ok(header_read) = header_read else {
- error!(
- "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting
for VSR response header for TCP request with code: {code}",
- );
- return Err(IggyError::Disconnected);
- };
- header_read.map_err(|error| {
- error!(
- "Failed to read VSR response header for TCP
request with code: {code}: {error}",
- );
- IggyError::Disconnected
- })?;
-
- let response_size =
crate::vsr::response_size(&response_header)?;
- let body_size = response_size -
iggy_binary_protocol::HEADER_SIZE;
- let body = if body_size > 0 {
- let mut body = BytesMut::with_capacity(body_size);
- let body_read = tokio::time::timeout_at(
- read_deadline,
- stream.read_buf(&mut body, body_size),
- )
- .await;
- let Ok(body_read) = body_read else {
- error!(
- "Timed out after {RESPONSE_READ_TIMEOUT:?}
waiting for VSR response body for TCP request with code: {code}",
- );
- return Err(IggyError::Disconnected);
- };
- body_read.map_err(|error| {
- error!(
- "Failed to read VSR response body for TCP
request with code: {code}: {error}",
- );
- IggyError::Disconnected
- })?;
- body.freeze()
- } else {
- Bytes::new()
- };
-
- frame_complete = true;
- crate::vsr::observe_metadata_reply(&metadata_watermark,
&response_header);
- match crate::vsr::decode_response_split(&response_header,
body)
- .map_err(|error| outcome_so_far.observe(error))
+ };
+ replay = true;
+ session_owner.observe_metadata_reply(&reply.header);
+ match crate::vsr::decode_response_split(&reply.header,
reply.body)
+ .map_err(|error| outcome_so_far.observe(error))
+ {
+ Err(error) if !retry_transient => break Err(error),
+ // `TransientNotCommitted`: the op's outcome is unknown
+ // (e.g. a view change canceled it in flight) -- ONLY a
+ // same-session replay of the same request id is safe
+ // (the client-table serves it from cache if it did
+ // commit). Replay on this connection for the whole
+ // request budget; never hand it to the failover path,
+ // which re-issues under a fresh session and could
+ // double-apply a committed write.
+ Err(IggyError::TransientNotCommitted)
+ if code !=
iggy_binary_protocol::codes::POLL_MESSAGES_ON_PRIMARY_CODE
+ && tokio::time::Instant::now() < read_deadline =>
{
- Err(error) if !retry_transient => return Err(error),
- // `TransientNotCommitted`: the op's outcome is unknown
- // (e.g. a view change canceled it in flight) -- ONLY a
- // same-session replay of the same request id is safe
- // (the client-table serves it from cache if it did
- // commit). Replay on this connection for the whole
- // request budget; never hand it to the failover path,
- // which re-issues under a fresh session and could
- // double-apply a committed write.
- Err(IggyError::TransientNotCommitted)
- if code !=
iggy_binary_protocol::codes::POLL_MESSAGES_ON_PRIMARY_CODE
- && tokio::time::Instant::now() < read_deadline
=>
- {
- let remaining = read_deadline
-
.saturating_duration_since(tokio::time::Instant::now());
-
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
- }
- // `TransientNotAccepted`: the server never admitted
the
- // request, so it is re-issuable anywhere. Replay here
- // briefly, then hand it back to the caller for a
- // leader recheck / failover.
- Err(IggyError::TransientNotAccepted)
- if tokio::time::Instant::now() <
transient_deadline =>
- {
- let remaining = transient_deadline
-
.saturating_duration_since(tokio::time::Instant::now());
-
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
- }
- other => return other,
+ let remaining =
+
read_deadline.saturating_duration_since(tokio::time::Instant::now());
+
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
Review Comment:
nit: the transient replay sleeps here while still holding the link guard (up
to 2s on 58/89, up to 30s on `TransientNotCommitted`), so every other call on
the client, PING included, waits. Releasing the guard before the pause and
keeping the retained header would fix it (same in WS/QUIC).
##########
core/server/src/dispatch/partition.rs:
##########
@@ -690,7 +688,7 @@ pub const fn fence_poll_context(
}
if requested.owner_generation != local.owner_generation {
return Err(IggyError::ConsumerGroupPartitionNotOwned(
- group_id,
+ client_id as u32,
Review Comment:
nit: `client_id as u32` truncates the u128 VSR client id, while
`iggy_partition.rs:7283` puts the group id in the same slot, so one error
reports two different identities (clients only see the code). Pre-existing.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1674,28 +1559,25 @@ impl TcpClient {
Option<iggy_binary_protocol::consensus::RequestHeader>,
Result<Bytes, IggyError>,
) {
- let stream = self.stream.clone();
- let consensus_session = self.consensus_session.clone();
- let metadata_watermark =
Arc::clone(&self.poll_router.metadata_watermark);
+ let link = self.link.clone();
+ let session_owner = Arc::clone(&self.session_owner);
let mut outcome_so_far = std::mem::take(retry_outcome);
- // SAFETY: we run code holding the `stream` lock in a task so we can't
be cancelled while holding the lock.
+ // A task finishes the exchange when its caller goes away. An unbound
+ // request that left before its reply would close the link.
let joined = tokio::spawn(async move {
- let mut stream_guard = stream.lock().await;
- let Some(stream) = stream_guard.as_mut() else {
+ let mut link_guard = link.lock().await;
Review Comment:
**warning**: the link mutex is held across the whole exchange (same in
`websocket_client.rs:1361`, `quic_client.rs:1194`), so the new 256-slot
`MuxLink` runs lockstep and only adds two task hand-offs, a semaphore, a map
and a oneshot per request: producer about -4..-5%, 2x client context switches.
A direct lockstep path (write, flush, read on the caller task) recovers master;
or lock only to clone the link if pipelining is the goal.
##########
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:
nit: every transport failure now surfaces as `Disconnected`, so the
`TcpError` / `QuicError` / `WebSocketSendError` / `WebSocketReceiveError` arms
in the clients (e.g. `websocket_client.rs:366-378`) are dead. In-tree reconnect
still works, but downstream code matching those variants silently stops
matching; worth dropping the dead arms and noting the error contract change.
##########
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),
+ Err(_) => {
+ error!(
+ "No reply to request {request} from {} before its
deadline.",
+ self.shared.peer
+ );
+ Err(IggyError::Disconnected)
+ }
+ }
+ }
+
+ /// Fails every waiting request. The writer task then ends the connection
+ /// the way its protocol expects.
+ pub(crate) fn close(&self) {
+ self.shared.close(&IggyError::Disconnected);
+ }
+
+ /// Returns when the link closes, for any reason.
+ pub(crate) async fn closed(&self) {
+ let mut closed = self.shared.closed.subscribe();
+ // The sender lives as long as the link.
+ let _ = closed.wait_for(|closed| *closed).await;
+ }
+
+ /// Ends the connection the way its protocol expects: a FIN for TCP, a
+ /// Close message for WebSocket. The link counts as closed first, so the
+ /// reader does not report the end of the connection as an error.
+ pub(crate) async fn shutdown(self) {
+ let Self {
+ shared,
+ queue,
+ mut tasks,
+ } = self;
+ shared.close(&IggyError::Disconnected);
+ drop(queue);
+ let _ = timeout(SHUTDOWN_TIMEOUT, &mut tasks.writer).await;
+ }
+
+ fn start<Writer, Reader>(
+ peer: SocketAddr,
+ writer: impl FnOnce(mpsc::Receiver<Frame>, Arc<MuxShared>) -> Writer,
+ reader: impl FnOnce(Arc<MuxShared>) -> Reader,
+ ) -> Self
+ where
+ Writer: Future<Output = ()> + Send + 'static,
+ Reader: Future<Output = ()> + Send + 'static,
+ {
+ let shared = Arc::new(MuxShared {
+ peer,
+ pending: StdMutex::new(Pending {
+ open: true,
+ waiters: HashMap::new(),
+ }),
+ permits: Semaphore::new(MAX_PENDING_PER_LINK),
+ unbound: Mutex::new(()),
+ closed: watch::Sender::new(false),
+ });
+ let (queue, frames) = mpsc::channel(LINK_WRITE_QUEUE);
+ let tasks = LinkTasks {
+ writer: tokio::spawn(writer(frames, Arc::clone(&shared))),
+ reader: tokio::spawn(reader(Arc::clone(&shared))),
+ };
+ Self {
+ shared,
+ queue,
+ tasks,
+ }
+ }
+}
+
+impl MuxShared {
+ fn wait_for(&self, request: u64, waiter: Waiter) ->
Result<PendingEntry<'_>, IggyError> {
+ let mut pending = self.lock_pending();
+ if !pending.open {
+ return Err(IggyError::NotConnected);
+ }
+ match pending.waiters.entry(request) {
+ // Two requests with one id could not tell their replies apart.
+ Entry::Occupied(_) => {
+ error!(
+ "Request {request} already waits for a reply from {}.",
+ self.peer
+ );
+ Err(IggyError::NotConnected)
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(waiter);
+ Ok(PendingEntry {
+ shared: self,
+ request,
+ close_if_abandoned: false,
+ })
+ }
+ }
+ }
+
+ /// Cuts every whole frame off the front of `buffer` and dispatches it.
+ fn deliver_buffered(&self, buffer: &mut BytesMut) -> Result<(), IggyError>
{
+ while let Some(&header) = buffer.first_chunk::<HEADER_SIZE>() {
+ let Ok(size) = response_size(&header) else {
+ error!(
+ "Received a frame shorter than its header from {}.",
+ self.peer
+ );
+ return Err(IggyError::Disconnected);
+ };
+ if buffer.len() < size {
+ break;
+ }
+ let body = buffer.split_to(size).freeze().slice(HEADER_SIZE..);
+ self.dispatch(Reply { header, body })?;
+ }
+ Ok(())
+ }
+
+ /// Gives a reply to the request that waits for it. An error is the reason
+ /// to close the link: an eviction ends the session, so it ends every
+ /// request.
+ fn dispatch(&self, reply: Reply) -> Result<(), IggyError> {
+ match peek_command(&reply.header) {
+ Command::Reply => {
+ let request = reply_request_id(&reply.header);
+ let waiter = self.lock_pending().waiters.remove(&request);
+ match waiter {
+ Some(waiter) => {
+ let _ = waiter.send(Ok(reply));
+ }
+ None => trace!(
+ "Dropped a reply to request {request} from {}, which
nothing waits for.",
+ self.peer
+ ),
+ }
+ Ok(())
+ }
+ Command::Eviction => Err(decode_eviction(&reply.header)),
+ _ => {
+ error!(
+ "Received a frame that is neither a reply nor an eviction
from {}.",
+ self.peer
+ );
+ Err(IggyError::Disconnected)
+ }
+ }
+ }
+
+ /// Fails every waiting request with `reason`. A request that comes later
+ /// fails with `NotConnected`.
+ fn close(&self, reason: &IggyError) {
+ let waiters = {
+ let mut pending = self.lock_pending();
+ pending.open = false;
+ std::mem::take(&mut pending.waiters)
+ };
+ self.permits.close();
+ self.closed.send_replace(true);
+ for waiter in waiters.into_values() {
+ let _ = waiter.send(Err(reason.clone()));
+ }
+ }
+
+ /// The next frame to write, or `None` once the link closed or every
+ /// sender went away.
+ async fn next_frame(&self, frames: &mut mpsc::Receiver<Frame>) ->
Option<Frame> {
+ let mut closed = self.closed.subscribe();
+ tokio::select! {
+ biased;
+ _ = closed.wait_for(|closed| *closed) => None,
+ frame = frames.recv() => frame,
+ }
+ }
+
+ fn is_open(&self) -> bool {
+ self.lock_pending().open
+ }
+
+ fn lock_pending(&self) -> MutexGuard<'_, Pending> {
+ self.pending.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
+impl Drop for PendingEntry<'_> {
+ fn drop(&mut self) {
+ let abandoned = self
+ .shared
+ .lock_pending()
+ .waiters
+ .remove(&self.request)
+ .is_some();
+ if abandoned && self.close_if_abandoned {
+ debug!(
+ "Closing the link to {}: request {} left before its reply.",
+ self.shared.peer, self.request
+ );
+ self.shared.close(&IggyError::Disconnected);
+ }
+ }
+}
+
+async fn write_tcp<W: AsyncWrite + Unpin>(
+ mut writer: W,
+ mut frames: mpsc::Receiver<Frame>,
+ shared: Arc<MuxShared>,
+) {
+ while let Some(frame) = shared.next_frame(&mut frames).await {
+ let written = async {
+ writer.write_all(bytemuck::bytes_of(&frame.header)).await?;
Review Comment:
nit: header and body go out as two `write_all` calls through the
`BufWriter`; a body of 8 KiB or more becomes two segments, and with
`nodelay=false` (default) an 8-64 KiB body can hit a Nagle / delayed-ACK stall.
One gather write avoids it.
##########
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:
nit: polls with auto-commit are NonReplicated and take no lane, but they
move the stored offset of their partition history, so "one mutation at a time
per partition history" doesn't hold for them (client-side twin of the
duplicate-delivery race). Give them the partition lane or document the
exemption.
##########
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:
nit: after a long-lived link drops, `retry` resets to 0 and the slot redials
immediately with no jitter, so a server restart makes every client redial in
the same instant. Drawing jitter for the first redial too (`backoff(0, ..)`)
avoids the herd. Driver unused today.
##########
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:
**warning** (unused driver, must be fixed before it is wired): once a call
is uncertain, a committed business rejection (e.g. replayed CreateStream
answered `StreamNameAlreadyExists`) falls here and ends as
`OperationUncertain`; `resolve()` rebuilds with `uncertain: true` and hits the
same arm forever, so the lane stays blocked. A fenced pre-consensus rejection
is byte-identical on the wire, so the server needs to mark committed replies
before the driver can tell them apart.
--
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]
