hubcio commented on code in PR #4388:
URL: https://github.com/apache/iggy/pull/4388#discussion_r4248001219
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -3793,4 +3837,118 @@ 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() {
+ let (listener, address) = live_endpoint().await;
Review Comment:
simplification: this test repeats the listener and counter from
`counted_endpoint_that_hangs_up()`. reuse that helper.
##########
core/sdk/src/clients/client.rs:
##########
@@ -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();
Review Comment:
warning: aborting the heartbeat can stop QUIC cleanup after it sets
`Disconnected`, so the next cleanup skips a live connection. finish transport
cleanup even when the state is already `Disconnected`.
--
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]
