Reconnect L1 WS stream if disconnected (#2287)

* Reconnect L1 WS stream if disconnected

Ethers internally tries to reconnect once, but if that fails, it
gives up and shuts down the whole provider backend. This is not
good in the case of extended outages of the L1 server.

This change handles errors from the underlying RPC client indicating
that the connection is closed by spawning a task to reestablish the
connection. This task tries until it succceeds.

* Cancel reconnect task if L1 client is shut down

* Rename L1 client lifecycle methods

sequencer/src/context.rs

@@ -111,7 +111,7 @@ impl<N: ConnectedNetwork<PubKey>, P: SequencerPersistence, V: Versions> Sequence
             .set(instance_state.node_id as usize);
 
         // Start L1 client if it isn't already.
-        instance_state.l1_client.start().await;
+        instance_state.l1_client.spawn_tasks().await;
 
         // Load saved consensus state from storage.
         let (initializer, anchor_view) = persistence
@@ -325,7 +325,7 @@ impl<N: ConnectedNetwork<PubKey>, P: SequencerPersistence, V: Versions> Sequence
         tracing::info!("shutting down SequencerContext");
         self.handle.write().await.shut_down().await;
         self.tasks.shut_down();
-        self.node_state.l1_client.stop().await;
+        self.node_state.l1_client.shut_down_tasks().await;
 
         // Since we've already shut down, we can set `detached` so the drop
         // handler doesn't call `shut_down` again.
@@ -366,7 +366,7 @@ impl<N: ConnectedNetwork<PubKey>, P: SequencerPersistence, V: Versions> Drop
                 tracing::info!("shutting down SequencerContext");
                 handle_clone.write().await.shut_down().await;
                 tasks_clone.shut_down();
-                node_state_clone.l1_client.stop().await;
+                node_state_clone.l1_client.shut_down_tasks().await;
             });
 
             // Set `detached` so the drop handler doesn't call `shut_down` again.

sequencer/src/lib.rs

@@ -492,7 +492,7 @@ pub async fn init_node<P: PersistenceOptions, V: Versions>(
     }
 
     let l1_client = l1_params.options.connect(l1_params.url).await?;
-    l1_client.start().await;
+    l1_client.spawn_tasks().await;
     let l1_genesis = match genesis.l1_finalized {
         L1Finalized::Block(b) => b,
         L1Finalized::Number { number } => l1_client.wait_for_finalized_block(number).await,

types/Cargo.toml

@@ -54,5 +54,8 @@ tracing = { workspace = true }
 url = { workspace = true }
 vbs = { workspace = true }
 
+[dev-dependencies]
+portpicker = { workspace = true }
+
 [package.metadata.cargo-machete]
 ignored = ["base64_bytes", "hotshot_testing"]