This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 8ded54789 fix(sdk): keep an explicit disconnect until the next connect
(#4388)
8ded54789 is described below
commit 8ded54789a0ac08414de4f73d7311789856ba2c6
Author: Tom Planche <[email protected]>
AuthorDate: Mon Oct 12 15:47:33 2026 +0200
fix(sdk): keep an explicit disconnect until the next connect (#4388)
---
core/common/src/traits/client.rs | 9 +-
core/integration/tests/sdk/client_lifecycle.rs | 28 +---
core/sdk/src/client_wrappers/binary_client.rs | 27 +++-
.../src/clients/binary_personal_access_tokens.rs | 4 +-
core/sdk/src/clients/binary_users.rs | 4 +-
core/sdk/src/clients/client.rs | 63 +++++---
core/sdk/src/quic/quic_client.rs | 101 +++++++++++-
core/sdk/src/tcp/tcp_client.rs | 175 +++++++++++++++++++--
core/sdk/src/websocket/websocket_client.rs | 72 ++++++++-
foreign/cpp/include/iggy.hpp | 6 +-
foreign/python/apache_iggy.pyi | 10 +-
foreign/python/src/client.rs | 10 +-
foreign/python/tests/test_connectivity.py | 13 +-
13 files changed, 431 insertions(+), 91 deletions(-)
diff --git a/core/common/src/traits/client.rs b/core/common/src/traits/client.rs
index 8df424215..be7afbcbc 100644
--- a/core/common/src/traits/client.rs
+++ b/core/common/src/traits/client.rs
@@ -48,7 +48,14 @@ pub trait Client:
/// If the client is already connected, it will do nothing.
async fn connect(&self) -> Result<(), IggyError>;
- /// Disconnect from the server. If the client is not connected, it will do
nothing.
+ /// Disconnect from the server. Repeated calls are safe.
+ ///
+ /// On TCP, QUIC and WebSocket, the disconnect holds until the next
`connect()`,
+ /// also on a client that was not connected: nothing reconnects the client
on
+ /// its own, even with auto-login credentials. Until then, a request that
needs
+ /// a session fails with `IggyError::Disconnected`, and any other request
with
+ /// `IggyError::NotConnected`. A connection that drops without this call
can
+ /// still reconnect on its own. Over HTTP this call does nothing.
async fn disconnect(&self) -> Result<(), IggyError>;
/// Shut down the client and release all the resources. Repeated calls are
safe.
diff --git a/core/integration/tests/sdk/client_lifecycle.rs
b/core/integration/tests/sdk/client_lifecycle.rs
index 801efd0f0..571795477 100644
--- a/core/integration/tests/sdk/client_lifecycle.rs
+++ b/core/integration/tests/sdk/client_lifecycle.rs
@@ -17,8 +17,7 @@
//! A client that is shut down stays shut down, and an explicit disconnect
//! stays in effect until the caller connects again. These tests pin that
-//! contract. The ignored ones still fail on the current SDK (#4287) and stay
-//! ignored until their fixes land.
+//! contract.
use iggy::prelude::*;
use integration::harness::TestHarness;
@@ -39,10 +38,11 @@ const FIRST_HEARTBEAT_SETTLE: Duration =
Duration::from_millis(500);
/// with no dial and no implicit wait, so it has to fail well within this
bound.
const SDK_SIDE_FAILURE: Duration = Duration::from_millis(500);
/// A disconnect closes local resources only, so it must return well within
-/// this bound. On QUIC, a heartbeat redial that lands inside the drain of the
old
+/// this bound. On QUIC, a redial that lands inside the drain of the old
/// connection blocks `disconnect()` in `endpoint.wait_idle()` for good, so the
/// tests bound the call to fail instead of hang. That redial depends on
timing,
-/// so it does not block every run.
+/// and a connect already past the caller-intent check can still make it
+/// (problem 6 of #4287).
const DISCONNECT_BOUND: Duration = Duration::from_secs(5);
#[iggy_harness]
@@ -73,19 +73,9 @@ async fn
given_a_shut_down_client_when_disconnected_should_stay_shut_down(harnes
}
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-#[ignore = "fails until #4287 is fixed (the heartbeat undoes an explicit
disconnect); run this test explicitly with --ignored"]
async fn
given_an_auto_login_client_when_explicitly_disconnected_should_stay_disconnected(
harness: &TestHarness,
) {
- // TODO: stop the heartbeat task in `IggyClient::disconnect`, and let
- // `connect` start it again. That alone is not enough: the heartbeat ping
- // reconnects through the same path as a direct `ping()`, so this test also
- // needs the caller-intent flag of the next test. The `IggyClient`, Python
- // and C++ docs describe the current behavior, so they must change with the
- // fix. On QUIC, the heartbeat redial also blocks `disconnect()` in
- // `endpoint.wait_idle()`: stopping the heartbeat first removes that redial
- // here, but `QuicClient::disconnect` must not wait on a connection that a
- // redial opens.
let client = auto_login_client(harness, FAST_HEARTBEAT);
client.connect().await.unwrap();
client
@@ -108,19 +98,9 @@ async fn
given_an_auto_login_client_when_explicitly_disconnected_should_stay_dis
}
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
-#[ignore = "fails until #4287 is fixed (a request after disconnect reconnects
an auto-login client); run this test explicitly with --ignored"]
async fn
given_a_disconnected_auto_login_client_when_sending_requests_should_fail_without_reconnecting(
harness: &TestHarness,
) {
- // TODO: set a caller-intent flag in the `Client` trait impl of
`disconnect()`
- // on each transport, and clear it in `connect()`. Only in the trait impl:
on
- // WebSocket and QUIC, `handle_leader_redirection` and the reconnect path
call
- // the inherent `disconnect()`, and a flag set there would block every
later
- // heal. While the flag is set, `send_raw_with_response` fails with
- // `NotConnected` on the SDK side instead of reconnecting with the
configured
- // credentials. Do not read the intent from `get_state() == Disconnected`:
a
- // socket error or a timeout also leaves the client `Disconnected`, and
that
- // loss must still heal on its own.
let client = disconnected_auto_login_client(harness).await;
let ping = timeout(SDK_SIDE_FAILURE, client.ping()).await;
diff --git a/core/sdk/src/client_wrappers/binary_client.rs
b/core/sdk/src/client_wrappers/binary_client.rs
index e8b9dfcda..f1cbe7225 100644
--- a/core/sdk/src/client_wrappers/binary_client.rs
+++ b/core/sdk/src/client_wrappers/binary_client.rs
@@ -25,11 +25,13 @@ use iggy_common::{DiagnosticEvent, IggyError};
impl Client for ClientWrapper {
async fn connect(&self) -> Result<(), IggyError> {
match self {
- ClientWrapper::Iggy(client) => client.connect().await,
- ClientWrapper::Http(client) => client.connect().await,
- ClientWrapper::Tcp(client) => client.connect().await,
- ClientWrapper::Quic(client) => client.connect().await,
- ClientWrapper::WebSocket(client) => client.connect().await,
+ // The trait `connect` is the caller's: it clears a held
disconnect,
+ // where each transport's own `connect` keeps it.
+ ClientWrapper::Iggy(client) => Client::connect(client).await,
+ ClientWrapper::Http(client) => Client::connect(client).await,
+ ClientWrapper::Tcp(client) => Client::connect(client).await,
+ ClientWrapper::Quic(client) => Client::connect(client).await,
+ ClientWrapper::WebSocket(client) => Client::connect(client).await,
}
}
@@ -63,3 +65,18 @@ impl Client for ClientWrapper {
}
}
}
+
+impl ClientWrapper {
+ /// Connects again through the transport's own `connect`, which leaves a
+ /// caller's `disconnect()` in force: a reconnect the client starts on its
+ /// own, such as after a leader redirect, is refused while it holds.
+ pub(crate) async fn reconnect_keeping_intent(&self) -> Result<(),
IggyError> {
+ match self {
+ ClientWrapper::Tcp(client) => client.connect().await,
+ ClientWrapper::Quic(client) => client.connect().await,
+ ClientWrapper::WebSocket(client) => client.connect().await,
+ ClientWrapper::Iggy(client) => Client::connect(client).await,
+ ClientWrapper::Http(client) => Client::connect(client).await,
+ }
+ }
+}
diff --git a/core/sdk/src/clients/binary_personal_access_tokens.rs
b/core/sdk/src/clients/binary_personal_access_tokens.rs
index cfbe169d5..c0456b441 100644
--- a/core/sdk/src/clients/binary_personal_access_tokens.rs
+++ b/core/sdk/src/clients/binary_personal_access_tokens.rs
@@ -18,8 +18,8 @@
use crate::clients::redirect_login_settled;
use crate::prelude::{ClientWrapper, IggyClient};
use async_trait::async_trait;
+use iggy_common::PersonalAccessTokenClient;
use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{Client, PersonalAccessTokenClient};
use iggy_common::{
IdentityInfo, IggyError, PersonalAccessTokenExpiry,
PersonalAccessTokenInfo,
RawPersonalAccessToken,
@@ -77,7 +77,7 @@ impl PersonalAccessTokenClient for IggyClient {
if should_redirect {
info!("Redirected to leader, reconnecting and re-authenticating");
- self.connect().await?;
+ self.reconnect_after_redirect().await?;
// The reconnect signs in with the credentials this very call just
// remembered, so on a client without a configured `AutoLogin` the
// session is already this user's: signing in again would cost a
diff --git a/core/sdk/src/clients/binary_users.rs
b/core/sdk/src/clients/binary_users.rs
index 606e11e44..e014d4dd6 100644
--- a/core/sdk/src/clients/binary_users.rs
+++ b/core/sdk/src/clients/binary_users.rs
@@ -19,9 +19,9 @@ use crate::client_wrappers::client_wrapper::ClientWrapper;
use crate::clients::redirect_login_settled;
use crate::prelude::IggyClient;
use async_trait::async_trait;
+use iggy_common::UserClient;
use iggy_common::UserUpdateOptions;
use iggy_common::locking::IggyRwLockFn;
-use iggy_common::{Client, UserClient};
use iggy_common::{
Identifier, IdentityInfo, IggyError, Permissions, UserInfo,
UserInfoDetails, UserStatus,
};
@@ -116,7 +116,7 @@ impl UserClient for IggyClient {
if should_redirect {
info!("Redirected to leader, reconnecting and re-authenticating");
- self.connect().await?;
+ self.reconnect_after_redirect().await?;
// The reconnect signs in with the credentials this very call just
// remembered, so on a client without a configured `AutoLogin` the
// session is already this user's: signing in again would cost a
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 64328d14a..992b6ecd4 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -123,9 +123,11 @@ const SESSION_CONTROL_CODES: [u32; 5] = [
/// [`IggyProducer`] so that _background_ producers flush the latest state.
Finally,
/// call [`shutdown()`] on the [`IggyClient`] which closes the connection.
/// Use [`disconnect()`] rather than [`shutdown()`] to close the connection
but keep the client usable, as a
-/// client that has been shut down cannot reconnect. Explicit disconnect
clears the remembered sign-in.
-/// Configured auto-login still authenticates on the next connection;
otherwise call
-/// [`connect()`] and [`login_user()`] again before issuing requests.
+/// client that has been shut down cannot reconnect. The disconnect holds
until the next
+/// [`connect()`]: requests fail and nothing reconnects in the background,
even with `auto-login`.
+/// Explicit disconnect clears the remembered sign-in. Configured
auto-login still authenticates
+/// on the next connection; otherwise call [`connect()`] and
[`login_user()`] again before
+/// issuing requests.
///
/// # Examples
///
@@ -954,27 +956,24 @@ impl IggyClient {
impl Drop for IggyClient {
fn drop(&mut self) {
- let heartbeat_handle = self
- .heartbeat_handle
- .get_mut()
- .unwrap_or_else(|error| error.into_inner())
- .take();
- if let Some(handle) = heartbeat_handle {
+ if let Some(handle) = self.take_heartbeat() {
handle.abort();
}
}
}
-#[async_trait]
-impl Client for IggyClient {
- async fn connect(&self) -> Result<(), IggyError> {
- let heartbeat_interval;
- {
- let client = self.client.read().await;
- client.connect().await?;
- heartbeat_interval = client.heartbeat_interval().await;
- }
+impl IggyClient {
+ /// Connects again after a leader redirect without clearing the caller's
+ /// `disconnect()`: one that lands during the redirect holds, and the
+ /// transport refuses the reconnect.
+ pub(crate) async fn reconnect_after_redirect(&self) -> Result<(),
IggyError> {
+ self.client.read().await.reconnect_keeping_intent().await?;
+ self.start_heartbeat().await;
+ Ok(())
+ }
+ async fn start_heartbeat(&self) {
+ let heartbeat_interval =
self.client.read().await.heartbeat_interval().await;
let mut heartbeat_handle = self
.heartbeat_handle
.lock()
@@ -983,7 +982,7 @@ impl Client for IggyClient {
.as_ref()
.is_some_and(|handle| !handle.is_finished())
{
- return Ok(());
+ return;
}
drop(heartbeat_handle.take());
@@ -1010,10 +1009,36 @@ impl Client for IggyClient {
sleep(heartbeat_interval.get_duration()).await
}
}));
+ }
+
+ fn take_heartbeat(&self) -> Option<JoinHandle<()>> {
+ self.heartbeat_handle
+ .lock()
+ .unwrap_or_else(|error| error.into_inner())
+ .take()
+ }
+}
+
+#[async_trait]
+impl Client for IggyClient {
+ async fn connect(&self) -> Result<(), IggyError> {
+ self.client.read().await.connect().await?;
+ self.start_heartbeat().await;
Ok(())
}
async fn disconnect(&self) -> Result<(), IggyError> {
+ // HTTP keeps no connection, so its disconnect changes nothing and the
+ // heartbeat keeps running.
+ let is_http = matches!(&*self.client.read().await,
ClientWrapper::Http(_));
+ // A heartbeat ping on a disconnected client only logs errors, and on
+ // QUIC a redial it starts during the disconnect blocks the call. The
+ // aborted task is awaited, so a reconnect it owned is over before the
+ // transport tears down. `connect` starts the heartbeat again.
+ if !is_http && let Some(handle) = self.take_heartbeat() {
+ handle.abort();
+ let _ = handle.await;
+ }
self.client.read().await.disconnect().await
}
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 3292d13e7..61399ae93 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -101,6 +101,11 @@ pub struct QuicClient {
/// admit (its replica of the target partition group is not the primary).
roster_endpoints: Mutex<Vec<String>>,
roster_learned: AtomicBool,
+ /// Set by an explicit `Client::disconnect` and cleared by
`Client::connect`.
+ /// While it is set, no request reconnects the client on its own. A
+ /// connection that drops without a `disconnect()` call leaves it clear, so
+ /// that loss still heals.
+ disconnected_by_caller: AtomicBool,
/// Serializes leader checks and roster walks after refused requests, so
/// concurrent QUIC streams cannot tear down each other's new connection.
routing_lock: Mutex<()>,
@@ -120,10 +125,12 @@ impl Default for QuicClient {
#[async_trait]
impl Client for QuicClient {
async fn connect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(false, Ordering::SeqCst);
QuicClient::connect(self).await
}
async fn disconnect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(true, Ordering::SeqCst);
self.forget_session_credentials().await;
QuicClient::disconnect_transport(self).await?;
self.reset_vsr_session().await
@@ -183,6 +190,12 @@ impl BinaryTransport for QuicClient {
payload: Bytes,
context: PartitionContext,
) -> Result<Bytes, IggyError> {
+ if self.caller_disconnected() {
+ return Err(match self.get_state().await {
+ ClientState::Shutdown => IggyError::ClientShutdown,
+ _ => IggyError::NotConnected,
+ });
+ }
if is_poll_routing_code(code) {
return self
.send_poll_request_with_context(code, payload, context)
@@ -656,6 +669,7 @@ impl QuicClient {
skip_auto_login_once: Mutex::new(false),
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
+ disconnected_by_caller: AtomicBool::new(false),
routing_lock: Mutex::new(()),
connect_coordinator: ConnectCoordinator::new(),
consumer_group_state:
Arc::new(iggy_common::ConsumerGroupClientState::new()),
@@ -702,7 +716,7 @@ impl QuicClient {
crate::vsr::decode_response(Bytes::from(buffer))
}
- async fn connect(&self) -> Result<(), IggyError> {
+ pub(crate) async fn connect(&self) -> Result<(), IggyError> {
self.connect_with_settlement(false).await
}
@@ -730,6 +744,13 @@ impl QuicClient {
.await
}
+ /// A reconnect sweep checks this before each dial and before it installs
+ /// the connection: once the caller disconnects, nothing in the sweep may
+ /// bring the client back.
+ fn caller_disconnected(&self) -> bool {
+ self.disconnected_by_caller.load(Ordering::SeqCst)
+ }
+
async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(),
IggyError> {
let settle_off_leader = context.settle_off_leader();
let single_attempt = context.single_attempt();
@@ -750,6 +771,10 @@ impl QuicClient {
trace!("Client is already connecting.");
return Ok(());
}
+ _ if self.caller_disconnected() => {
+ trace!("Cannot connect. The caller disconnected the
client.");
+ return Err(IggyError::NotConnected);
+ }
_ => {}
}
@@ -773,6 +798,11 @@ impl QuicClient {
let connection;
let remote_address;
loop {
+ if self.caller_disconnected() {
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::NotConnected);
+ }
let server_address_str =
self.current_server_address.lock().await.clone();
let server_address =
tokio::net::lookup_host(&server_address_str)
.await
@@ -850,6 +880,13 @@ impl QuicClient {
break;
}
+ if self.caller_disconnected() {
+ connection.close(0u32.into(), b"");
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::NotConnected);
+ }
+
let now = IggyTimestamp::now();
info!("{NAME} client has connected to server: {remote_address} at
{now}",);
self.set_state(ClientState::Connected).await;
@@ -911,6 +948,12 @@ impl QuicClient {
.map(|_| ()),
};
if let Err(error) = result {
+ // An explicit disconnect landed during the sign-in. The
+ // non-retryable arm below would keep the connection up.
+ if self.caller_disconnected() {
+ self.disconnect_transport().await?;
+ return Err(IggyError::NotConnected);
+ }
if crate::vsr::resume_error_is_retryable(&error) {
self.disconnect_transport().await?;
} else {
@@ -1030,7 +1073,10 @@ impl QuicClient {
}
async fn disconnect_transport(&self) -> Result<(), IggyError> {
- if self.get_state().await == ClientState::Disconnected {
+ // Taken before the state is read: a teardown aborted after it set
+ // `Disconnected` can leave the connection installed.
+ let connection = self.connection.lock().await.take();
+ if connection.is_none() && self.get_state().await ==
ClientState::Disconnected {
return Ok(());
}
@@ -1039,7 +1085,7 @@ impl QuicClient {
self.config.client_address
);
self.set_state(ClientState::Disconnected).await;
- self.connection.lock().await.take();
+ drop(connection);
self.endpoint.wait_idle().await;
self.publish_event(DiagnosticEvent::Disconnected).await;
let now = IggyTimestamp::now();
@@ -2114,4 +2160,53 @@ mod tests {
let client = QuicClient::create(config);
assert!(client.is_err());
}
+
+ #[tokio::test]
+ async fn a_reconnect_after_an_explicit_disconnect_is_refused() {
+ let client = QuicClient::default();
+ Client::disconnect(&client).await.unwrap();
+
+ // The recovery path of a request that was in flight during the
+ // disconnect calls the inherent `connect`, not the trait one.
+ let reconnect = tokio::time::timeout(Duration::from_secs(1),
client.connect())
+ .await
+ .expect("a refused reconnect must not dial");
+
+ assert!(
+ matches!(reconnect, Err(IggyError::NotConnected)),
+ "got {reconnect:?}"
+ );
+ }
+
+ #[tokio::test]
+ async fn
a_teardown_cut_short_after_the_state_change_is_finished_by_the_next_one() {
+ let _ = rustls::crypto::ring::default_provider().install_default();
+ let certified =
rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
+ let config = quinn::ServerConfig::with_single_cert(
+ vec![certified.cert.der().clone()],
+
rustls::pki_types::PrivatePkcs8KeyDer::from(certified.signing_key.serialize_der())
+ .into(),
+ )
+ .unwrap();
+ let endpoint = quinn::Endpoint::server(config,
"127.0.0.1:0".parse().unwrap()).unwrap();
+ let client = QuicClient::create(Arc::new(QuicClientConfig {
+ server_address: endpoint.local_addr().unwrap().to_string(),
+ ..Default::default()
+ }))
+ .unwrap();
+ let (peer, connected) = tokio::join!(
+ async { endpoint.accept().await.unwrap().await.unwrap() },
+ Client::connect(&client)
+ );
+ connected.unwrap();
+ // What an abort leaves between the state change and the cleanup.
+ client.set_state(ClientState::Disconnected).await;
+
+ client.disconnect_transport().await.unwrap();
+
+ assert!(client.connection.lock().await.is_none());
+ tokio::time::timeout(Duration::from_secs(1), peer.closed())
+ .await
+ .expect("the live connection must be closed");
+ }
}
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 9907a9529..ba9517d32 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -111,6 +111,11 @@ pub struct TcpClient {
/// Set once a sign-in on this client has gone looking for the roster, so
/// that read happens once (see [`TcpClient::learn_roster_once`]).
roster_learned: AtomicBool,
+ /// Set by an explicit `Client::disconnect` and cleared by
`Client::connect`.
+ /// While it is set, no request reconnects the client on its own. A
+ /// connection that drops without a `disconnect()` call leaves it clear, so
+ /// that loss still heals.
+ disconnected_by_caller: AtomicBool,
/// Credentials a sign-in on this client succeeded with, so a reconnect --
/// onto this node or, after a failover, another one -- can re-establish
/// the session instead of surfacing `Unauthenticated`. Cleared on logout.
@@ -169,10 +174,12 @@ impl Default for TcpClient {
#[async_trait]
impl Client for TcpClient {
async fn connect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(false, Ordering::SeqCst);
TcpClient::connect(self).await
}
async fn disconnect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(true, Ordering::SeqCst);
// An explicit disconnect is caller intent, like a logout: the session
// it ends must not be resurrected by the next reconnect, so the
// remembered sign-in goes with it. Involuntary drops (a dead socket,
@@ -236,6 +243,12 @@ impl BinaryTransport for TcpClient {
payload: Bytes,
context: PartitionContext,
) -> Result<Bytes, IggyError> {
+ if self.caller_disconnected() {
+ return Err(match self.get_state().await {
+ ClientState::Shutdown => IggyError::ClientShutdown,
+ _ => IggyError::NotConnected,
+ });
+ }
if is_poll_routing_code(code) {
return self
.send_poll_request_with_context(code, payload, context)
@@ -619,6 +632,7 @@ impl TcpClient {
current_server_address: Mutex::new(server_address),
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
+ disconnected_by_caller: AtomicBool::new(false),
configured_password: Mutex::new(None),
configured_username: Mutex::new(None),
session_credentials: Mutex::new(None),
@@ -630,7 +644,7 @@ impl TcpClient {
})
}
- async fn connect(&self) -> Result<(), IggyError> {
+ pub(crate) async fn connect(&self) -> Result<(), IggyError> {
self.connect_with_settlement(false).await
}
@@ -682,6 +696,10 @@ impl TcpClient {
trace!("Client is already connecting.");
return Ok(());
}
+ _ if self.caller_disconnected() => {
+ trace!("Cannot connect. The caller disconnected the
client.");
+ return Err(IggyError::NotConnected);
+ }
_ => *state = ClientState::Connecting,
}
}
@@ -725,8 +743,19 @@ impl TcpClient {
sleep(remaining.get_duration()).await;
}
+ if self.caller_disconnected() {
+ self.fail_connect().await;
+ return Err(IggyError::NotConnected);
+ }
+
info!("{NAME} client is connecting to server:
{server_address}...");
match self.establish_bounded(&server_address,
&candidates).await {
+ // Dropped before it is installed: the caller disconnected
+ // while it was being dialed.
+ Ok(_) if self.caller_disconnected() => {
+ self.fail_connect().await;
+ return Err(IggyError::NotConnected);
+ }
Ok(connection) => {
let dialed = server_address.clone();
// The endpoint that answered is where this client now
@@ -785,6 +814,11 @@ impl TcpClient {
}
candidate = 0;
+ if self.caller_disconnected() {
+ self.fail_connect().await;
+ return Err(IggyError::NotConnected);
+ }
+
// An unreadable CA file, a domain that will not parse: no
// endpoint answered and at least one said why in a way that a
// retry cannot change, so the caller gets that reason instead
@@ -1293,6 +1327,13 @@ impl TcpClient {
self.publish_event(DiagnosticEvent::Disconnected).await;
}
+ /// A reconnect sweep checks this before each dial, before each retry pause
+ /// and before it signs in: once the caller disconnects, nothing in the
+ /// sweep may bring the client back.
+ fn caller_disconnected(&self) -> bool {
+ self.disconnected_by_caller.load(Ordering::SeqCst)
+ }
+
/// [`Self::establish`], bounded while other endpoints are queued behind
/// this one (see `FAILOVER_DIAL_TIMEOUT`). The bound covers the handshake
/// as well as the connect: a peer that accepts TCP and then never answers
@@ -1344,20 +1385,25 @@ impl TcpClient {
/// failover unauthenticated. The public [`Client::disconnect`] wraps this
/// and forgets them first.
async fn disconnect_transport(&self) -> Result<(), IggyError> {
- match self.get_state().await {
- ClientState::Disconnected => return Ok(()),
- // Preserve the stream a concurrent reconnect is installing.
- ClientState::Connecting => {
- trace!("Not disconnecting; a connect is already in flight.");
- return Ok(());
- }
- _ => {}
+ // Preserve the stream a concurrent reconnect is installing.
+ // A `Connecting` that no sweep owns, such as one left by an aborted
+ // heartbeat, is torn down like any other state.
+ if self.get_state().await == ClientState::Connecting &&
self.connect_coordinator.is_active()
+ {
+ trace!("Not disconnecting; a connect is already in flight.");
+ return Ok(());
+ }
+ // Taken before the state is read: a teardown aborted after it set
+ // `Disconnected` can leave the stream installed.
+ let stream = self.stream.lock().await.take();
+ if stream.is_none() && self.get_state().await ==
ClientState::Disconnected {
+ return Ok(());
}
let client_address = self.get_client_address_value().await;
info!("{NAME} client: {client_address} is disconnecting from
server...");
self.set_state(ClientState::Disconnected).await;
- self.stream.lock().await.take();
+ drop(stream);
self.publish_event(DiagnosticEvent::Disconnected).await;
let now = IggyTimestamp::now();
info!("{NAME} client: {client_address} has disconnected from server
at: {now}.");
@@ -1817,7 +1863,7 @@ mod tests {
use super::*;
use crate::vsr::replay_after_session_reset_is_safe;
use iggy_binary_protocol::codes::{
- GET_ME_CODE, JOIN_CONSUMER_GROUP_CODE, LOGOUT_USER_CODE,
SEND_MESSAGES_CODE,
+ GET_ME_CODE, JOIN_CONSUMER_GROUP_CODE, LOGOUT_USER_CODE, PING_CODE,
SEND_MESSAGES_CODE,
};
use iggy_binary_protocol::consensus::{REJECTION_SECTION_LEN,
write_rejection_section};
use iggy_binary_protocol::requests::system::BindSessionRequest;
@@ -1827,6 +1873,7 @@ mod tests {
WireDecode, WireEncode, WireName,
};
use std::sync::atomic::AtomicUsize;
+ use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
const SESSION_USER_ID: u32 = 7;
@@ -3793,4 +3840,110 @@ mod tests {
IggyDuration::from_str("5s").unwrap()
);
}
+
+ #[tokio::test]
+ async fn a_reconnect_after_an_explicit_disconnect_is_refused() {
+ let client = TcpClient::default();
+ Client::disconnect(&client).await.unwrap();
+
+ // The recovery path of a request that was in flight during the
+ // disconnect calls the inherent `connect`, not the trait one.
+ let reconnect = tokio::time::timeout(Duration::from_secs(1),
client.connect())
+ .await
+ .expect("a refused reconnect must not dial");
+
+ assert!(
+ matches!(reconnect, Err(IggyError::NotConnected)),
+ "got {reconnect:?}"
+ );
+ }
+
+ fn auto_login_client(server_address: String) -> TcpClient {
+ TcpClient::create(Arc::new(TcpClientConfig {
+ server_address,
+ auto_login: AutoLogin::Enabled(Credentials::UsernamePassword(
+ "iggy".to_string(),
+ "iggy".into(),
+ )),
+ reconnection: TcpClientReconnectionConfig {
+ enabled: true,
+ max_retries: None,
+ interval:
NonZeroIggyDuration::from_str("50ms").expect("duration"),
+ // The sweep checks the disconnect between pauses, so a zero
+ // pause keeps this test about the check, not the pacing.
+ reestablish_after: IggyDuration::new(Duration::ZERO),
+ },
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client")
+ }
+
+ #[tokio::test]
+ async fn a_disconnect_during_a_reconnect_sweep_ends_the_sweep() {
+ // Every sign-in fails on the socket, so with unlimited retries the
+ // sweep only ends on a disconnect.
+ let (address, accepts) = counted_endpoint_that_hangs_up().await;
+ let client = Arc::new(auto_login_client(address));
+ let sweep = tokio::spawn({
+ let client = Arc::clone(&client);
+ async move { TcpClient::connect(&client).await }
+ });
+ while accepts.load(Ordering::SeqCst) == 0 {
+ sleep(Duration::from_millis(10)).await;
+ }
+
+ Client::disconnect(&*client).await.unwrap();
+ let result = tokio::time::timeout(Duration::from_secs(2), sweep)
+ .await
+ .expect("the sweep must end after the disconnect")
+ .expect("the sweep task must not panic");
+ let accepts_after_end = accepts.load(Ordering::SeqCst);
+ sleep(Duration::from_millis(300)).await;
+
+ assert!(
+ matches!(result, Err(IggyError::NotConnected)),
+ "got {result:?}"
+ );
+ assert_eq!(
+ accepts.load(Ordering::SeqCst),
+ accepts_after_end,
+ "nothing may redial after the sweep ends"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_request_after_an_explicit_disconnect_fails_without_a_dial() {
+ let (listener, address) = live_endpoint().await;
+ let client = auto_login_client(address);
+ Client::disconnect(&client).await.unwrap();
+
+ let ping = tokio::time::timeout(
+ Duration::from_secs(1),
+ client.send_raw_with_response(PING_CODE, Bytes::new()),
+ )
+ .await
+ .expect("a refused request must not wait");
+
+ assert!(matches!(ping, Err(IggyError::NotConnected)), "got {ping:?}");
+ assert!(
+ tokio::time::timeout(Duration::from_millis(200), listener.accept())
+ .await
+ .is_err(),
+ "the request must not dial"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_disconnect_after_a_shutdown_keeps_reporting_the_shutdown() {
+ let client = auto_login_client("127.0.0.1:8090".to_string());
+ Client::shutdown(&client).await.unwrap();
+ Client::disconnect(&client).await.unwrap();
+
+ let ping = client.send_raw_with_response(PING_CODE,
Bytes::new()).await;
+
+ assert!(
+ matches!(ping, Err(IggyError::ClientShutdown)),
+ "got {ping:?}"
+ );
+ }
}
diff --git a/core/sdk/src/websocket/websocket_client.rs
b/core/sdk/src/websocket/websocket_client.rs
index 46deacf7d..dbc482caa 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -98,6 +98,11 @@ pub struct WebSocketClient {
/// admit (its replica of the target partition group is not the primary).
roster_endpoints: Mutex<Vec<String>>,
roster_learned: AtomicBool,
+ /// Set by an explicit `Client::disconnect` and cleared by
`Client::connect`.
+ /// While it is set, no request reconnects the client on its own. A
+ /// connection that drops without a `disconnect()` call leaves it clear, so
+ /// that loss still heals.
+ disconnected_by_caller: AtomicBool,
/// Serializes leader checks and roster walks after refused requests, so
/// concurrent callers cannot tear down each other's new connection.
routing_lock: Mutex<()>,
@@ -114,10 +119,12 @@ impl Default for WebSocketClient {
#[async_trait]
impl Client for WebSocketClient {
async fn connect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(false, Ordering::SeqCst);
WebSocketClient::connect(self).await
}
async fn disconnect(&self) -> Result<(), IggyError> {
+ self.disconnected_by_caller.store(true, Ordering::SeqCst);
self.forget_session_credentials().await;
WebSocketClient::disconnect_transport(self).await?;
self.reset_vsr_session().await
@@ -177,6 +184,12 @@ impl BinaryTransport for WebSocketClient {
payload: Bytes,
context: PartitionContext,
) -> Result<Bytes, IggyError> {
+ if self.caller_disconnected() {
+ return Err(match self.get_state().await {
+ ClientState::Shutdown => IggyError::ClientShutdown,
+ _ => IggyError::NotConnected,
+ });
+ }
if is_poll_routing_code(code) {
return self
.send_poll_request_with_context(code, payload, context)
@@ -605,6 +618,7 @@ impl WebSocketClient {
skip_auto_login_once: Mutex::new(false),
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
+ disconnected_by_caller: AtomicBool::new(false),
routing_lock: Mutex::new(()),
connect_coordinator: ConnectCoordinator::new(),
consumer_group_state:
Arc::new(iggy_common::ConsumerGroupClientState::new()),
@@ -627,7 +641,7 @@ impl WebSocketClient {
}
}
- async fn connect(&self) -> Result<(), IggyError> {
+ pub(crate) async fn connect(&self) -> Result<(), IggyError> {
self.connect_with_settlement(false).await
}
@@ -655,6 +669,13 @@ impl WebSocketClient {
.await
}
+ /// A reconnect sweep checks this before each dial and before it installs
+ /// the stream: once the caller disconnects, nothing in the sweep may bring
+ /// the client back.
+ fn caller_disconnected(&self) -> bool {
+ self.disconnected_by_caller.load(Ordering::SeqCst)
+ }
+
async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(),
IggyError> {
let settle_off_leader = context.settle_off_leader();
let single_attempt = context.single_attempt();
@@ -668,6 +689,10 @@ impl WebSocketClient {
ClientState::Connected
| ClientState::Authenticating
| ClientState::Authenticated => return Ok(()),
+ _ if self.caller_disconnected() => {
+ trace!("Cannot connect. The caller disconnected the
client.");
+ return Err(IggyError::NotConnected);
+ }
_ => {}
}
@@ -705,6 +730,12 @@ impl WebSocketClient {
}
}
+ if self.caller_disconnected() {
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::NotConnected);
+ }
+
let server_addr = tokio::net::lookup_host(&*current_address)
.await
.map_err(|e| {
@@ -754,6 +785,13 @@ impl WebSocketClient {
Err(error) => return Err(error),
};
+ if self.caller_disconnected() {
+ drop(connection_stream);
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::NotConnected);
+ }
+
*self.stream.lock().await = Some(connection_stream);
*self.client_address.lock().await = Some(server_addr);
self.set_state(ClientState::Connected).await;
@@ -1098,6 +1136,12 @@ impl WebSocketClient {
.map(|_| ()),
};
if let Err(error) = result {
+ // An explicit disconnect landed during the sign-in. The
+ // non-retryable arm below would keep the stream up.
+ if self.caller_disconnected() {
+ self.disconnect_transport().await?;
+ return Err(IggyError::NotConnected);
+ }
if crate::vsr::resume_error_is_retryable(&error) {
self.disconnect_transport().await?;
} else {
@@ -1109,15 +1153,17 @@ impl WebSocketClient {
}
async fn disconnect_transport(&self) -> Result<(), IggyError> {
- if self.get_state().await == ClientState::Disconnected {
+ // Taken before the state is read: a teardown aborted after it set
+ // `Disconnected` can leave the stream installed.
+ let stream = self.stream.lock().await.take();
+ if stream.is_none() && self.get_state().await ==
ClientState::Disconnected {
return Ok(());
}
let client_address = self.get_client_address_value().await;
info!("{NAME} client: {client_address} is disconnecting from
server...");
self.set_state(ClientState::Disconnected).await;
-
- self.stream.lock().await.take();
+ drop(stream);
self.publish_event(DiagnosticEvent::Disconnected).await;
let now = IggyTimestamp::now();
@@ -1429,6 +1475,7 @@ const fn is_login_register_code(code: u32) -> bool {
mod tests {
use super::*;
use std::str::FromStr;
+ use std::time::Duration;
use futures_util::{SinkExt, StreamExt};
use iggy_binary_protocol::codes::{JOIN_CONSUMER_GROUP_CODE,
SEND_MESSAGES_CODE};
@@ -2073,4 +2120,21 @@ mod tests {
let client = client.unwrap();
assert_eq!(client.config.server_address, "localhost:8092");
}
+
+ #[tokio::test]
+ async fn a_reconnect_after_an_explicit_disconnect_is_refused() {
+ let client = WebSocketClient::default();
+ Client::disconnect(&client).await.unwrap();
+
+ // The recovery path of a request that was in flight during the
+ // disconnect calls the inherent `connect`, not the trait one.
+ let reconnect = tokio::time::timeout(Duration::from_secs(1),
client.connect())
+ .await
+ .expect("a refused reconnect must not dial");
+
+ assert!(
+ matches!(reconnect, Err(IggyError::NotConnected)),
+ "got {reconnect:?}"
+ );
+ }
}
diff --git a/foreign/cpp/include/iggy.hpp b/foreign/cpp/include/iggy.hpp
index 685952408..20430666d 100644
--- a/foreign/cpp/include/iggy.hpp
+++ b/foreign/cpp/include/iggy.hpp
@@ -3458,9 +3458,9 @@ class IggyBlockingClient final {
* Call Connect() to establish a new connection. Configured automatic login
* is applied when reconnecting.
*
- * @note Disconnect() does not stop the existing heartbeat task. With
- * automatic login configured, a heartbeat may reconnect and
- * authenticate the client in the background.
+ * @note The disconnect holds until the next Connect(): requests fail and
+ * nothing reconnects the client in the background, even with
+ * automatic login configured.
* @note The HTTP transport is stateless and treats this operation as a
* no-op.
* @throws IggyException if the client cannot disconnect cleanly.
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index ba7b247cb..7e32341a2 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -1377,12 +1377,10 @@ class IggyClient:
for the rest of `reestablish_after` (5 s by default). The sign-in made
with `login_user` is dropped, so it must be repeated after
reconnecting.
A client configured with auto-login credentials signs in again on
- `connect`. Over HTTP there is no connection to close and this call does
- nothing.
-
- Known issue: unless reconnection is disabled, the heartbeat of a client
- with auto-login credentials connects it again and signs in within one
- heartbeat interval. See https://github.com/apache/iggy/issues/4287.
+ `connect`. Until then, requests fail with `RuntimeError`, and nothing
+ connects the client again in the background, even with auto-login
+ credentials. Over HTTP there is no connection to close and this call
+ does nothing.
Raises:
RuntimeError: If the connection cannot be closed.
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index c237e7f45..7f35fb29b 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -522,12 +522,10 @@ impl IggyClient {
/// for the rest of `reestablish_after` (5 s by default). The sign-in made
/// with `login_user` is dropped, so it must be repeated after
reconnecting.
/// A client configured with auto-login credentials signs in again on
- /// `connect`. Over HTTP there is no connection to close and this call does
- /// nothing.
- ///
- /// Known issue: unless reconnection is disabled, the heartbeat of a client
- /// with auto-login credentials connects it again and signs in within one
- /// heartbeat interval. See https://github.com/apache/iggy/issues/4287.
+ /// `connect`. Until then, requests fail with `RuntimeError`, and nothing
+ /// connects the client again in the background, even with auto-login
+ /// credentials. Over HTTP there is no connection to close and this call
+ /// does nothing.
///
/// Raises:
/// RuntimeError: If the connection cannot be closed.
diff --git a/foreign/python/tests/test_connectivity.py
b/foreign/python/tests/test_connectivity.py
index c17111f5a..3b77d36f7 100644
--- a/foreign/python/tests/test_connectivity.py
+++ b/foreign/python/tests/test_connectivity.py
@@ -43,8 +43,8 @@ from .utils import (
def binary_transport_configs(auto_login: AutoLogin | None = None) -> list:
"""Return a config for every transport that holds a connection.
- Auto-login is disabled by default: with credentials to replay, a ping sent
- while disconnected reconnects on its own instead of failing.
+ Auto-login is disabled by default, so a test opts in only when it checks
+ the auto-login behavior.
TCP and QUIC wait out `reestablish_after` before they connect again, so the
configs set it to zero to keep the reconnect cases fast.
@@ -275,17 +275,20 @@ class TestLifecycle:
binary_transport_configs(AutoLogin.username_password("iggy", "iggy")),
)
@pytest.mark.asyncio
- async def test_disconnected_client_with_auto_login_signs_in_on_reconnect(
+ async def test_auto_login_client_stays_disconnected_until_connect(
self, config: TcpConfig | WebSocketConfig | QuicConfig
):
- """Test a client with auto-login credentials signs in again on
connect."""
+ """Test an auto-login client stays disconnected, then signs in on
connect."""
client = IggyClient(config)
await client.connect()
await wait_for_ping(client)
await client.disconnect()
- await client.connect()
+ with pytest.raises(RuntimeError):
+ await client.ping()
+
+ await client.connect()
await client.get_streams()
@pytest.mark.parametrize("config", binary_transport_configs())
