harden: supervise the task tree and add worker liveness

bin/dipper-service/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2024"
 anyhow.workspace = true
 async-signal = "0.2.10"
 async-trait.workspace = true
+axum = { version = "0.8", default-features = false, features = ["http1", "tokio"] }
 custom_debug = "0.6.1"
 dashmap = "6.1.0"
 dipper-core = { version = "0.1.0", path = "../../dipper-core" }

bin/dipper-service/src/chain_client.rs

@@ -145,3 +145,85 @@ impl<T: ChainClient + Send + Sync + ?Sized> ChainClient for Arc<T> {
         (**self).post_offer(rca).await
     }
 }
+
+/// Runs the periodic RCA EIP-712 domain refresh until `stop_rx` fires.
+///
+/// Generic over the refresh action so the stop wiring is unit testable without
+/// a live chain. The first (immediate) interval tick is skipped; thereafter
+/// each tick invokes `refresh`, whose errors are logged and swallowed (the
+/// current domain is kept). Returns `Ok(())` on stop.
+pub async fn run_domain_refresh<F, Fut>(
+    interval: std::time::Duration,
+    mut stop_rx: tokio::sync::mpsc::Receiver<()>,
+    mut refresh: F,
+) -> anyhow::Result<()>
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = Result<bool, ChainClientError>>,
+{
+    let mut ticker = tokio::time::interval(interval);
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    ticker.tick().await; // the first tick fires immediately; skip it
+    loop {
+        tokio::select! { biased;
+            _ = stop_rx.recv() => return Ok(()),
+            _ = ticker.tick() => {
+                if let Err(err) = refresh().await {
+                    tracing::warn!(
+                        error = %err,
+                        "RCA EIP-712 domain refresh failed; keeping the current domain"
+                    );
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        },
+        time::Duration,
+    };
+
+    use super::run_domain_refresh;
+
+    /// The refresh loop must exit promptly when stopped, even mid-wait, so it
+    /// participates in graceful shutdown instead of being a detached task.
+    #[tokio::test]
+    async fn refresh_loop_stops_on_signal() {
+        let (tx, rx) = tokio::sync::mpsc::channel(1);
+        let calls = Arc::new(AtomicUsize::new(0));
+        let calls_in = calls.clone();
+
+        // A long interval so no refresh tick fires during the test; the stop
+        // arm is what must end the loop.
+        let handle = tokio::spawn(run_domain_refresh(
+            Duration::from_secs(3600),
+            rx,
+            move || {
+                let calls = calls_in.clone();
+                async move {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    Ok(true)
+                }
+            },
+        ));
+
+        tx.send(()).await.unwrap();
+        let result = tokio::time::timeout(Duration::from_secs(5), handle)
+            .await
+            .expect("refresh loop did not stop on signal")
+            .expect("refresh task panicked");
+
+        assert!(result.is_ok());
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            0,
+            "no refresh should have fired before the stop signal"
+        );
+    }
+}

bin/dipper-service/src/config.rs

@@ -80,6 +80,30 @@ pub struct Config {
     /// The chain client configuration (for sending on-chain transactions)
     #[serde(default)]
     pub chain_client: Option<ChainClientConfig>,
+    /// The health endpoint configuration. When unset, no health server is
+    /// started.
+    #[serde(default)]
+    pub health: Option<HealthConfig>,
+}
+
+/// Configuration for the HTTP health endpoint used by orchestrator liveness
+/// probes.
+#[serde_as]
+#[derive(Debug, serde::Deserialize)]
+pub struct HealthConfig {
+    /// The health server listen address (e.g. `0.0.0.0:8080`).
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub listen_addr: std::net::SocketAddr,
+
+    /// Staleness threshold in seconds after which the worker is reported
+    /// unhealthy. Defaults to [`crate::health::DEFAULT_HEALTH_THRESHOLD`].
+    #[serde(default = "default_health_threshold")]
+    #[serde_as(as = "serde_with::DurationSeconds")]
+    pub threshold: Duration,
+}
+
+fn default_health_threshold() -> Duration {
+    crate::health::DEFAULT_HEALTH_THRESHOLD
 }
 
 /// The IISA (Indexing Indexer Selection Algorithm) service configuration