Repository: gakonst/ethers-rs
Pull request #2419 (2023-05-15 04:07:28)

fix: fix re-subscription on websocket reconnect
Add an in-flight request for each re-subscription request to ensure that the response from the server with the new internal subscription id is handled.

## Motivation

This change fixes issue 2418, which is an issue where websocket subscriptions can stop working after a websocket reconnect.

## Solution

The underlying issue is that the re-subscription code did not have an in-flight request for the resubscription requests. They were issued properly, but the response from the server containing the new server-side subscription ID was then dropped as a result. This change adds the appropriate in-flight requests so the new server-side ID is handled properly and updated.

## PR Checklist

-   [ x] Added Tests
-   [ ] Added Documentation
-   [ ] Breaking changes


gbrew: Thanks for the review @prestwich , and thanks for all the work on reconnects!

I had a quick look at the failing CI tests, and they all seem likely related to a solc version change ( ethers-solc compile::tests::test_detect_version ), and unrelated to my change. Should I worry about making those pass or just ignore them? And is there anything else I should be doing to get this merged?
prestwich: > Should I worry about making those pass or just ignore them? And is there anything else I should be doing to get this merged?

just ignore those 👍  I'll be doing a pass later today to merge things and hopefully get those addressed as well

Patch:
diff --git a/ethers-providers/src/rpc/transports/ws/manager.rs b/ethers-providers/src/rpc/transports/ws/manager.rs
index ff38b53d..c685da4c 100644
--- a/ethers-providers/src/rpc/transports/ws/manager.rs
+++ b/ethers-providers/src/rpc/transports/ws/manager.rs
@@ -351,25 +351,29 @@ impl RequestManager {
         old_backend.shutdown();
 
         tracing::debug!(count = self.subs.count(), "Re-starting active subscriptions");
+        let req_cnt = self.reqs.len();
 
-        // reissue subscriptionps
+        // reissue subscriptions
         for (id, sub) in self.subs.to_reissue() {
-            self.backend
-                .dispatcher
-                .unbounded_send(sub.serialize_raw(*id)?)
-                .map_err(|_| WsClientError::DeadChannel)?;
+            let (tx, _rx) = oneshot::channel();
+            let in_flight = InFlight {
+                method: "eth_subscribe".to_string(),
+                params: sub.params.clone(),
+                channel: tx,
+            };
+            // Need an entry in reqs to ensure response with new server sub ID is processed
+            self.reqs.insert(*id, in_flight);
         }
 
-        tracing::debug!(count = self.reqs.len(), "Re-issuing pending requests");
-        // reissue requests. We filter these to prevent in-flight requests for
-        // subscriptions to be re-issued twice (once in above loop, once in this loop).
-        for (id, req) in self.reqs.iter().filter(|(id, _)| !self.subs.has(**id)) {
+        tracing::debug!(count = req_cnt, "Re-issuing pending requests");
+        // reissue requests, including the re-subscription requests we just added above
+        for (id, req) in self.reqs.iter() {
             self.backend
                 .dispatcher
                 .unbounded_send(req.serialize_raw(*id)?)
                 .map_err(|_| WsClientError::DeadChannel)?;
         }
-        tracing::info!(subs = self.subs.count(), reqs = self.reqs.len(), "Re-connection complete");
+        tracing::info!(subs = self.subs.count(), reqs = req_cnt, "Re-connection complete");
 
         Ok(())
     }


Test patch:
diff --git a/ethers-providers/tests/it/ws_errors.rs b/ethers-providers/tests/it/ws_errors.rs
index f4eced17..168b24e9 100644
--- a/ethers-providers/tests/it/ws_errors.rs
+++ b/ethers-providers/tests/it/ws_errors.rs
@@ -1,4 +1,4 @@
-use ethers_core::types::Filter;
+use ethers_core::{types::Filter, utils::Anvil};
 use ethers_providers::{Middleware, Provider, StreamExt};
 use futures_util::SinkExt;
 use std::time::Duration;
@@ -64,3 +64,26 @@ async fn graceful_disconnect_on_ws_errors() {
 
     assert!(stream.next().await.is_none());
 }
+
+#[tokio::test]
+async fn resubscribe_on_ws_reconnect() {
+    let anvil = Anvil::new().block_time(1u64).spawn();
+    let port = anvil.port();
+    let provider = Provider::connect_with_reconnects(anvil.ws_endpoint(), 1).await.unwrap();
+
+    // Attempt to ensure a different server-side subscription id after reconnect by making
+    // the subscription we care about be the second one after initial startup, but the first
+    // (and only) one after reconnection.
+    let ignored_sub = provider.subscribe_blocks().await.unwrap();
+    let mut blocks = provider.subscribe_blocks().await.unwrap();
+    ignored_sub.unsubscribe().await.expect("unsubscribe failed");
+
+    blocks.next().await.expect("no block notice before reconnect");
+
+    // Kill & restart using the same port so we end up with the same endpoint url:
+    drop(anvil);
+    let _anvil = Anvil::new().port(port).block_time(1u64).spawn();
+
+    // Wait for the next block on existing subscription. Will fail w/o resubscription:
+    blocks.next().await.expect("no block notice after reconnect");
+}