diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index ec7044d..2a6b843 100644 --- a/.github/workflows/docker.yml +++ b/.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 diff --git a/blacklight-node/src/args.rs b/blacklight-node/src/args.rs index 86dc707..007050b 100644 --- a/blacklight-node/src/args.rs +++ b/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,6 +78,7 @@ 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 { @@ -62,6 +86,7 @@ impl NodeConfig { /// Generates a new wallet if none exists /// Returns (NodeConfig, was_wallet_created) pub async fn load(cli_args: CliArgs) -> Result { + 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, }) } } diff --git a/blacklight-node/src/main.rs b/blacklight-node/src/main.rs index 0565078..2378c58 100644 --- a/blacklight-node/src/main.rs +++ b/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"); } diff --git a/blacklight-node/src/shutdown.rs b/blacklight-node/src/shutdown.rs index 6a211f7..39eea7f 100644 --- a/blacklight-node/src/shutdown.rs +++ b/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?; diff --git a/blacklight-node/src/supervisor/events.rs b/blacklight-node/src/supervisor/events.rs index 50b5658..b9698bc 100644 --- a/blacklight-node/src/supervisor/events.rs +++ b/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, +) -> 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(); diff --git a/blacklight-node/src/supervisor/htx.rs b/blacklight-node/src/supervisor/htx.rs index 402d20a..9c82a6f 100644 --- a/blacklight-node/src/supervisor/htx.rs +++ b/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"]; diff --git a/blacklight-node/src/supervisor/mod.rs b/blacklight-node/src/supervisor/mod.rs index a8167e6..c4eea2f 100644 --- a/blacklight-node/src/supervisor/mod.rs +++ b/blacklight-node/src/supervisor/mod.rs @@ -1,13 +1,13 @@ use alloy::primitives::Address; use anyhow::{Result, bail}; -use blacklight_contract_clients::{BlacklightClient, ContractConfig}; +use blacklight_contract_clients::{BlacklightClient, ContractConfig, StreamWatchdog}; use std::sync::Arc; -use std::sync::atomic::AtomicU64; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; -use crate::args::{NodeConfig, validate_node_requirements}; +use crate::args::{EVENT_IDLE_TIMEOUT_JITTER_SECS, NodeConfig, validate_node_requirements}; use crate::verification::HtxVerifier; use crate::supervisor::htx::HtxProcessor; @@ -23,6 +23,15 @@ const INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(1); /// Maximum reconnection delay const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(60); +/// How many times alloy's transport retries the socket before giving up. +/// +/// Bounded rather than infinite because this supervisor does its own +/// reconnection. While alloy retries — on a flat 3s interval, with no backoff or +/// jitter — the failure is invisible here: the subscription stream stays open, so +/// nothing below can react. Letting the transport give up (~30s) closes the +/// stream, which hands control to the backoff and full client rebuild in `run`. +const MAX_WS_RETRIES: u32 = 10; + /// Node supervisor - manages WebSocket connection, reconnection, and event processing pub struct Supervisor<'a> { config: &'a NodeConfig, @@ -32,6 +41,9 @@ pub struct Supervisor<'a> { node_address: Address, reconnect_delay: Duration, client: BlacklightClient, + /// Events delivered by the current subscription, used to tell a working + /// stream from one that never delivered anything. + events_seen: Arc, } impl<'a> Supervisor<'a> { @@ -60,11 +72,32 @@ impl<'a> Supervisor<'a> { node_address, reconnect_delay: INITIAL_RECONNECT_DELAY, client, + events_seen: Arc::new(AtomicU64::new(0)), }) } - /// Run the supervisor loop, returns the client for use in shutdown - pub async fn run(mut self) -> Result { + /// Build the liveness watchdog for the event subscription, if enabled. + /// + /// The timeout is jittered per node so that a network-wide lull does not + /// make the whole fleet reconnect simultaneously. + fn watchdog(&self) -> Option { + let base_secs = self.config.event_idle_timeout_secs; + if base_secs == 0 { + warn!("Event stream watchdog disabled (EVENT_IDLE_TIMEOUT_SECS=0)"); + return None; + } + let timeout = Duration::from_secs( + base_secs + jitter_secs(self.node_address, EVENT_IDLE_TIMEOUT_JITTER_SECS), + ); + info!( + idle_timeout_secs = timeout.as_secs(), + "Event stream watchdog armed" + ); + Some(StreamWatchdog::new(timeout, self.events_seen.clone())) + } + + /// Run the supervisor loop until shutdown is requested + pub async fn run(mut self) -> Result<()> { loop { info!("Starting WebSocket event listener with auto-reconnection"); info!("Press Ctrl+C to gracefully shutdown and deactivate"); @@ -84,36 +117,48 @@ impl<'a> Supervisor<'a> { } // Start listening for events - match self.listen_for_events(client).await { + let events_before = self.events_seen.load(Ordering::Relaxed); + let outcome = self.listen_for_events(client).await; + let stream_was_healthy = self.events_seen.load(Ordering::Relaxed) > events_before; + + match outcome { Ok(_) => { warn!("WebSocket listener exited normally. Reconnecting..."); - if self.reconnect_client().await? { - break; - } } Err(e) if e.to_string().contains("Shutdown") => { break; } Err(e) => { error!(error = %e, "WebSocket listener error. Reconnecting..."); - if self.reconnect_client().await? { - break; - } } } + + // Only a stream that actually delivered events counts as healthy. + // Resetting on connection success instead would keep the backoff + // pinned at its minimum whenever the RPC accepts the socket but + // fails the subscription, turning recovery into a hot loop across + // the whole fleet. + if stream_was_healthy { + self.reconnect_delay = INITIAL_RECONNECT_DELAY; + } + + if self.reconnect_client().await? { + break; + } } - Ok(self.client) + Ok(()) } /// Create a new WebSocket client - async fn create_client(config: &NodeConfig) -> Result { + pub(crate) async fn create_client(config: &NodeConfig) -> Result { let contract_config = ContractConfig::new( config.rpc_url.clone(), config.manager_contract_address, config.staking_contract_address, config.token_contract_address, - ); + ) + .with_max_ws_retries(MAX_WS_RETRIES); BlacklightClient::new(contract_config, config.private_key.clone()).await } @@ -172,7 +217,12 @@ impl<'a> Supervisor<'a> { /// Listen for HTX assignment events async fn listen_for_events(&self, client: BlacklightClient) -> Result<()> { - events::run_event_listener(client.clone(), self.build_htx_processor(client)).await + events::run_event_listener( + client.clone(), + self.build_htx_processor(client), + self.watchdog(), + ) + .await } fn build_htx_processor(&self, client: BlacklightClient) -> HtxProcessor { @@ -186,19 +236,22 @@ impl<'a> Supervisor<'a> { } /// Reconnect the client with retry/backoff. Returns true if shutdown was requested. + /// + /// Always waits before reconnecting. Creating the client can succeed while + /// the subscription that follows still fails, so returning immediately on a + /// successful connect would let the caller spin with no delay at all. async fn reconnect_client(&mut self) -> Result { loop { + if self.wait_before_reconnect().await { + return Ok(true); + } match Self::create_client(self.config).await { Ok(client) => { self.client = client; - self.reconnect_delay = INITIAL_RECONNECT_DELAY; return Ok(false); } Err(e) => { error!(error = %e, "Failed to create client. Retrying..."); - if self.wait_before_reconnect().await { - return Ok(true); - } } } } @@ -220,3 +273,57 @@ impl<'a> Supervisor<'a> { } } } + +/// Deterministic per-node offset in `0..spread` seconds, derived from the node +/// address so it is stable across restarts but differs between operators. +fn jitter_secs(node_address: Address, spread: u64) -> u64 { + if spread == 0 { + return 0; + } + let bytes = node_address.into_array(); + let seed = u64::from_be_bytes(bytes[..8].try_into().expect("address is 20 bytes")); + seed % spread +} + +#[cfg(test)] +mod tests { + use super::*; + + fn addr(hex: &str) -> Address { + hex.parse().expect("valid address") + } + + #[test] + fn jitter_is_within_spread() { + for a in [ + "0x8413388033aC4F79c34285ad6f7b3684231A5c45", + "0xe0d1e31C3c7cC3a2554ead6BCB8035eD6f69f6Ab", + "0xc831594C6748D900B4FD9068705ee828d3e2DBAe", + ] { + assert!(jitter_secs(addr(a), 180) < 180); + } + } + + #[test] + fn jitter_is_stable_for_the_same_address() { + let a = addr("0x8413388033aC4F79c34285ad6f7b3684231A5c45"); + assert_eq!(jitter_secs(a, 180), jitter_secs(a, 180)); + } + + #[test] + fn jitter_differs_between_nodes() { + // Not a guarantee for arbitrary inputs, but it must hold for real + // operator addresses or the fleet would still reconnect in lockstep. + let a = jitter_secs(addr("0x8413388033aC4F79c34285ad6f7b3684231A5c45"), 180); + let b = jitter_secs(addr("0xe0d1e31C3c7cC3a2554ead6BCB8035eD6f69f6Ab"), 180); + assert_ne!(a, b); + } + + #[test] + fn zero_spread_disables_jitter() { + assert_eq!( + jitter_secs(addr("0x8413388033aC4F79c34285ad6f7b3684231A5c45"), 0), + 0 + ); + } +} diff --git a/crates/blacklight-contract-clients/src/heartbeat_manager.rs b/crates/blacklight-contract-clients/src/heartbeat_manager.rs index 0931124..6308b14 100644 --- a/crates/blacklight-contract-clients/src/heartbeat_manager.rs +++ b/crates/blacklight-contract-clients/src/heartbeat_manager.rs @@ -7,7 +7,9 @@ use alloy::{ sol_types::SolValue, }; use anyhow::{Context, Result, anyhow, bail}; -use contract_clients_common::event_helper::{BlockRange, listen_events, listen_events_filtered}; +use contract_clients_common::event_helper::{ + BlockRange, StreamWatchdog, listen_events, listen_events_filtered, +}; use contract_clients_common::tx_submitter::TransactionSubmitter; use std::sync::Arc; use tokio::sync::Mutex; @@ -233,9 +235,15 @@ impl HeartbeatManagerClient

{ } /// Start listening for HTX assigned events for a specific node + /// + /// Committee membership is filtered client-side because `members` is not an + /// indexed event field. That means the subscription carries every round on + /// the network, so `watchdog` can watch the whole stream rather than this + /// node's much sparser (and stake-weighted) share of it. pub async fn listen_htx_assigned_for_node( self: Arc, node_address: Address, + watchdog: Option, callback: F, ) -> Result<()> where @@ -250,6 +258,7 @@ impl HeartbeatManagerClient

{ listen_events_filtered( subscription.into_stream(), "RoundStarted", + watchdog, move |event: &RoundStartedEvent| event.members.contains(&node_address), callback, ) diff --git a/crates/blacklight-contract-clients/src/lib.rs b/crates/blacklight-contract-clients/src/lib.rs index 45d5410..4cebc45 100644 --- a/crates/blacklight-contract-clients/src/lib.rs +++ b/crates/blacklight-contract-clients/src/lib.rs @@ -53,6 +53,10 @@ pub use node_operator_factory::NodeOperatorFactory; /// Type alias for private key strings pub type PrivateKey = String; +/// Re-exported so binaries can configure event-stream liveness without taking a +/// direct dependency on `contract-clients-common`. +pub use contract_clients_common::event_helper::StreamWatchdog; + // ============================================================================ // Contract Configuration // ============================================================================ @@ -72,6 +76,10 @@ pub struct ContractConfig { pub token_contract_address: Address, pub rpc_url: String, /// Maximum number of WebSocket reconnection attempts (default: u32::MAX for infinite) + /// + /// Consumers that supervise their own reconnection should bound this via + /// [`ContractConfig::with_max_ws_retries`], so a dead transport surfaces to + /// them instead of being retried forever inside alloy. pub max_ws_retries: u32, } diff --git a/crates/contract-clients-common/Cargo.toml b/crates/contract-clients-common/Cargo.toml index 42e75f7..9b96a7e 100644 --- a/crates/contract-clients-common/Cargo.toml +++ b/crates/contract-clients-common/Cargo.toml @@ -8,5 +8,10 @@ anyhow = "1.0" alloy = { version = "1.1", features = ["contract", "providers", "pubsub"] } alloy-provider = { version = "1.1", features = ["ws"] } futures-util = "0.3" -tokio = { version = "1.49", features = ["sync"] } +# "time" is needed by the event-stream watchdog. It also arrives via alloy's +# feature unification today, but this crate uses it directly. +tokio = { version = "1.49", features = ["sync", "time"] } tracing = "0.1" + +[dev-dependencies] +tokio = { version = "1.49", features = ["macros", "rt", "test-util", "time"] } diff --git a/crates/contract-clients-common/src/event_helper.rs b/crates/contract-clients-common/src/event_helper.rs index e4fd7e5..ebe8c9f 100644 --- a/crates/contract-clients-common/src/event_helper.rs +++ b/crates/contract-clients-common/src/event_helper.rs @@ -12,10 +12,44 @@ //! let range = BlockRange::last_n_blocks(1000); //! ``` -use anyhow::Result; +use anyhow::{Result, bail}; use futures_util::StreamExt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; use tracing::error; +/// Detects an event subscription that has gone silently dead. +/// +/// An RPC can accept a WebSocket connection and still fail to register the +/// subscription server-side (for example by answering `eth_subscribe` with an +/// error while it is overloaded). Alloy re-issues subscriptions on reconnect but +/// only re-registers them on a *successful* response, and it discards failures, +/// so the local stream stays open and simply never yields again. No transport +/// error is reported and the stream never terminates, which means silence on the +/// wire is the only available signal. +#[derive(Clone, Debug)] +pub struct StreamWatchdog { + /// How long the stream may stay silent before it is presumed dead. + idle_timeout: Duration, + /// Number of events delivered, so callers can distinguish a stream that is + /// working from one that never delivered anything. + events_seen: Arc, +} + +impl StreamWatchdog { + pub fn new(idle_timeout: Duration, events_seen: Arc) -> Self { + Self { + idle_timeout, + events_seen, + } + } + + pub fn idle_timeout(&self) -> Duration { + self.idle_timeout + } +} + /// Represents a block range for event queries. /// /// Provides convenient constructors for common query patterns. @@ -89,11 +123,14 @@ impl Default for BlockRange { /// /// * `stream` - The event stream to listen to /// * `event_name` - Name of the event for logging purposes +/// * `watchdog` - Optional liveness watchdog; returns an error if the stream +/// stays silent longer than its idle timeout /// * `predicate` - Function that returns true if the event should be processed /// * `callback` - Async function to process each matching event pub async fn listen_events_filtered( mut stream: L, event_name: &str, + watchdog: Option, predicate: P, mut callback: F, ) -> Result<()> @@ -105,9 +142,37 @@ where F: FnMut(E) -> Fut + Send, Fut: std::future::Future> + Send, { - while let Some(event_result) = stream.next().await { + loop { + // Any item resets the idle timer, including a decode failure: it still + // proves the subscription is delivering. + let next_item = match &watchdog { + Some(watchdog) => { + match tokio::time::timeout(watchdog.idle_timeout, stream.next()).await { + Ok(item) => item, + // Deliberately worded to avoid the substring "Shutdown", which + // the node's supervisor uses to recognise an intentional stop. + Err(_elapsed) => bail!( + "no {} events received in {}s; subscription presumed dead", + event_name, + watchdog.idle_timeout.as_secs() + ), + } + } + None => stream.next().await, + }; + + let Some(event_result) = next_item else { + // Stream closed: the transport gave up, which the caller can see. + break; + }; + match event_result { Ok((event, _log)) => { + // Counted before the predicate so liveness reflects the whole + // subscription rather than this node's share of it. + if let Some(watchdog) = &watchdog { + watchdog.events_seen.fetch_add(1, Ordering::Relaxed); + } if predicate(&event) && let Err(e) = callback(event).await { @@ -144,7 +209,7 @@ where F: FnMut(E) -> Fut + Send, Fut: std::future::Future> + Send, { - listen_events_filtered(stream, event_name, |_| true, callback).await + listen_events_filtered(stream, event_name, None, |_| true, callback).await } #[cfg(test)] @@ -185,4 +250,103 @@ mod tests { assert_eq!(range.from_block, 0); assert_eq!(range.to_block, None); } + + // ------------------------------------------------------------------------ + // Watchdog + // ------------------------------------------------------------------------ + + type TestItem = Result<(u8, alloy::rpc::types::Log), String>; + + fn watchdog(secs: u64) -> (StreamWatchdog, Arc) { + let seen = Arc::new(AtomicU64::new(0)); + ( + StreamWatchdog::new(Duration::from_secs(secs), seen.clone()), + seen, + ) + } + + fn event(v: u8) -> TestItem { + Ok((v, alloy::rpc::types::Log::default())) + } + + #[tokio::test(start_paused = true)] + async fn watchdog_errors_when_stream_goes_silent() { + let (wd, _seen) = watchdog(900); + let stream = futures_util::stream::pending::(); + + let err = listen_events_filtered( + stream, + "RoundStarted", + Some(wd), + |_| true, + |_| async { Ok(()) }, + ) + .await + .expect_err("a silent stream must be reported"); + + let message = err.to_string(); + assert!(message.contains("RoundStarted"), "message: {message}"); + assert!(message.contains("900"), "message: {message}"); + // The supervisor distinguishes an intentional stop by looking for this + // substring, so the watchdog must never produce it. + assert!(!message.contains("Shutdown"), "message: {message}"); + } + + #[tokio::test(start_paused = true)] + async fn silent_stream_is_tolerated_without_a_watchdog() { + // Ends only because the stream terminates; with `pending` this would + // hang, which is exactly the pre-watchdog behaviour. + let stream = futures_util::stream::iter(Vec::::new()); + listen_events_filtered(stream, "RoundStarted", None, |_| true, |_| async { Ok(()) }) + .await + .expect("stream closing is not an error"); + } + + #[tokio::test(start_paused = true)] + async fn events_are_counted_before_the_predicate() { + let (wd, seen) = watchdog(900); + let stream = futures_util::stream::iter(vec![event(1), event(2), event(3)]); + let delivered = Arc::new(AtomicU64::new(0)); + let delivered_cb = delivered.clone(); + + // Reject everything, as a node not in the committee would. + listen_events_filtered( + stream, + "RoundStarted", + Some(wd), + |_| false, + move |_| { + let delivered = delivered_cb.clone(); + async move { + delivered.fetch_add(1, Ordering::Relaxed); + Ok(()) + } + }, + ) + .await + .expect("stream completed"); + + // Liveness must track the whole subscription, not this node's share: + // rounds it is not a member of still prove the stream is alive. + assert_eq!(seen.load(Ordering::Relaxed), 3); + assert_eq!(delivered.load(Ordering::Relaxed), 0); + } + + #[tokio::test(start_paused = true)] + async fn stream_errors_do_not_count_as_events() { + let (wd, seen) = watchdog(900); + let stream = futures_util::stream::iter(vec![Err("decode failed".to_string()), event(1)]); + + listen_events_filtered( + stream, + "RoundStarted", + Some(wd), + |_| true, + |_| async { Ok(()) }, + ) + .await + .expect("stream completed"); + + assert_eq!(seen.load(Ordering::Relaxed), 1); + } } diff --git a/docker/blacklight_node.dockerfile b/docker/blacklight_node.dockerfile index cb5a69b..4024686 100644 --- a/docker/blacklight_node.dockerfile +++ b/docker/blacklight_node.dockerfile @@ -11,6 +11,7 @@ COPY Cargo.toml ./ COPY crates ./crates COPY simulator ./simulator COPY keeper ./keeper +COPY managed-node-keeper ./managed-node-keeper COPY blacklight-node ./blacklight-node COPY monitor ./monitor diff --git a/docker/keeper.dockerfile b/docker/keeper.dockerfile index d3dc7a7..9036a49 100644 --- a/docker/keeper.dockerfile +++ b/docker/keeper.dockerfile @@ -9,6 +9,7 @@ COPY Cargo.toml ./ COPY crates ./crates COPY simulator ./simulator COPY keeper ./keeper +COPY managed-node-keeper ./managed-node-keeper COPY blacklight-node ./blacklight-node COPY monitor ./monitor