fix(connection): retry failed session establishment (#761)

* fix(connection): retry failed session establishment

- Retry when TCP reconnects before the TWS session is ready
- Cover sync and async paths with transient handshake tests

* fix(connection): surface last reconnect error and add changelog entry

Exhausted reconnect attempts now return the final attempt's error instead
of a bare Error::ConnectionFailed, so a permanent cause (e.g. incompatible
server version) is named. Dispatchers still map any reconnect failure to
the ConnectionFailed sentinel, so loop-exit behavior is unchanged.

---------

Co-authored-by: Wil Boayue <wil.boayue@gmail.com>

Author: faysou

CHANGELOG.md

@@ -52,6 +52,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `OrderBuilder::analyze()` (what-if orders, blocking and async) now returns the TWS rejection instead of `Error::UnexpectedEndOfStream`. A rejected what-if order arrives as a routed error, which the response read discarded — the blocking path via `if let Ok(..)` inside the loop, the async path by ending its `while let Some(Ok(..))` loop — so the caller lost the reason (e.g. code 201, `Order rejected - reason:...`) and got a generic end-of-stream error. Rejection is a routine outcome for a what-if order, so this was the likeliest path to hit it (#735).
 
+- Reconnection now survives the window where TWS accepts the TCP connection but its API handshake is not yet ready — common during an automated restart. A session-establishment failure (handshake, `startAPI`, account info) used to abort the reconnect loop on the first occurrence; it now consumes one attempt and follows the same fibonacci backoff as a socket failure, on both the blocking and async clients. When every attempt fails, `reconnect` returns the last attempt's error instead of a bare `Error::ConnectionFailed`, so a permanent cause — say, an incompatible server version — is named rather than hidden behind a generic failure (#761).
+
 - `matching_symbols()` on the blocking client now returns the TWS error instead of an empty list. A routed error arrives as `Some(Err(_))`, which the `if let Some(Ok(_))` read discarded, so a rejected pattern silently returned `Ok(vec![])` — indistinguishable from "no symbols matched". The async client already propagated it (#735).
 
 - `TickTypes::MarketDataType` now reaches `Client::market_data` subscriptions. The message type was missing from the request-id routing allow-list, so TWS's market-data-type notifications (real-time / frozen / delayed / delayed-frozen, sent on subscribe and whenever the feed switches) were routed to a shared channel nobody subscribes to and dropped. The decoder has produced the variant since #516; nothing could ever yield it (#730).

src/connection/async.rs

@@ -128,9 +128,12 @@ impl<S: AsyncStream> AsyncConnection<S> {
     }
 
     /// Reconnect to TWS with fibonacci backoff. Replays the handshake and
-    /// re-fires the persisted startup / notice callbacks.
+    /// re-fires the persisted startup / notice callbacks. When every attempt
+    /// fails, returns the last attempt's error — a permanent cause (say, an
+    /// incompatible server version) must not exit as a generic failure.
     pub async fn reconnect(&self) -> Result<(), Error> {
         let mut backoff = FibonacciBackoff::new(30);
+        let mut last_error = None;
 
         for i in 0..MAX_RECONNECT_ATTEMPTS {
             let next_delay = backoff.next_delay();
@@ -140,18 +143,30 @@ impl<S: AsyncStream> AsyncConnection<S> {
 
             match self.socket.reconnect().await {
                 Ok(_) => {
-                    info!("reconnected !!!");
                     self.reset_connection_metadata().await;
-                    self.establish_connection().await?;
-                    return Ok(());
+                    match self.establish_connection().await {
+                        Ok(()) => {
+                            info!("reconnected");
+                            return Ok(());
+                        }
+                        Err(e) => {
+                            info!(
+                                "reconnection attempt {}/{} failed while establishing session: {e}",
+                                i + 1,
+                                MAX_RECONNECT_ATTEMPTS
+                            );
+                            last_error = Some(e);
+                        }
+                    }
                 }
                 Err(e) => {
                     info!("reconnection attempt {}/{} failed: {e}", i + 1, MAX_RECONNECT_ATTEMPTS);
+                    last_error = Some(e);
                 }
             }
         }
 
-        Err(Error::ConnectionFailed)
+        Err(last_error.unwrap_or(Error::ConnectionFailed))
     }
 
     async fn reset_connection_metadata(&self) {

src/connection/async_tests.rs

@@ -210,12 +210,33 @@ async fn reconnect_succeeds_after_transient_failures() {
     assert_eq!(connection.server_version(), SERVER_VERSION);
 }
 
+#[tokio::test]
+async fn reconnect_retries_after_transient_handshake_failure() {
+    let stream = MemoryStream::default();
+    let connection = AsyncConnection::stubbed(stream.clone(), CLIENT_ID);
+
+    push_handshake(&stream);
+    connection.establish_connection().await.expect("initial establish_connection failed");
+
+    let too_old = server_versions::PROTOBUF_REST_MESSAGES_3 - 1;
+    stream.push_inbound(format!("{}\020240120 12:00:00 EST\0", too_old).into_bytes());
+    push_handshake(&stream);
+
+    connection.reconnect().await.expect("reconnect must retry a failed handshake");
+
+    assert_eq!(connection.server_version(), SERVER_VERSION);
+    let metadata = connection.connection_metadata().await;
+    assert_eq!(metadata.next_order_id, 90);
+    assert_eq!(metadata.managed_accounts, "DU1234567");
+}
+
 /// When the socket refuses reconnects through every Fibonacci attempt, the
-/// loop exits with `Error::ConnectionFailed`. Pre-arming with exactly
+/// loop exits with the *last attempt's* error — not a generic
+/// `Error::ConnectionFailed` that hides the cause. Pre-arming with exactly
 /// `MAX_RECONNECT_ATTEMPTS` failures binds the test to the loop's exit
 /// condition (rather than a hardcoded count).
 #[tokio::test]
-async fn reconnect_returns_connection_failed_after_exhausting_attempts() {
+async fn reconnect_returns_last_error_after_exhausting_attempts() {
     let stream = MemoryStream::default();
     let connection = AsyncConnection::stubbed(stream.clone(), CLIENT_ID);
 
@@ -225,7 +246,10 @@ async fn reconnect_returns_connection_failed_after_exhausting_attempts() {
     stream.set_reconnect_failures(MAX_RECONNECT_ATTEMPTS as usize);
 
     let err = connection.reconnect().await.expect_err("must give up after MAX_RECONNECT_ATTEMPTS");
-    assert!(matches!(err, crate::errors::Error::ConnectionFailed), "got {err:?}");
+    assert!(
+        matches!(&err, crate::errors::Error::Simple(msg) if msg == "simulated reconnect failure"),
+        "got {err:?}"
+    );
 }
 
 /// During a reconnect, any caller of `connection_metadata()` must see cleared

src/connection/sync.rs

@@ -111,9 +111,12 @@ impl<S: Stream> Connection<S> {
     }
 
     /// Reconnect to TWS with fibonacci backoff. Replays the handshake and
-    /// re-fires the persisted startup / notice callbacks.
+    /// re-fires the persisted startup / notice callbacks. When every attempt
+    /// fails, returns the last attempt's error — a permanent cause (say, an
+    /// incompatible server version) must not exit as a generic failure.
     pub fn reconnect(&self) -> Result<(), Error> {
         let mut backoff = FibonacciBackoff::new(30);
+        let mut last_error = None;
 
         for i in 0..self.max_retries {
             let next_delay = backoff.next_delay();
@@ -123,19 +126,30 @@ impl<S: Stream> Connection<S> {
 
             match self.socket.reconnect() {
                 Ok(_) => {
-                    info!("reconnected !!!");
                     self.reset_connection_metadata();
-                    self.establish_connection()?;
-
-                    return Ok(());
+                    match self.establish_connection() {
+                        Ok(()) => {
+                            info!("reconnected");
+                            return Ok(());
+                        }
+                        Err(e) => {
+                            info!(
+                                "reconnection attempt {}/{} failed while establishing session: {e}",
+                                i + 1,
+                                self.max_retries
+                            );
+                            last_error = Some(e);
+                        }
+                    }
                 }
                 Err(e) => {
                     info!("reconnection attempt {}/{} failed: {e}", i + 1, self.max_retries);
+                    last_error = Some(e);
                 }
             }
         }
 
-        Err(Error::ConnectionFailed)
+        Err(last_error.unwrap_or(Error::ConnectionFailed))
     }
 
     fn reset_connection_metadata(&self) {

src/connection/sync_tests.rs

@@ -77,6 +77,26 @@ fn establish_connection_populates_metadata() {
     assert_eq!(metadata.time_zone, Some(timezones::db::EST));
 }
 
+#[test]
+fn reconnect_retries_after_transient_handshake_failure() {
+    let stream = MemoryStream::default();
+    let connection = Connection::stubbed(stream.clone(), CLIENT_ID);
+
+    push_handshake(&stream);
+    connection.establish_connection().expect("initial establish_connection failed");
+
+    let too_old = server_versions::PROTOBUF_REST_MESSAGES_3 - 1;
+    stream.push_inbound(format!("{}\020240120 12:00:00 EST\0", too_old).into_bytes());
+    push_handshake(&stream);
+
+    connection.reconnect().expect("reconnect must retry a failed handshake");
+
+    assert_eq!(connection.server_version(), SERVER_VERSION);
+    let metadata = connection.connection_metadata();
+    assert_eq!(metadata.next_order_id, 90);
+    assert_eq!(metadata.managed_accounts, "DU1234567");
+}
+
 #[test]
 fn disconnect_completes() {
     let (client, stream) = make_client();

src/transport/sync_tests.rs

@@ -401,9 +401,11 @@ fn test_reconnect_failed() -> Result<(), Error> {
 
     let _ = connection.read_message();
 
+    // Exhausted attempts surface the last attempt's error, not a generic
+    // `Error::ConnectionFailed`.
     match connection.reconnect() {
-        Err(Error::ConnectionFailed) => Ok(()),
-        _ => panic!(""),
+        Err(Error::Io(e)) if e.kind() == ErrorKind::ConnectionRefused => Ok(()),
+        other => panic!("expected the last reconnect error, got {other:?}"),
     }
 }