This is an automated email from the ASF dual-hosted git repository.
spetz 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 b1166a590 fix(sdk): stop a reconnect sweep when the client shuts down
(#4507)
b1166a590 is described below
commit b1166a590ad0a3f0d809bf13ad095412c1dbf60d
Author: Tom Planche <[email protected]>
AuthorDate: Wed Oct 14 23:22:41 2026 +0200
fix(sdk): stop a reconnect sweep when the client shuts down (#4507)
`shutdown()` must be final, also for a reconnect that runs at the same time.
Relates to #4287
---
core/sdk/src/quic/quic_client.rs | 167 ++++++++++++++++++++++++++---
core/sdk/src/tcp/tcp_client.rs | 144 ++++++++++++++++++++++---
core/sdk/src/websocket/websocket_client.rs | 156 ++++++++++++++++++++++++---
3 files changed, 428 insertions(+), 39 deletions(-)
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 037e0a008..7e23a3a22 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -110,6 +110,11 @@ pub struct QuicClient {
/// connection that drops without a `disconnect()` call leaves it clear, so
/// that loss still heals.
disconnected_by_caller: AtomicBool,
+ /// Set when `shutdown()` finishes its cleanup. The state turns `Shutdown`
+ /// before that cleanup and stays there, so it cannot tell a finished
+ /// shutdown from a cancelled one: a cancelled `shutdown()` leaves the
+ /// cleanup to the next call.
+ shutdown_complete: 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<()>,
@@ -679,6 +684,7 @@ impl QuicClient {
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
disconnected_by_caller: AtomicBool::new(false),
+ shutdown_complete: AtomicBool::new(false),
routing_lock: Mutex::new(()),
connect_coordinator: ConnectCoordinator::new(),
consumer_group_state:
Arc::new(iggy_common::ConsumerGroupClientState::new()),
@@ -739,6 +745,27 @@ impl QuicClient {
self.disconnected_by_caller.load(Ordering::SeqCst)
}
+ /// A reconnect sweep checks this before each dial and after it installs
+ /// the link: once the client is shut down or the caller disconnects,
+ /// nothing in the sweep may bring the client back.
+ async fn sweep_interrupted(&self) -> Option<IggyError> {
+ if self.get_state().await == ClientState::Shutdown {
+ return Some(IggyError::ClientShutdown);
+ }
+ self.caller_disconnected()
+ .then_some(IggyError::NotConnected)
+ }
+
+ async fn fail_connect(&self) {
+ // The state stays `Shutdown`, and a `Disconnected` event would
+ // contradict the `Shutdown` one.
+ if self.get_state().await == ClientState::Shutdown {
+ return;
+ }
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ }
+
async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(),
IggyError> {
let settle_off_leader = context.settle_off_leader();
let single_attempt = context.single_attempt();
@@ -786,10 +813,9 @@ 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);
+ if let Some(error) = self.sweep_interrupted().await {
+ self.fail_connect().await;
+ return Err(error);
}
let server_address_str =
self.current_server_address.lock().await.clone();
let server_address =
tokio::net::lookup_host(&server_address_str)
@@ -868,18 +894,22 @@ 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 link = QuicLink::new(connection,
self.config.response_buffer_size as usize);
+ self.link.lock().await.replace(link);
+ // Checked once the link is installed, not before: `shutdown()`
sets
+ // `Shutdown` before it takes the link, so a shutdown during the
+ // dial either takes this link or is seen here.
+ if let Some(error) = self.sweep_interrupted().await {
+ if let Some(link) = self.link.lock().await.take() {
+ link.close();
+ }
+ self.fail_connect().await;
+ return Err(error);
}
let now = IggyTimestamp::now();
info!("{NAME} client has connected to server: {remote_address} at
{now}",);
self.set_state(ClientState::Connected).await;
- let link = QuicLink::new(connection,
self.config.response_buffer_size as usize);
- self.link.lock().await.replace(link);
self.connected_at.lock().await.replace(now);
self.publish_event(DiagnosticEvent::Connected).await;
@@ -1043,11 +1073,14 @@ impl QuicClient {
}
async fn shutdown(&self) -> Result<(), IggyError> {
- if self.get_state().await == ClientState::Shutdown {
+ if self.shutdown_complete.load(Ordering::SeqCst) {
return Ok(());
}
info!("Shutting down the {NAME} QUIC client.");
+ // Set before the link is taken: a reconnect sweep that installs a link
+ // after this point sees `Shutdown` and closes it.
+ self.set_state(ClientState::Shutdown).await;
let link = self.link.lock().await.take();
if let Some(link) = link {
link.close();
@@ -1055,8 +1088,8 @@ impl QuicClient {
self.endpoint.wait_idle().await;
self.reset_vsr_session().await?;
- self.set_state(ClientState::Shutdown).await;
self.publish_event(DiagnosticEvent::Shutdown).await;
+ self.shutdown_complete.store(true, Ordering::SeqCst);
info!("{NAME} QUIC client has been shutdown.");
Ok(())
}
@@ -2396,4 +2429,112 @@ mod tests {
"a publish waited for room that an unread receiver never frees"
);
}
+
+ #[tokio::test]
+ async fn a_shutdown_during_a_reconnect_sweep_ends_the_sweep() {
+ 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 = Arc::new(
+ QuicClient::create(Arc::new(QuicClientConfig {
+ server_address: endpoint.local_addr().unwrap().to_string(),
+ reconnection: QuicClientReconnectionConfig {
+ reestablish_after:
IggyDuration::from_str("500ms").unwrap(),
+ ..Default::default()
+ },
+ ..Default::default()
+ }))
+ .unwrap(),
+ );
+ // A connection a moment ago: the sweep first waits out
+ // `reestablish_after`, and the shutdown lands during that wait.
+ client
+ .connected_at
+ .lock()
+ .await
+ .replace(IggyTimestamp::now());
+ let sweep = tokio::spawn({
+ let client = Arc::clone(&client);
+ async move { QuicClient::connect(&client).await }
+ });
+ while client.get_state().await != ClientState::Connecting {
+ sleep(Duration::from_millis(10)).await;
+ }
+
+ Client::shutdown(&*client).await.unwrap();
+ let result = tokio::time::timeout(Duration::from_secs(2), sweep)
+ .await
+ .expect("the sweep must end after the shutdown")
+ .expect("the sweep task must not panic");
+
+ assert!(
+ matches!(result, Err(IggyError::ClientShutdown)),
+ "got {result:?}"
+ );
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(
+ client.link.lock().await.is_none(),
+ "no link may be installed after the shutdown"
+ );
+ assert!(
+ tokio::time::timeout(Duration::from_millis(200), endpoint.accept())
+ .await
+ .is_err(),
+ "the sweep must not dial after the shutdown"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_cancelled_shutdown_is_finished_by_the_next_call() {
+ 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();
+ let mut events = client.subscribe_events().await;
+ {
+ let _transport = client.link.lock().await;
+ // The shutdown waits for the transport lock, and its caller gives
up.
+ assert!(
+ tokio::time::timeout(Duration::from_millis(100),
Client::shutdown(&client))
+ .await
+ .is_err()
+ );
+ }
+
+ Client::shutdown(&client).await.unwrap();
+
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(client.link.lock().await.is_none());
+ tokio::time::timeout(Duration::from_secs(1), peer.closed())
+ .await
+ .expect("the live connection must be closed");
+ let mut shutdowns = 0;
+ while let Ok(event) = events.try_recv() {
+ if event == DiagnosticEvent::Shutdown {
+ shutdowns += 1;
+ }
+ }
+ assert_eq!(shutdowns, 1);
+ }
}
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index bf89a3efa..c9837c902 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -115,6 +115,11 @@ pub struct TcpClient {
/// connection that drops without a `disconnect()` call leaves it clear, so
/// that loss still heals.
disconnected_by_caller: AtomicBool,
+ /// Set when `shutdown()` finishes its cleanup. The state turns `Shutdown`
+ /// before that cleanup and stays there, so it cannot tell a finished
+ /// shutdown from a cancelled one: a cancelled `shutdown()` leaves the
+ /// cleanup to the next call.
+ shutdown_complete: AtomicBool,
/// The session and the sign-in, shared with the poll router and the
/// poll clients it dials.
session_owner: Arc<SessionOwner>,
@@ -542,6 +547,7 @@ impl TcpClient {
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
disconnected_by_caller: AtomicBool::new(false),
+ shutdown_complete: AtomicBool::new(false),
session_owner,
skip_auto_login_once: Mutex::new(false),
routing_lock: Mutex::new(()),
@@ -664,19 +670,13 @@ impl TcpClient {
sleep(remaining.get_duration()).await;
}
- if self.caller_disconnected() {
+ if let Some(error) = self.sweep_interrupted().await {
self.fail_connect().await;
- return Err(IggyError::NotConnected);
+ return Err(error);
}
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
@@ -694,6 +694,15 @@ impl TcpClient {
connection.remote_address,
);
self.link.lock().await.replace(connection.link);
+ // Checked once the link is installed, not before:
+ // `shutdown()` sets `Shutdown` before it takes the
+ // link, so a shutdown during the dial either takes
+ // this link or is seen here.
+ if let Some(error) = self.sweep_interrupted().await {
+ self.link.lock().await.take();
+ self.fail_connect().await;
+ return Err(error);
+ }
self.set_state(ClientState::Connected).await;
self.connected_at.lock().await.replace(now);
self.publish_event(DiagnosticEvent::Connected).await;
@@ -742,9 +751,9 @@ impl TcpClient {
}
candidate = 0;
- if self.caller_disconnected() {
+ if let Some(error) = self.sweep_interrupted().await {
self.fail_connect().await;
- return Err(IggyError::NotConnected);
+ return Err(error);
}
// An unreadable CA file, a domain that will not parse: no
@@ -1211,6 +1220,11 @@ impl TcpClient {
/// left at `Connecting`, the next `connect()` returns ok at the top
/// without ever dialing.
async fn fail_connect(&self) {
+ // The state stays `Shutdown`, and a `Disconnected` event would
+ // contradict the `Shutdown` one.
+ if self.get_state().await == ClientState::Shutdown {
+ return;
+ }
self.set_state(ClientState::Disconnected).await;
self.publish_event(DiagnosticEvent::Disconnected).await;
}
@@ -1222,6 +1236,17 @@ impl TcpClient {
self.disconnected_by_caller.load(Ordering::SeqCst)
}
+ /// A reconnect sweep checks this before each dial, after it installs a
+ /// link and before each retry pause: once the client is shut down or the
+ /// caller disconnects, nothing in the sweep may bring the client back.
+ async fn sweep_interrupted(&self) -> Option<IggyError> {
+ if self.get_state().await == ClientState::Shutdown {
+ return Some(IggyError::ClientShutdown);
+ }
+ self.caller_disconnected()
+ .then_some(IggyError::NotConnected)
+ }
+
/// [`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
@@ -1299,19 +1324,22 @@ impl TcpClient {
}
async fn shutdown(&self) -> Result<(), IggyError> {
- if self.get_state().await == ClientState::Shutdown {
+ if self.shutdown_complete.load(Ordering::SeqCst) {
return Ok(());
}
let client_address = self.get_client_address_value().await;
info!("Shutting down the {NAME} TCP client: {client_address}");
+ // Set before the link is taken: a reconnect sweep that installs a link
+ // after this point sees `Shutdown` and drops it.
+ self.set_state(ClientState::Shutdown).await;
let link = self.link.lock().await.take();
if let Some(link) = link {
link.shutdown().await;
}
self.reset_vsr_session().await?;
- self.set_state(ClientState::Shutdown).await;
self.publish_event(DiagnosticEvent::Shutdown).await;
+ self.shutdown_complete.store(true, Ordering::SeqCst);
info!("{NAME} TCP client: {client_address} has been shutdown.");
Ok(())
}
@@ -3912,4 +3940,96 @@ mod tests {
"a publish waited for room that an unread receiver never frees"
);
}
+
+ #[tokio::test]
+ async fn a_shutdown_during_a_reconnect_sweep_ends_the_sweep() {
+ let address = dead_endpoint().await;
+ let client = Arc::new(
+ TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: address.clone(),
+ reconnection: TcpClientReconnectionConfig {
+ enabled: true,
+ max_retries: None,
+ interval:
NonZeroIggyDuration::from_str("500ms").expect("duration"),
+ reestablish_after: IggyDuration::new(Duration::ZERO),
+ },
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client"),
+ );
+ let sweep = tokio::spawn({
+ let client = Arc::clone(&client);
+ async move { TcpClient::connect(&client).await }
+ });
+ while client.get_state().await != ClientState::Connecting {
+ sleep(Duration::from_millis(10)).await;
+ }
+ // The first dial is refused at once, so the sweep now waits out its
+ // retry interval.
+ sleep(Duration::from_millis(100)).await;
+
+ Client::shutdown(&*client).await.unwrap();
+ // The server comes back before the next dial.
+ let listener = TcpListener::bind(&address).await.expect("bind again");
+ let result = tokio::time::timeout(Duration::from_secs(2), sweep)
+ .await
+ .expect("the sweep must end after the shutdown")
+ .expect("the sweep task must not panic");
+
+ assert!(
+ matches!(result, Err(IggyError::ClientShutdown)),
+ "got {result:?}"
+ );
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(
+ client.link.lock().await.is_none(),
+ "no link may be installed after the shutdown"
+ );
+ assert!(
+ tokio::time::timeout(Duration::from_millis(200), listener.accept())
+ .await
+ .is_err(),
+ "the sweep must not dial after the shutdown"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_cancelled_shutdown_is_finished_by_the_next_call() {
+ let (listener, address) = live_endpoint().await;
+ let client = TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: address,
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client");
+ let (accepted, connected) = tokio::join!(listener.accept(),
TcpClient::connect(&client));
+ connected.expect("connect");
+ let (mut peer, _) = accepted.expect("accept");
+ let mut events = client.subscribe_events().await;
+ {
+ let _transport = client.link.lock().await;
+ // The shutdown waits for the transport lock, and its caller gives
up.
+ assert!(
+ tokio::time::timeout(Duration::from_millis(100),
Client::shutdown(&client))
+ .await
+ .is_err()
+ );
+ }
+
+ Client::shutdown(&client).await.unwrap();
+
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(client.link.lock().await.is_none());
+ let mut byte = [0u8; 1];
+ let read = tokio::time::timeout(Duration::from_secs(1), peer.read(&mut
byte))
+ .await
+ .expect("the peer must see the connection close");
+ assert!(matches!(read, Ok(0) | Err(_)), "got {read:?}");
+ let mut shutdowns = 0;
+ while let Ok(event) = events.try_recv() {
+ if event == DiagnosticEvent::Shutdown {
+ shutdowns += 1;
+ }
+ }
+ assert_eq!(shutdowns, 1);
+ }
}
diff --git a/core/sdk/src/websocket/websocket_client.rs
b/core/sdk/src/websocket/websocket_client.rs
index 51b3b54c4..082f53386 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -104,6 +104,11 @@ pub struct WebSocketClient {
/// connection that drops without a `disconnect()` call leaves it clear, so
/// that loss still heals.
disconnected_by_caller: AtomicBool,
+ /// Set when `shutdown()` finishes its cleanup. The state turns `Shutdown`
+ /// before that cleanup and stays there, so it cannot tell a finished
+ /// shutdown from a cancelled one: a cancelled `shutdown()` leaves the
+ /// cleanup to the next call.
+ shutdown_complete: AtomicBool,
/// Serializes leader checks and roster walks after refused requests, so
/// concurrent callers cannot tear down each other's new connection.
routing_lock: Mutex<()>,
@@ -624,6 +629,7 @@ impl WebSocketClient {
roster_endpoints: Mutex::new(Vec::new()),
roster_learned: AtomicBool::new(false),
disconnected_by_caller: AtomicBool::new(false),
+ shutdown_complete: AtomicBool::new(false),
routing_lock: Mutex::new(()),
connect_coordinator: ConnectCoordinator::new(),
consumer_group_state:
Arc::new(iggy_common::ConsumerGroupClientState::new()),
@@ -689,6 +695,27 @@ impl WebSocketClient {
self.disconnected_by_caller.load(Ordering::SeqCst)
}
+ /// A reconnect sweep checks this before each dial and after it installs
+ /// the link: once the client is shut down or the caller disconnects,
+ /// nothing in the sweep may bring the client back.
+ async fn sweep_interrupted(&self) -> Option<IggyError> {
+ if self.get_state().await == ClientState::Shutdown {
+ return Some(IggyError::ClientShutdown);
+ }
+ self.caller_disconnected()
+ .then_some(IggyError::NotConnected)
+ }
+
+ async fn fail_connect(&self) {
+ // The state stays `Shutdown`, and a `Disconnected` event would
+ // contradict the `Shutdown` one.
+ if self.get_state().await == ClientState::Shutdown {
+ return;
+ }
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ }
+
async fn connect_inner(&self, context: ConnectOwnerContext) -> Result<(),
IggyError> {
let settle_off_leader = context.settle_off_leader();
let single_attempt = context.single_attempt();
@@ -743,10 +770,9 @@ impl WebSocketClient {
}
}
- if self.caller_disconnected() {
- self.set_state(ClientState::Disconnected).await;
- self.publish_event(DiagnosticEvent::Disconnected).await;
- return Err(IggyError::NotConnected);
+ if let Some(error) = self.sweep_interrupted().await {
+ self.fail_connect().await;
+ return Err(error);
}
let server_addr = tokio::net::lookup_host(&*current_address)
@@ -798,14 +824,15 @@ impl WebSocketClient {
Err(error) => return Err(error),
};
- if self.caller_disconnected() {
- drop(link);
- self.set_state(ClientState::Disconnected).await;
- self.publish_event(DiagnosticEvent::Disconnected).await;
- return Err(IggyError::NotConnected);
- }
-
*self.link.lock().await = Some(link);
+ // Checked once the link is installed, not before: `shutdown()`
+ // sets `Shutdown` before it takes the link, so a shutdown
+ // during the dial either takes this link or is seen here.
+ if let Some(error) = self.sweep_interrupted().await {
+ self.link.lock().await.take();
+ self.fail_connect().await;
+ return Err(error);
+ }
*self.client_address.lock().await = Some(server_addr);
self.set_state(ClientState::Connected).await;
*self.connected_at.lock().await = Some(IggyTimestamp::now());
@@ -1182,14 +1209,16 @@ impl WebSocketClient {
}
async fn shutdown(&self) -> Result<(), IggyError> {
- if self.get_state().await == ClientState::Shutdown {
+ if self.shutdown_complete.load(Ordering::SeqCst) {
return Ok(());
}
let client_address = self.get_client_address_value().await;
info!("Shutting down the {NAME} client: {client_address}");
- self.set_state(ClientState::Disconnected).await;
+ // Set before the link is taken: a reconnect sweep that installs a link
+ // after this point sees `Shutdown` and drops it.
+ self.set_state(ClientState::Shutdown).await;
let link = self.link.lock().await.take();
if let Some(link) = link {
@@ -1197,8 +1226,8 @@ impl WebSocketClient {
}
self.reset_vsr_session().await?;
- self.set_state(ClientState::Shutdown).await;
self.publish_event(DiagnosticEvent::Shutdown).await;
+ self.shutdown_complete.store(true, Ordering::SeqCst);
info!("{NAME} client: {client_address} has been shutdown.");
Ok(())
}
@@ -2461,4 +2490,103 @@ mod tests {
"a publish waited for room that an unread receiver never frees"
);
}
+
+ #[tokio::test]
+ async fn a_shutdown_during_a_reconnect_sweep_ends_the_sweep() {
+ let dead_listener =
tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let address = dead_listener.local_addr().unwrap().to_string();
+ drop(dead_listener);
+ let client = Arc::new(
+ WebSocketClient::create(Arc::new(WebSocketClientConfig {
+ server_address: address.clone(),
+ reconnection: WebSocketClientReconnectionConfig {
+ enabled: true,
+ max_retries: None,
+ interval: NonZeroIggyDuration::from_str("500ms").unwrap(),
+ reestablish_after: IggyDuration::from_str("0s").unwrap(),
+ },
+ ..Default::default()
+ }))
+ .unwrap(),
+ );
+ let sweep = tokio::spawn({
+ let client = Arc::clone(&client);
+ async move { WebSocketClient::connect(&client).await }
+ });
+ // The first dial is refused at once, so the sweep now waits out its
+ // retry interval.
+ sleep(Duration::from_millis(100)).await;
+
+ Client::shutdown(&*client).await.unwrap();
+ // The server comes back before the next dial.
+ let listener = tokio::net::TcpListener::bind(&address).await.unwrap();
+ let result = tokio::time::timeout(Duration::from_secs(2), sweep)
+ .await
+ .expect("the sweep must end after the shutdown")
+ .expect("the sweep task must not panic");
+
+ assert!(
+ matches!(result, Err(IggyError::ClientShutdown)),
+ "got {result:?}"
+ );
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(
+ client.link.lock().await.is_none(),
+ "no link may be installed after the shutdown"
+ );
+ assert!(
+ tokio::time::timeout(Duration::from_millis(200), listener.accept())
+ .await
+ .is_err(),
+ "the sweep must not dial after the shutdown"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_cancelled_shutdown_is_finished_by_the_next_call() {
+ let listener =
tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let client = WebSocketClient::create(Arc::new(WebSocketClientConfig {
+ server_address: listener.local_addr().unwrap().to_string(),
+ ..Default::default()
+ }))
+ .unwrap();
+ let (mut peer, connected) = tokio::join!(
+ async {
+ accept_async(listener.accept().await.unwrap().0)
+ .await
+ .unwrap()
+ },
+ Client::connect(&client)
+ );
+ connected.unwrap();
+ let mut events = client.subscribe_events().await;
+ {
+ let _transport = client.link.lock().await;
+ // The shutdown waits for the transport lock, and its caller gives
up.
+ assert!(
+ tokio::time::timeout(Duration::from_millis(100),
Client::shutdown(&client))
+ .await
+ .is_err()
+ );
+ }
+
+ Client::shutdown(&client).await.unwrap();
+
+ assert_eq!(client.get_state().await, ClientState::Shutdown);
+ assert!(client.link.lock().await.is_none());
+ let next = tokio::time::timeout(Duration::from_secs(1), peer.next())
+ .await
+ .expect("the peer must see the connection close");
+ assert!(
+ matches!(next, None | Some(Ok(Message::Close(_))) | Some(Err(_))),
+ "got {next:?}"
+ );
+ let mut shutdowns = 0;
+ while let Ok(event) = events.try_recv() {
+ if event == DiagnosticEvent::Shutdown {
+ shutdowns += 1;
+ }
+ }
+ assert_eq!(shutdowns, 1);
+ }
}
