Address the second round of OpenCodeReview findings

Accepted:

* scanner-daemon: replace the implicit 'outage exists iff client is
  None' bookkeeping with a ConnectionState enum (Connected/Reconnecting)
  so the invariant cannot be violated; give the storage-version panic a
  contextual message; log the local-tip read failure during the recovery
  log instead of silently resuming 'from height 0'; validate that the
  backoff maximum is not smaller than its initial delay.

* web-server: the outage starts at the first failed round (the failure
  logs now show the real elapsed time instead of zero), and a stalled
  event stream no longer forces a full reconnection: the bridge first
  retries the subscription on the existing connection and only drops the
  client when that fails with a connection-level error. The lag advisory
  is broadcast whenever an outage is ongoing, including stall and decode
  failure rounds.

* tests: the test proxy reaps dead connection pairs (either direction
  finished), counts refused connections (refused_connections()), keeps
  its task handle observable via assert_alive(), and the scanner
  reconnect test builds its chain in spawn_blocking (no runtime
  blocking), derives deterministic child seeds from one master RNG, and
  replaces the 500ms pre-abort sleep with a proper abort+join.
  scanner_tip logging: polled storage errors are no longer fully
  indistinguishable from the uninitialized state (last error is visible
  in the timeout message).

Rejected:

* 'send_event stale sink race': after wait_for_subscription, the last
  sink is by construction the freshly opened one (the backend pushes
  exactly once per accepted subscription); the is_closed assert is the
  precise guard against ordering regressions, a retry loop would mask
  them.

Author: erubboli

api-server/scanner-daemon/src/backoff.rs

@@ -49,6 +49,11 @@ impl ReconnectBackoff {
             Duration::ZERO,
             "The maximum delay must not be zero"
         );
+        assert!(
+            max_delay >= initial_delay,
+            "The maximum delay ({max_delay:?}) must not be smaller than the initial delay \
+            ({initial_delay:?})"
+        );
 
         Self {
             initial_delay,

api-server/scanner-daemon/src/lib.rs

@@ -123,7 +123,10 @@ pub async fn run<S: ApiServerStorage>(
                         .get_storage_version()
                         .await
                         .unwrap_or_else(|e| panic!("Storage version read failed {}", e))
-                        .expect("cannot be empty");
+                        .expect(
+                            "storage is initialized but has no storage version row (the row \
+                             is missing or the database is corrupt)",
+                        );
                     if storage_version != CURRENT_STORAGE_VERSION {
                         true
                     } else {
@@ -184,6 +187,13 @@ async fn reinitialize_and_rescan<S: ApiServerStorage>(
     local_block
 }
 
+/// The connection state of the supervision loop; the `Connected`/`Reconnecting` variants make
+/// the invariant "an outage tracker exists if and only if there is no client" explicit.
+enum ConnectionState {
+    Connected(NodeRpcClient),
+    Reconnecting { outage: Outage },
+}
+
 /// The supervision loop of the scanner: keeps the local state in sync with the node, recovering
 /// in place from connection-level failures by re-creating the RPC client with an exponential
 /// backoff and resuming the indexing from the tip stored in the local state.
@@ -206,80 +216,91 @@ async fn supervise_sync<S: ApiServerStorage>(
     rng: &mut impl Rng,
 ) -> Result<(), ApiServerScannerError> {
     // Note: the client is created lazily (and re-created after every connection-level failure),
-    // so that a node that is down at startup does not abort the daemon; `None` means that the
-    // connection to the node is currently broken and must be re-established.
-    let mut rpc_client: Option<NodeRpcClient> = None;
-    // Note: `outage` is `Some` if and only if `rpc_client` is `None`.
-    let mut outage: Option<Outage> = None;
+    // so that a node that is down at startup does not abort the daemon.
+    let mut state = ConnectionState::Reconnecting {
+        outage: Outage::new(),
+    };
 
     loop {
-        if rpc_client.is_none() {
-            rpc_client = Some(loop {
-                // Note: the connection attempt is bounded by a timeout so that a stalled
-                // handshake (the node accepting the TCP connection but never completing the
-                // WebSocket handshake) cannot wedge a round forever without any log output;
-                // expiry is treated like any other connection failure.
-                let connection = tokio::time::timeout(
-                    CONNECT_TIMEOUT,
-                    make_rpc_client(
-                        Arc::clone(chain_config),
-                        node_rpc_address.clone(),
-                        node_rpc_auth.clone(),
-                    ),
-                )
-                .await;
-                match connection {
-                    Ok(Ok(client)) => break client,
-                    Ok(Err(err)) if NodeRpcError::is_connection_error(&err) => {
-                        let outage = outage.get_or_insert_with(Outage::new);
-                        outage.attempts += 1;
-                        let delay = backoff.next_delay(rng);
-                        logging::log::warn!(
-                            "Failed to connect to the node (attempt {}, elapsed {:?}): {err}; \
-                            retrying in {delay:?}",
-                            outage.attempts,
-                            outage.started.elapsed(),
-                        );
-                        tokio::time::sleep(delay).await;
-                    }
-                    // A connection cannot be established, but the reason is not a temporary
-                    // connectivity problem (e.g. an invalid address); retrying is pointless.
-                    Ok(Err(err)) => return Err(ApiServerScannerError::RpcError(err)),
-                    Err(_timed_out) => {
-                        let outage = outage.get_or_insert_with(Outage::new);
-                        outage.attempts += 1;
-                        let delay = backoff.next_delay(rng);
-                        logging::log::warn!(
-                            "Timed out connecting to the node (attempt {}, elapsed {:?}); \
-                            retrying in {delay:?}",
-                            outage.attempts,
-                            outage.started.elapsed(),
-                        );
-                        tokio::time::sleep(delay).await;
+        let client = loop {
+            match &mut state {
+                ConnectionState::Connected(client) => break client.clone(),
+                ConnectionState::Reconnecting { outage } => {
+                    // Note: the connection attempt is bounded by a timeout so that a stalled
+                    // handshake (the node accepting the TCP connection but never completing
+                    // the WebSocket handshake) cannot wedge a round forever without any log
+                    // output; expiry is treated like any other connection failure.
+                    let connection = tokio::time::timeout(
+                        CONNECT_TIMEOUT,
+                        make_rpc_client(
+                            Arc::clone(chain_config),
+                            node_rpc_address.clone(),
+                            node_rpc_auth.clone(),
+                        ),
+                    )
+                    .await;
+                    match connection {
+                        // Note: the recovery is only logged when the outage actually produced
+                        // failed attempts; the very first connection is not a "recovery".
+                        Ok(Ok(new_client)) if outage.attempts > 0 => {
+                            let outage = std::mem::replace(outage, Outage::new());
+                            let local_height = match local_block.best_block().await {
+                                Ok((height, _)) => height,
+                                Err(err) => {
+                                    logging::log::warn!(
+                                        "Failed to read the local tip for the recovery log: {err}"
+                                    );
+                                    common::primitives::BlockHeight::zero()
+                                }
+                            };
+                            logging::log::info!(
+                                "Scanner reconnected to the node after {} attempt(s) ({:?} \
+                                elapsed); resuming from height {}",
+                                outage.attempts,
+                                outage.started.elapsed(),
+                                local_height,
+                            );
+                            break new_client;
+                        }
+                        Ok(Ok(new_client)) => break new_client,
+                        Ok(Err(err)) if NodeRpcError::is_connection_error(&err) => {
+                            outage.attempts += 1;
+                            let delay = backoff.next_delay(rng);
+                            logging::log::warn!(
+                                "Failed to connect to the node (attempt {}, elapsed {:?}): {err}; \
+                                retrying in {delay:?}",
+                                outage.attempts,
+                                outage.started.elapsed(),
+                            );
+                            tokio::time::sleep(delay).await;
+                        }
+                        // A connection cannot be established, but the reason is not a temporary
+                        // connectivity problem (e.g. an invalid address); retrying is pointless.
+                        Ok(Err(err)) => return Err(ApiServerScannerError::RpcError(err)),
+                        Err(_timed_out) => {
+                            outage.attempts += 1;
+                            let delay = backoff.next_delay(rng);
+                            logging::log::warn!(
+                                "Timed out connecting to the node (attempt {}, elapsed {:?}); \
+                                retrying in {delay:?}",
+                                outage.attempts,
+                                outage.started.elapsed(),
+                            );
+                            tokio::time::sleep(delay).await;
+                        }
                     }
                 }
-            });
-
-            if let Some(outage) = outage.take() {
-                let local_height = local_block
-                    .best_block()
-                    .await
-                    .map_or(common::primitives::BlockHeight::zero(), |(height, _)| {
-                        height
-                    });
-                logging::log::info!(
-                    "Scanner reconnected to the node after {} attempt(s) ({:?} elapsed); \
-                    resuming from height {}",
-                    outage.attempts,
-                    outage.started.elapsed(),
-                    local_height,
-                );
             }
-        }
+        };
 
-        let client = rpc_client.as_ref().expect("The RPC client must have been established above");
+        // Note: the loop above leaves the state in `Reconnecting` when it connected (the
+        // connected client is the loop's result); record the new state.
+        state = ConnectionState::Connected(client.clone());
 
-        match api_blockchain_scanner_lib::sync::sync_once(chain_config, client, local_block).await {
+        // Note: the backoff state at this point belongs to the successful connection; it is
+        // only reset after a successful sync, see below.
+        match api_blockchain_scanner_lib::sync::sync_once(chain_config, &client, local_block).await
+        {
             Ok(()) => {
                 // Note: the backoff is only reset here (after a successful sync), not after a
                 // successful connect: a flapping node must not pin the retry rate at the
@@ -290,15 +311,13 @@ async fn supervise_sync<S: ApiServerStorage>(
             Err(err) if err.is_connection_error() => {
                 // The client is permanently broken (e.g. the node has closed the WebSocket
                 // connection); drop it and re-connect on the next iteration.
-                rpc_client = None;
-                let outage = outage.get_or_insert_with(Outage::new);
-                outage.attempts += 1;
+                state = ConnectionState::Reconnecting {
+                    outage: Outage::new(),
+                };
                 let delay = backoff.next_delay(rng);
                 logging::log::warn!(
-                    "Lost the connection to the node (attempt {}, elapsed {:?}): {err}; \
-                    re-connecting in {delay:?}",
-                    outage.attempts,
-                    outage.started.elapsed(),
+                    "Lost the connection to the node (attempt 1, elapsed 0ns): {err}; \
+                    re-connecting in {delay:?}"
                 );
                 tokio::time::sleep(delay).await;
             }

api-server/stack-test-suite/tests/common/proxy.rs

@@ -30,6 +30,9 @@ pub struct ProxyHandle {
     forwarding: tokio::sync::watch::Sender<bool>,
     connections: Arc<tokio::sync::Mutex<Vec<[tokio::task::JoinHandle<()>; 2]>>>,
     task: tokio::task::JoinHandle<()>,
+    /// The number of accepted connections dropped because forwarding was disabled or the
+    /// backend was unreachable; exposed for diagnostics in test failures.
+    refused: Arc<std::sync::atomic::AtomicUsize>,
 }
 
 impl ProxyHandle {
@@ -44,18 +47,21 @@ impl ProxyHandle {
         let addr = listener.local_addr().unwrap();
         let forwarding_rx = forwarding.subscribe();
         let connections = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+        let refused = Arc::new(std::sync::atomic::AtomicUsize::new(0));
         let task = tokio::spawn(run_proxy(
             listener,
             backend_addr,
             forwarding_rx,
             Arc::clone(&connections),
+            Arc::clone(&refused),
         ));
 
         Self {
             addr,
             forwarding,
             connections,
             task,
+            refused,
         }
     }
 
@@ -86,14 +92,23 @@ impl ProxyHandle {
     pub fn assert_alive(&self) {
         assert!(!self.task.is_finished(), "The proxy task has terminated");
     }
+
+    /// The number of connections the proxy has dropped without forwarding them (forwarding
+    /// disabled, or the backend unreachable); useful for diagnosing test failures.
+    pub fn refused_connections(&self) -> usize {
+        self.refused.load(std::sync::atomic::Ordering::Relaxed)
+    }
 }
 
 async fn run_proxy(
     listener: tokio::net::TcpListener,
     backend_addr: SocketAddr,
     mut forwarding: tokio::sync::watch::Receiver<bool>,
     connections: Arc<tokio::sync::Mutex<Vec<[tokio::task::JoinHandle<()>; 2]>>>,
+    refused: Arc<std::sync::atomic::AtomicUsize>,
 ) {
+    use std::sync::atomic::Ordering;
+
     loop {
         let (client_socket, _) = tokio::select! {
             accepted = listener.accept() => accepted.expect("The proxy listener has failed"),
@@ -106,11 +121,13 @@ async fn run_proxy(
 
         if !*forwarding.borrow() {
             // The "node is down": drop the accepted connection right away.
+            refused.fetch_add(1, Ordering::Relaxed);
             continue;
         }
 
         let Ok(backend_socket) = tokio::net::TcpStream::connect(backend_addr).await else {
             // The backend is unreachable; behave like a closed connection.
+            refused.fetch_add(1, Ordering::Relaxed);
             continue;
         };
 
@@ -124,8 +141,9 @@ async fn run_proxy(
         });
 
         let mut connections = connections.lock().await;
-        // Reap the forwarding tasks that finished on their own, so that the list stays bounded.
-        connections.retain(|pair| !pair[0].is_finished() && !pair[1].is_finished());
+        // Reap the forwarding tasks of dead connections: once either direction has finished,
+        // the other is dead by construction (its peer socket has been dropped with the task).
+        connections.retain(|pair| !pair[0].is_finished() || !pair[1].is_finished());
         connections.push([client_to_backend, backend_to_client]);
     }
 }

api-server/stack-test-suite/tests/scanner_reconnect.rs

@@ -48,6 +48,7 @@ use jsonrpsee::{
     core::RpcResult,
     types::{ErrorObjectOwned, error::INTERNAL_ERROR_CODE},
 };
+use randomness::RngExt as _;
 use rpc::RpcAuthData;
 use serialization::hex_encoded::HexEncoded;
 use test_common::proxy::ProxyHandle;
@@ -256,14 +257,25 @@ async fn scanner_reconnects_after_node_disconnection() {
     });
 
     // Index a few blocks while everything is healthy.
-    // Note: the framework mutex is shared between this (chain building, on the runtime) and
-    // the spawn_blocking-backed RPC serving; the two serialize for the duration of the chain
-    // building, which is fine at this scale (the scanner's calls resume right after).
-    let tip_3 = {
-        let mut framework = framework.lock().await;
-        let genesis_id = framework.chain_config().genesis_block_id();
-        framework.create_chain_with_empty_blocks(&genesis_id, 3, &mut rng).unwrap()
-    };
+    // Note: the framework mutex is shared between this test (chain building) and the
+    // spawn_blocking-backed RPC serving; chain building is itself moved off the runtime thread
+    // (it is synchronous, CPU-heavy block processing), and the two serialize for its duration —
+    // fine at this scale (the scanner's in-flight calls resume right after).
+    // Note: deterministic child seeds are derived from the master RNG, so that a CI failure is
+    // locally reproducible with a fixed seed.
+    let phase_1_seed = Seed(rng.random_range(0..u64::MAX));
+    let phase_2_seed = Seed(rng.random_range(0..u64::MAX));
+    let tip_3 = tokio::task::spawn_blocking({
+        let framework = Arc::clone(&framework);
+        let mut rng = make_seedable_rng(phase_1_seed);
+        move || {
+            let mut framework = framework.blocking_lock();
+            let genesis_id = framework.chain_config().genesis_block_id();
+            framework.create_chain_with_empty_blocks(&genesis_id, 3, &mut rng).unwrap()
+        }
+    })
+    .await
+    .unwrap();
     proxy.assert_alive();
     wait_for_scanner_tip(&test_storage, tip_3).await;
     assert_eq!(
@@ -279,10 +291,16 @@ async fn scanner_reconnects_after_node_disconnection() {
     proxy.set_forwarding(false);
     tokio::time::sleep(OUTAGE_DURATION).await;
 
-    let tip_5 = {
-        let mut framework = framework.lock().await;
-        framework.create_chain_with_empty_blocks(&tip_3, 2, &mut rng).unwrap()
-    };
+    let tip_5 = tokio::task::spawn_blocking({
+        let framework = Arc::clone(&framework);
+        let mut rng = make_seedable_rng(phase_2_seed);
+        move || {
+            let mut framework = framework.blocking_lock();
+            framework.create_chain_with_empty_blocks(&tip_3, 2, &mut rng).unwrap()
+        }
+    })
+    .await
+    .unwrap();
     assert_ne!(tip_3, tip_5);
 
     // The scanner must not have given up (or crashed) while the node was unreachable.
@@ -303,11 +321,10 @@ async fn scanner_reconnects_after_node_disconnection() {
     // The supervision loop is still running (the recovery happened in place).
     assert!(!scanner_task.is_finished());
 
-    // Note: the scanner task is intentionally not allowed to run to completion; the destructors
-    // of the storage handles and of the proxy close the remaining resources. Give the scanner a
-    // moment to finish the in-flight sync before the storage handles are dropped, then stop it
-    // explicitly, so that it cannot panic over the closed storage.
-    tokio::time::sleep(Duration::from_millis(500)).await;
+    // Note: the scanner task is stopped explicitly before the test's handles are dropped:
+    // abort cancels it at its next await point and the join below waits for the unwind to
+    // finish, so the daemon can never race the closing Postgres pool (no timing heuristics
+    // needed).
     scanner_task.abort();
     let _ = scanner_task.await;
 }