fix(stream): enable public reconnect() with needs_reconnect() flag

[Canvinus]

Summary

Enables the existing reconnection logic by making it public and adding a flag to signal when reconnection is needed.

Changes

• Add needs_reconnect field to WebSocketStream
• Make reconnect() public with exponential backoff (1s→60s, 2x multiplier, max 5 retries)
• Add needs_reconnect() getter for consumers to check reconnection status
• Set needs_reconnect=true on Close frame, WS error, or stream end
• Reset needs_reconnect=false on successful reconnection
• Auto-resubscribe to all previous subscriptions on reconnect

Usage

// In consumer code
if ws_stream.needs_reconnect() {
    if let Err(e) = ws_stream.reconnect().await {
        error!("Reconnection failed: {}", e);
    }
}

[Copilot]

Pull request overview

Enables consumers of WebSocketStream to detect when a WebSocket connection has been lost and explicitly trigger reconnection, including automatic resubscription.

Changes:

• Add a needs_reconnect: bool flag to WebSocketStream, with a public needs_reconnect() getter.
• Make reconnect() public and implement exponential backoff retries.
• Set needs_reconnect = true when the websocket closes, errors, or ends; reset it on successful reconnection.

Comments suppressed due to low confidence (3)

src/stream.rs:552

• New reconnection behavior (needs_reconnect() + public reconnect()) isn’t covered by the existing unit tests in this module. Adding tests that (a) set needs_reconnect when the stream receives Close / Err / None, and (b) verify the flag is only cleared after a successful reconnect+resubscribe, would help prevent regressions. If direct websocket I/O is hard to test, consider extracting the resubscribe/flag-update logic behind a small injectable trait or helper that can be unit-tested.

    /// Check if reconnection is needed (connection was lost)
    pub fn needs_reconnect(&self) -> bool {
        self.needs_reconnect
    }

    /// Reconnect with exponential backoff
    /// 
    /// Call this method when `needs_reconnect()` returns true.
    /// It will attempt to reconnect to the WebSocket and resubscribe to all previous subscriptions.
    pub async fn reconnect(&mut self) -> Result<()> {
        let mut delay = self.reconnect_config.base_delay;
        let mut retries = 0;

        while retries < self.reconnect_config.max_retries {
            warn!("Attempting to reconnect (attempt {})", retries + 1);

            match self.connect().await {
                Ok(()) => {
                    info!("Successfully reconnected");
                    self.stats.reconnect_count += 1;
                    self.needs_reconnect = false;  // Reset flag on success

                    // Resubscribe to all previous subscriptions
                    let subscriptions = self.subscriptions.clone();
                    for subscription in subscriptions {
                        self.send_message(serde_json::to_value(subscription)?)
                            .await?;
                    }

                    return Ok(());
                },
                Err(e) => {
                    error!("Reconnection attempt {} failed: {}", retries + 1, e);
                    retries += 1;

                    if retries < self.reconnect_config.max_retries {
                        tokio::time::sleep(delay).await;
                        delay = std::cmp::min(
                            delay.mul_f64(self.reconnect_config.backoff_multiplier),
                            self.reconnect_config.max_delay,
                        );
                    }
                },
            }
        }

        Err(PolyfillError::stream(
            format!(
                "Failed to reconnect after {} attempts",
                self.reconnect_config.max_retries
            ),
            crate::errors::StreamErrorKind::ConnectionFailed,
        ))
    }
}

src/stream.rs:525

• reconnect() clears needs_reconnect before the resubscribe loop. If send_message(...) (or serialization) fails mid-resubscribe, the method returns Err but the stream is left with needs_reconnect = false, which is inconsistent with a failed reconnect/resubscribe attempt. Consider only resetting the flag (and incrementing reconnect_count) after the full reconnect+resubscribe sequence succeeds; on failure, keep needs_reconnect true (and optionally close the new connection).

                Ok(()) => {
                    info!("Successfully reconnected");
                    self.stats.reconnect_count += 1;
                    self.needs_reconnect = false;  // Reset flag on success

                    // Resubscribe to all previous subscriptions
                    let subscriptions = self.subscriptions.clone();
                    for subscription in subscriptions {
                        self.send_message(serde_json::to_value(subscription)?)
                            .await?;
                    }

src/stream.rs:525

• After a disconnect, pending_books may still contain messages from the prior connection, and poll_next will yield them before reading from the new socket (it checks pending_books first). With the new public reconnect(), this can cause stale/out-of-order messages to be delivered after reconnect. Consider clearing pending_books (and any other per-connection transient state) when the connection is lost and/or at the start of reconnect().

                    // Resubscribe to all previous subscriptions
                    let subscriptions = self.subscriptions.clone();
                    for subscription in subscriptions {
                        self.send_message(serde_json::to_value(subscription)?)
                            .await?;
                    }

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The reconnect() rustdoc has an empty doc line (/// ) and doesn’t mention that poll_next returns None on Close/stream end (terminating typical while let Some(...) consumption). Consider tightening the rustdoc (remove the blank doc line) and documenting the expected consumer pattern (e.g., handle None/end-of-stream, then call reconnect() and resume polling).

[Copilot]

needs_reconnect is only reset inside reconnect(). If the stream re-establishes a connection via another path that calls connect() (e.g., subscribe_async when connection.is_none()), needs_reconnect() can continue to return true even after a successful connect. Consider resetting the flag on any successful (re)connection (e.g., in connect()), or ensuring all reconnection entry points funnel through reconnect().