fix(blacklight-node): recover from silently dead event subscriptions

.github/workflows/docker.yml

@@ -6,11 +6,18 @@ on:
   workflow_dispatch:
 env:
   REGISTRY: ghcr.io
-  IMAGE_NAME: ${{ github.repository }}
 
+# Each architecture is built on a native runner (cross-building Rust under QEMU
+# is prohibitively slow), then the per-arch images are combined into a single
+# multi-arch manifest.
+#
+# The per-arch builds deliberately push *by digest only*. Previously every arch
+# job applied the same tags, so whichever finished last silently overwrote the
+# other and the published tag resolved to one architecture — which is how every
+# release from 0.10.1 onward ended up amd64-only.
 jobs:
-  build-and-push:
-    name: Build and Push Docker Images
+  build:
+    name: Build ${{ matrix.target }} (${{ matrix.arch }})
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read
@@ -56,30 +63,115 @@ jobs:
             echo "version=dev-$(git rev-parse --short HEAD)" >> $GITHUB_OUTPUT
           fi
 
-      - name: Extract metadata (tags, labels)
+      - name: Resolve image name
+        run: echo "IMAGE=${REGISTRY}/${REPO,,}/${TARGET}" >> "$GITHUB_ENV"
+        env:
+          REGISTRY: ${{ env.REGISTRY }}
+          REPO: ${{ github.repository }}
+          TARGET: ${{ matrix.target }}
+
+      - name: Extract metadata (labels)
         id: meta
         uses: docker/metadata-action@v5
         with:
-          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}/${{ matrix.target }}
-          tags: |
-            type=semver,pattern={{version}}
-            type=semver,pattern={{major}}.{{minor}}
-            type=semver,pattern={{major}}
-            type=sha,prefix=sha-
-            type=raw,value=latest,enable={{is_default_branch}}
+          images: ${{ env.IMAGE }}
 
-      - name: Build and push docker image
+      - name: Build and push by digest
         id: build
         uses: docker/build-push-action@v6
         with:
           context: .
           file: ./docker/${{ matrix.target }}.dockerfile
-          push: true
-          tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
-          cache-from: type=gha,scope=${{ matrix.target }}
-          cache-to: type=gha,mode=max,scope=${{ matrix.target }}
+          cache-from: type=gha,scope=${{ matrix.target }}-${{ matrix.arch }}
+          cache-to: type=gha,mode=max,scope=${{ matrix.target }}-${{ matrix.arch }}
           platforms: ${{ matrix.platform }}
           build-args: |
             BLACKLIGHT_VERSION=${{ steps.version.outputs.version }}
           provenance: false
+          outputs: type=image,name=${{ env.IMAGE }},push-by-digest=true,name-canonical=true,push=true
+
+      - name: Export digest
+        run: |
+          mkdir -p /tmp/digests
+          digest="${{ steps.build.outputs.digest }}"
+          touch "/tmp/digests/${digest#sha256:}"
+
+      - name: Upload digest
+        uses: actions/upload-artifact@v4
+        with:
+          name: digests-${{ matrix.target }}-${{ matrix.arch }}
+          path: /tmp/digests/*
+          if-no-files-found: error
+          retention-days: 1
+
+  merge:
+    name: Publish ${{ matrix.target }} manifest
+    runs-on: ubuntu-latest
+    needs: build
+    permissions:
+      contents: read
+      packages: write
+    strategy:
+      fail-fast: false
+      matrix:
+        target:
+          - blacklight_node
+          - keeper
+    steps:
+      - name: Download digests
+        uses: actions/download-artifact@v4
+        with:
+          path: /tmp/digests
+          pattern: digests-${{ matrix.target }}-*
+          merge-multiple: true
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Resolve image name
+        run: echo "IMAGE=${REGISTRY}/${REPO,,}/${TARGET}" >> "$GITHUB_ENV"
+        env:
+          REGISTRY: ${{ env.REGISTRY }}
+          REPO: ${{ github.repository }}
+          TARGET: ${{ matrix.target }}
+
+      # Tags are applied here, once, to the manifest list — so no two jobs can
+      # race for the same tag.
+      - name: Extract metadata (tags)
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.IMAGE }}
+          tags: |
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=semver,pattern={{major}}
+            type=sha,prefix=sha-
+            type=raw,value=latest,enable={{is_default_branch}}
+
+      - name: Create and push manifest list
+        working-directory: /tmp/digests
+        run: |
+          docker buildx imagetools create \
+            $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< "$DOCKER_METADATA_OUTPUT_JSON") \
+            $(printf "${IMAGE}@sha256:%s " *)
+
+      - name: Verify manifest is multi-arch
+        run: |
+          tag="$(jq -cr '.tags[0]' <<< "$DOCKER_METADATA_OUTPUT_JSON")"
+          docker buildx imagetools inspect "$tag"
+          arches="$(docker buildx imagetools inspect --raw "$tag" \
+            | jq -r '[.manifests[].platform.architecture | select(. != "unknown")] | sort | unique | join(",")')"
+          echo "architectures: $arches"
+          if [ "$arches" != "amd64,arm64" ]; then
+            echo "::error::expected amd64,arm64 in $tag but found '$arches'"
+            exit 1
+          fi

blacklight-node/src/args.rs

@@ -44,8 +44,31 @@ pub struct CliArgs {
     /// The path where AMD certificates will be cached.
     #[clap(short, long, default_value = default_cert_cache_path().into_os_string(), env = "CERT_CACHE")]
     pub cert_cache: PathBuf,
+
+    /// How long the RoundStarted subscription may stay silent before the node
+    /// assumes it has gone deaf and rebuilds its connection.
+    ///
+    /// The node receives every round on the network (committee membership is
+    /// filtered locally), and rounds are started on a fixed ~5 minute cadence,
+    /// so prolonged silence means the subscription is dead rather than the
+    /// network being quiet. Set to 0 to disable the watchdog entirely.
+    #[arg(long, default_value_t = DEFAULT_EVENT_IDLE_TIMEOUT_SECS, env = "EVENT_IDLE_TIMEOUT_SECS")]
+    pub event_idle_timeout_secs: u64,
 }
 
+/// Default idle timeout before the event subscription is presumed dead.
+///
+/// Roughly three missed round cadences: long enough to never trip during normal
+/// operation, short enough that at most one assignment is missed.
+pub const DEFAULT_EVENT_IDLE_TIMEOUT_SECS: u64 = 900;
+
+/// Spread of the per-node jitter applied to the idle timeout.
+///
+/// Silence is a network-wide signal, so without jitter every node would trip its
+/// watchdog in the same instant and reconnect as one herd against a single RPC —
+/// which is the failure this whole mechanism exists to recover from.
+pub const EVENT_IDLE_TIMEOUT_JITTER_SECS: u64 = 180;
+
 /// Node configuration with all required values resolved
 #[derive(Debug, Clone)]
 pub struct NodeConfig {
@@ -55,13 +78,15 @@ pub struct NodeConfig {
     pub token_contract_address: Address,
     pub private_key: String,
     pub was_wallet_created: bool,
+    pub event_idle_timeout_secs: u64,
 }
 
 impl NodeConfig {
     /// Load configuration with priority: CLI/env -> state file -> defaults
     /// Generates a new wallet if none exists
     /// Returns (NodeConfig, was_wallet_created)
     pub async fn load(cli_args: CliArgs) -> Result<Self> {
+        let event_idle_timeout_secs = cli_args.event_idle_timeout_secs;
         let state_file = StateFile::new(STATE_FILE_NODE);
         let ChainConfig {
             rpc_url,
@@ -127,6 +152,7 @@ impl NodeConfig {
             token_contract_address,
             private_key,
             was_wallet_created,
+            event_idle_timeout_secs,
         })
     }
 }

blacklight-node/src/main.rs

@@ -46,10 +46,10 @@ async fn main() -> Result<()> {
 
     // Create and run supervisor (handles connection, validation, and event processing)
     let supervisor = supervisor::Supervisor::new(&config, &verifier, shutdown_token).await?;
-    let client = supervisor.run().await?;
+    supervisor.run().await?;
 
     // Graceful shutdown - deactivate node
-    if let Err(e) = shutdown::deactivate_node(&client).await {
+    if let Err(e) = shutdown::deactivate_node(&config).await {
         error!(error = %e, "Failed to deactivate node gracefully");
     }
 

blacklight-node/src/shutdown.rs

@@ -1,8 +1,10 @@
-use anyhow::Result;
-use blacklight_contract_clients::BlacklightClient;
+use anyhow::{Context, Result};
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 
+use crate::args::NodeConfig;
+use crate::supervisor::Supervisor;
+
 /// Setup shutdown signal handler (Ctrl+C / SIGTERM)
 pub async fn shutdown_signal(shutdown_token: CancellationToken) {
     #[cfg(unix)]
@@ -43,9 +45,19 @@ pub async fn shutdown_signal(shutdown_token: CancellationToken) {
 }
 
 /// Deactivate node from contract on shutdown
-pub async fn deactivate_node(client: &BlacklightClient) -> Result<()> {
-    let node_address = client.signer_address();
+///
+/// Builds a fresh client rather than reusing the supervisor's. Shutdown is often
+/// reached precisely because the WebSocket connection died, and on that path the
+/// supervisor still holds the dead client — deactivation would then fail and
+/// leave the operator active on-chain, so it keeps being selected into
+/// committees it can no longer vote in.
+pub async fn deactivate_node(config: &NodeConfig) -> Result<()> {
     info!("Initiating graceful shutdown");
+
+    let client = Supervisor::create_client(config)
+        .await
+        .context("Failed to create client for deactivation")?;
+    let node_address = client.signer_address();
     info!(node_address = %node_address, "Deactivating node from contract");
 
     let tx_hash = client.staking.deactivate_operator().await?;

blacklight-node/src/supervisor/events.rs

@@ -1,26 +1,35 @@
 use anyhow::Result;
-use blacklight_contract_clients::BlacklightClient;
+use blacklight_contract_clients::{BlacklightClient, StreamWatchdog};
 use std::sync::Arc;
 use tracing::info;
 
 use super::htx::{HtxEventSource, HtxProcessor};
 /// Listen for HTX assignment events and process them
-pub async fn run_event_listener(client: BlacklightClient, processor: HtxProcessor) -> Result<()> {
+///
+/// `watchdog` is optional so the listener can also be driven without liveness
+/// checking; when present, this returns an error once the subscription has been
+/// silent for longer than the configured idle timeout.
+pub async fn run_event_listener(
+    client: BlacklightClient,
+    processor: HtxProcessor,
+    watchdog: Option<StreamWatchdog>,
+) -> Result<()> {
     let client_for_callback = client.clone();
     let processor_for_callback = processor.clone();
 
     let manager = Arc::new(client.manager.clone());
     let node_address = processor.node_address();
-    let listen_future = manager.listen_htx_assigned_for_node(node_address, move |event| {
-        let client = client_for_callback.clone();
-        let processor = processor_for_callback.clone();
-        async move {
-            let vote_address = client.signer_address();
-            processor.spawn_processing(event, vote_address, HtxEventSource::Realtime, true);
+    let listen_future =
+        manager.listen_htx_assigned_for_node(node_address, watchdog, move |event| {
+            let client = client_for_callback.clone();
+            let processor = processor_for_callback.clone();
+            async move {
+                let vote_address = client.signer_address();
+                processor.spawn_processing(event, vote_address, HtxEventSource::Realtime, true);
 
-            Ok(())
-        }
-    });
+                Ok(())
+            }
+        });
 
     // Listen for either events or shutdown signal
     let shutdown_token = processor.shutdown_token();

blacklight-node/src/supervisor/htx.rs

@@ -78,6 +78,16 @@ impl HtxProcessor {
                             }
                         }
                         Ok(None) => {}
+                        Err(e) if source.is_expected_duplicate(&e) => {
+                            // Backlog rescans re-offer rounds this node already
+                            // voted in, because the "already responded" check
+                            // above cannot see them (get_node_vote queries round
+                            // 0, while rounds are numbered from 1). Submission
+                            // pre-simulates, so this costs no gas and sends no
+                            // transaction — logging it as an error would make
+                            // every routine reconnect look like a failure.
+                            debug!(htx_id = ?htx_id, error = %e, "Already voted in this round, skipping");
+                        }
                         Err(e) => {
                             error!(htx_id = ?htx_id, error = %e, "{}", source.process_error_message());
                         }
@@ -160,10 +170,9 @@ impl HtxProcessor {
 
                 Ok(Some(count))
             }
-            Err(e) => {
-                error!(htx_id = ?htx_id, error = %e, "Failed to respond to HTX");
-                Err(e)
-            }
+            // Logged by the caller, which knows whether this was a real-time or
+            // a backlog event and so whether the failure is worth an error.
+            Err(e) => Err(e),
         }
     }
 
@@ -248,4 +257,25 @@ impl HtxEventSource {
             Self::Backlog => "Failed to check assignment status",
         }
     }
+
+    /// Whether this failure is the harmless "we already voted" revert that a
+    /// backlog rescan is expected to produce.
+    ///
+    /// Only treated as routine for backlog events; hitting it on a real-time
+    /// event would mean something genuinely unexpected happened.
+    fn is_expected_duplicate(self, error: &anyhow::Error) -> bool {
+        if !matches!(self, Self::Backlog) {
+            return false;
+        }
+        let message = error.to_string();
+        DUPLICATE_VOTE_REVERTS
+            .iter()
+            .any(|revert| message.contains(revert))
+    }
 }
+
+/// Reverts that mean "this round no longer accepts our vote".
+///
+/// Matched on the decoded revert name, which is what the transaction submitter
+/// puts in the error message.
+const DUPLICATE_VOTE_REVERTS: [&str; 3] = ["AlreadyResponded", "NotPending", "RoundClosed"];