From 3e382606da962709c7029daa78dccadee1fc8d0c Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Mon, 15 Jun 2026 13:51:38 +0000 Subject: [PATCH 1/7] fix(worker): degrade to poll-only on listener failure instead of panicking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker loop called panic! when the LISTEN/NOTIFY listener returned an error. The panic killed the worker task; main's JoinSet join loop only logged the dead task while every other service kept running, so the process looked healthy but processed zero jobs — a silent total stall. queue.pop() is independent of LISTEN/NOTIFY (the notification only wakes the loop earlier than the poll interval), so a listener fault is recoverable by construction: drop the listener, continue in poll-only mode, and re-subscribe on a later poll. Job processing never stops and the worker is never left in an uncertain state. Extract the wait-for-trigger logic into await_next_tick over a JobNotifications trait so the degrade-to-polling path is unit tested without a live Postgres. --- bin/dipper-service/src/worker/queue.rs | 16 ++- bin/dipper-service/src/worker/service.rs | 162 +++++++++++++++++++++-- 2 files changed, 164 insertions(+), 14 deletions(-) diff --git a/bin/dipper-service/src/worker/queue.rs b/bin/dipper-service/src/worker/queue.rs index f6f87363..9619f2a6 100644 --- a/bin/dipper-service/src/worker/queue.rs +++ b/bin/dipper-service/src/worker/queue.rs @@ -54,9 +54,19 @@ where /// A listener for the queue job available notification pub struct QueueImplListener(PgQueueListener); -impl QueueImplListener { - /// Waits for a new job available notification - pub async fn wait_for_notification(&mut self) -> anyhow::Result<()> { +/// A source of "a job may be available" notifications. +/// +/// Abstracted behind a trait so the worker loop's degrade-to-polling behaviour +/// can be unit tested without a live Postgres `LISTEN`/`NOTIFY` connection. +#[async_trait] +pub trait JobNotifications: Send { + /// Waits for the next job-available notification. + async fn wait_for_notification(&mut self) -> anyhow::Result<()>; +} + +#[async_trait] +impl JobNotifications for QueueImplListener { + async fn wait_for_notification(&mut self) -> anyhow::Result<()> { self.0.wait_for_notification().await } } diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 4555b0ae..45e4da0e 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -13,7 +13,7 @@ use super::{ SendIndexingAgreementProposalCtx, SubmitOfferCtx, }, messages::Message, - queue::Queue, + queue::{JobNotifications, Queue}, result::{JobError, JobResult, calculate_backoff_delay}, }; use crate::{ @@ -28,6 +28,56 @@ use crate::{ /// Default period to poll the queue for new jobs const DEFAULT_QUEUE_POLL_PERIOD: Duration = Duration::from_secs(1); +/// What the worker should do after waiting for the next trigger. +#[derive(Debug, PartialEq, Eq)] +enum Tick { + /// A stop signal was received; the worker loop should exit. + Stop, + /// Attempt to pop and process the next job. + Poll, +} + +/// Waits for the next trigger to attempt a queue poll: a stop signal, a +/// `LISTEN`/`NOTIFY` notification, or the poll-interval fallback. +/// +/// On a listener error the listener is dropped (`*listener = None`) and the +/// worker degrades to poll-only operation. This is correct, not a degraded +/// state to be feared: `queue.pop()` is independent of `LISTEN`/`NOTIFY` — the +/// notification only wakes the loop earlier than the poll interval, a latency +/// optimisation — so a listener fault never stops job processing and never +/// leaves the worker in an uncertain state. The caller re-subscribes on a +/// later poll. +async fn await_next_tick( + stop_rx: &mut mpsc::Receiver<()>, + listener: &mut Option, + poll_period: Duration, +) -> Tick { + match listener { + Some(l) => { + tokio::select! { biased; + _ = stop_rx.recv() => Tick::Stop, + res = l.wait_for_notification() => { + if let Err(err) = res { + tracing::warn!( + error=?err, + "job-available listener failed; degrading to poll-only until it can be re-established" + ); + *listener = None; + } + Tick::Poll + } + _ = tokio::time::sleep(poll_period) => Tick::Poll, + } + } + None => { + tokio::select! { biased; + _ = stop_rx.recv() => Tick::Stop, + _ = tokio::time::sleep(poll_period) => Tick::Poll, + } + } + } +} + /// Create a new worker and a future that processes jobs from the queue. /// /// The worker pulls jobs from the queue and processes them concurrently every 1 second. @@ -91,19 +141,31 @@ where }; let mut stop_rx = rx_stop; - let mut listener = queue.subscribe().await?; + // `Some` while LISTEN/NOTIFY is healthy; `None` once it has degraded to + // poll-only operation (see `await_next_tick`). + let mut listener = Some(queue.subscribe().await?); loop { - tokio::select! { biased; - _ = stop_rx.recv() => { - return Ok(()); - } - res = listener.wait_for_notification() => { - if let Err(err) = res { - tracing::error!(error=?err, "Failed to wait for job available notification"); - panic!("An unexpected error occurred while waiting for job available notification"); + match await_next_tick(&mut stop_rx, &mut listener, DEFAULT_QUEUE_POLL_PERIOD).await { + Tick::Stop => return Ok(()), + Tick::Poll => {} + } + + // If the listener degraded on a previous tick, try to re-establish + // it. This is bounded to at most once per poll period and is + // non-fatal: a failure keeps us in correct poll-only mode. + if listener.is_none() { + match queue.subscribe().await { + Ok(l) => { + tracing::info!("re-subscribed to job-available notifications"); + listener = Some(l); + } + Err(err) => { + tracing::debug!( + error=?err, + "job-available listener re-subscription failed; staying in poll-only mode" + ); } } - _ = tokio::time::sleep(DEFAULT_QUEUE_POLL_PERIOD) => {} } // Process the job @@ -218,3 +280,81 @@ impl Handle { self.tx_stop.closed().await; } } + +#[cfg(test)] +mod tests { + use std::future; + + use async_trait::async_trait; + use tokio::sync::mpsc; + + use super::{Tick, await_next_tick}; + use crate::worker::queue::JobNotifications; + + /// A listener stub whose `wait_for_notification` always errors. Models a + /// dropped/broken `LISTEN`/`NOTIFY` connection. + struct FailingNotifier; + + #[async_trait] + impl JobNotifications for FailingNotifier { + async fn wait_for_notification(&mut self) -> anyhow::Result<()> { + anyhow::bail!("simulated listener failure") + } + } + + /// A listener stub that never notifies (its future stays pending). + struct SilentNotifier; + + #[async_trait] + impl JobNotifications for SilentNotifier { + async fn wait_for_notification(&mut self) -> anyhow::Result<()> { + future::pending().await + } + } + + /// A listener error must degrade to poll-only (drop the listener) and return + /// `Poll`, never panic or stop. This is the regression test for the worker + /// `panic!` that turned a recoverable listener fault into a silent total + /// stall. + #[tokio::test] + async fn listener_error_degrades_to_poll_without_panicking() { + // Hold the sender so the stop branch stays pending. + let (_tx, mut rx) = mpsc::channel::<()>(1); + let mut listener = Some(FailingNotifier); + + let tick = + await_next_tick(&mut rx, &mut listener, std::time::Duration::from_secs(60)).await; + + assert_eq!(tick, Tick::Poll, "a listener error should still poll"); + assert!( + listener.is_none(), + "a listener error should degrade to poll-only (listener dropped)" + ); + } + + /// A pending stop signal wins over a silent listener (biased select). + #[tokio::test] + async fn stop_signal_wins() { + let (tx, mut rx) = mpsc::channel::<()>(1); + tx.send(()).await.unwrap(); + let mut listener = Some(SilentNotifier); + + let tick = + await_next_tick(&mut rx, &mut listener, std::time::Duration::from_secs(60)).await; + + assert_eq!(tick, Tick::Stop); + } + + /// With no listener (already degraded), the poll-interval fallback drives + /// the loop so job processing continues. + #[tokio::test] + async fn poll_only_mode_ticks_on_interval() { + let (_tx, mut rx) = mpsc::channel::<()>(1); + let mut listener: Option = None; + + let tick = + await_next_tick(&mut rx, &mut listener, std::time::Duration::from_millis(10)).await; + + assert_eq!(tick, Tick::Poll); + } +} From 49a8311d88b0daf7c8c3828dcc315243ca116850 Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Mon, 15 Jun 2026 13:58:57 +0000 Subject: [PATCH 2/7] feat(service): supervise the task tree and exit non-zero on unexpected task death MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The main join loop merely logged a completed/failed task and kept blocking on the rest. A critical task dying (panic, or a loop returning Err) left the process up and looking healthy while doing no useful work — the same silent- stall class the worker panic produced. Add a supervisor: every long-running service runs until the shutdown sequence stops it, so a task finishing before shutdown was requested is by definition an unexpected critical-task exit. supervise() treats the first such exit as fatal: it requests shutdown (a shared Shutdown handle the signal-handler task also waits on, so the existing reverse-order stop sequence runs), keeps draining, then returns an error so the process exits non-zero for the orchestrator to restart cleanly. Shutdown::requested_signal uses register-then-check ordering so a request racing the wait can't be lost. Tests cover both the fatal path and the deliberate-shutdown path, bounded by a timeout so a regression fails fast instead of hanging. --- bin/dipper-service/src/main.rs | 49 +++---- bin/dipper-service/src/supervisor.rs | 186 +++++++++++++++++++++++++++ 2 files changed, 213 insertions(+), 22 deletions(-) create mode 100644 bin/dipper-service/src/supervisor.rs diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 75a1ba6b..fcc02f96 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -20,6 +20,7 @@ mod indexer_rpc_client; mod network; mod registry; mod signing; +mod supervisor; mod worker; #[global_allocator] @@ -428,6 +429,10 @@ pub async fn main() -> anyhow::Result<()> { // Construct the task tree let mut task_tree = JoinSet::new(); + // Shared shutdown coordination. A signal or an unexpected task exit both + // drive the same graceful stop sequence below. + let shutdown = supervisor::Shutdown::new(); + let network_topology_task_handle = task_tree.spawn(network_topology_service); tracing::debug!(task_id=%network_topology_task_handle.id(), "Graph network topology service started"); @@ -473,17 +478,28 @@ pub async fn main() -> anyhow::Result<()> { let admin_rpc_task_handle = task_tree.spawn(admin_rpc_service); tracing::debug!(task_id=%admin_rpc_task_handle.id(), "Admin RPC service started"); + let shutdown_coordinator = shutdown.clone(); let signal_handler_task_handle = task_tree.spawn(async move { - let signal = signal_task().await; - match signal { - Ok(AppSignal::Shutdown) => { - tracing::info!("shutting down"); - } - Err(err) => { - tracing::error!(error=?err, "signal handler registration failed. shutting down"); + // Wake on either an OS signal or an unexpected critical-task exit + // (the supervisor requests shutdown in the latter case). + tokio::select! { + signal = signal_task() => match signal { + Ok(AppSignal::Shutdown) => { + tracing::info!("shutting down"); + } + Err(err) => { + tracing::error!(error=?err, "signal handler registration failed. shutting down"); + } + }, + _ = shutdown_coordinator.requested_signal() => { + tracing::warn!("shutting down due to an unexpected critical task exit"); } } + // Ensure the flag is set on the signal path too, so the supervisor + // treats the resulting service completions as a deliberate shutdown. + shutdown_coordinator.request(); + // Stop all services in reverse dependency order, so a service is stopped before the // services it depends on. tracing::trace!("stopping Admin RPC service"); @@ -541,22 +557,11 @@ pub async fn main() -> anyhow::Result<()> { }); tracing::debug!(task_id=%signal_handler_task_handle.id(), "signal handler registered"); - // Block on the task tree. Wait for all tasks to complete + // Supervise the task tree. An unexpected task exit (one that happens before + // shutdown was requested) tears the rest of the tree down and returns an + // error so the process exits non-zero for the orchestrator to restart. tracing::info!("starting service"); - while let Some(res) = task_tree.join_next_with_id().await { - match res { - Ok((id, Ok(()))) => { - tracing::debug!(task_id=%id, "task completed"); - } - Ok((id, Err(err))) => { - tracing::error!(task_id=%id, error=?err, "task failed"); - } - Err(err) => { - tracing::error!(task_id=%err.id(), error=?err, "task join error"); - } - } - } - Ok(()) + supervisor::supervise(task_tree, &shutdown).await } /// Signals that the application can receive diff --git a/bin/dipper-service/src/supervisor.rs b/bin/dipper-service/src/supervisor.rs new file mode 100644 index 00000000..edf1ccee --- /dev/null +++ b/bin/dipper-service/src/supervisor.rs @@ -0,0 +1,186 @@ +//! Process supervision for the task tree. +//! +//! Every long-running service is spawned into a single [`JoinSet`]. In normal +//! operation none of them ever completes on its own — they run until the +//! shutdown sequence stops them. So a task that finishes *before* shutdown was +//! deliberately requested is, by definition, an unexpected critical-task exit +//! (a panic, or a loop returning `Err` for an unrecoverable reason). +//! +//! Leaving the process running in that state — other services up, the daemon +//! looking healthy, but a critical task dead — is exactly the silent-stall +//! failure mode we refuse to allow. [`supervise`] therefore treats any such +//! exit as fatal: it requests shutdown so the rest of the tree is torn down +//! cleanly, then returns an error so the process exits non-zero and the +//! orchestrator restarts it. + +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; + +use tokio::{sync::Notify, task::JoinSet}; + +/// Shared shutdown coordination between [`supervise`] and the task that runs +/// the graceful stop sequence. +/// +/// Cloneable; all clones observe the same state. +#[derive(Clone, Default)] +pub struct Shutdown { + requested: Arc, + trigger: Arc, +} + +impl Shutdown { + pub fn new() -> Self { + Self::default() + } + + /// Marks shutdown as requested and wakes whoever awaits + /// [`Shutdown::requested_signal`]. Idempotent. + pub fn request(&self) { + self.requested.store(true, Ordering::SeqCst); + self.trigger.notify_one(); + } + + /// Whether shutdown has been requested (by a signal or by an unexpected + /// task exit). + pub fn is_requested(&self) -> bool { + self.requested.load(Ordering::SeqCst) + } + + /// Resolves once shutdown has been requested. + /// + /// Uses the register-then-check ordering so a [`Shutdown::request`] racing + /// between the flag read and the wait can't be lost. + pub async fn requested_signal(&self) { + let notified = self.trigger.notified(); + tokio::pin!(notified); + // Register interest before re-reading the flag. + notified.as_mut().enable(); + if self.is_requested() { + return; + } + notified.await; + } +} + +/// Drains `task_tree`, treating any task that finishes before shutdown was +/// requested as a fatal unexpected exit. +/// +/// On the first such exit it logs, requests shutdown (so the stop-sequence task +/// tears the rest of the tree down), and keeps draining. Returns `Err` if any +/// unexpected exit occurred, otherwise `Ok(())` (a deliberate shutdown). +pub async fn supervise( + mut task_tree: JoinSet>, + shutdown: &Shutdown, +) -> anyhow::Result<()> { + let mut unexpected_exit = false; + + while let Some(res) = task_tree.join_next_with_id().await { + match res { + Ok((id, Ok(()))) => { + tracing::debug!(task_id = %id, "task completed"); + } + Ok((id, Err(err))) => { + tracing::error!(task_id = %id, error = ?err, "task failed"); + } + Err(err) => { + tracing::error!(task_id = %err.id(), error = ?err, "task join error"); + } + } + + if !shutdown.is_requested() { + tracing::error!( + "a critical task exited unexpectedly; initiating shutdown so the process \ + restarts rather than running on with a dead task" + ); + unexpected_exit = true; + shutdown.request(); + } + } + + if unexpected_exit { + anyhow::bail!("a critical task exited unexpectedly"); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::task::JoinSet; + + use super::{Shutdown, supervise}; + + /// Bound on how long `supervise` may run in a test. A broken impl that + /// never requests shutdown would otherwise hang the coordinator stand-in + /// forever; this turns that into a deterministic failure. + const SUPERVISE_TIMEOUT: Duration = Duration::from_secs(5); + + /// A task finishing before shutdown was requested is fatal: `supervise` + /// returns an error and requests shutdown so the rest of the tree is torn + /// down. Regression test for the join loop that merely logged a dead task + /// and let the process keep running. + #[tokio::test] + async fn unexpected_task_exit_is_fatal_and_triggers_shutdown() { + let shutdown = Shutdown::new(); + let mut task_tree: JoinSet> = JoinSet::new(); + + // Stand-in for the stop-sequence task: drains once shutdown is asked + // for. If `supervise` fails to request shutdown, this never completes + // and the test hangs (caught by the harness timeout). + let coordinator = shutdown.clone(); + task_tree.spawn(async move { + coordinator.requested_signal().await; + Ok(()) + }); + + // A critical task that exits on its own, with nobody having asked for + // shutdown. + task_tree.spawn(async { Ok(()) }); + + let result = tokio::time::timeout(SUPERVISE_TIMEOUT, supervise(task_tree, &shutdown)) + .await + .expect("supervise did not request shutdown; coordinator never drained"); + + assert!(result.is_err(), "an unexpected task exit must be fatal"); + assert!( + shutdown.is_requested(), + "an unexpected task exit must request shutdown" + ); + } + + /// When shutdown was deliberately requested, tasks completing afterwards is + /// the normal path and must not be reported as fatal. + #[tokio::test] + async fn deliberate_shutdown_is_not_fatal() { + let shutdown = Shutdown::new(); + let mut task_tree: JoinSet> = JoinSet::new(); + + // The "signal handler": requests shutdown, then completes. + let coordinator = shutdown.clone(); + task_tree.spawn(async move { + coordinator.request(); + Ok(()) + }); + + // Other services that exit only after shutdown is requested. + for _ in 0..3 { + let s = shutdown.clone(); + task_tree.spawn(async move { + s.requested_signal().await; + Ok(()) + }); + } + + let result = tokio::time::timeout(SUPERVISE_TIMEOUT, supervise(task_tree, &shutdown)) + .await + .expect("supervise did not drain after a deliberate shutdown"); + + assert!( + result.is_ok(), + "a deliberate shutdown must not be reported as fatal: {result:?}" + ); + } +} From 6b35342274cf578bb3a63cf790f4ce8e7a10a0d1 Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:01:42 +0000 Subject: [PATCH 3/7] fix(worker): bound process_job with a backstop timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker holds the pgmq transaction (and the row's Running lock + a pooled DB connection) open across the entire process_job call, which makes external calls. Although each external call is individually bounded, there was no overall timeout: a dependency that accepts the connection but never responds would pin those resources and wedge the single sequential worker indefinitely. Wrap process_job in a 300s backstop timeout — comfortably above the legitimate worst case (summed per-call timeouts, ~couple of minutes) so it only fires on a genuine hang. On elapse the job becomes a retryable error: it reschedules and the JobGuard drops, rolling back the transaction and releasing the lock and connection. Recovery is idempotent (chain-as-source-of-truth), so re-running is safe. Extract run_job_with_timeout so the timeout-to-retryable mapping and result pass-through are unit tested without a live job. --- bin/dipper-service/src/worker/service.rs | 74 +++++++++++++++++++++++- 1 file changed, 72 insertions(+), 2 deletions(-) diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 45e4da0e..f4ed4430 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -28,6 +28,41 @@ use crate::{ /// Default period to poll the queue for new jobs const DEFAULT_QUEUE_POLL_PERIOD: Duration = Duration::from_secs(1); +/// Backstop timeout for processing a single job. +/// +/// Every external call `process_job` makes is already individually bounded +/// (IISA HTTP, indexer RPC, chain RPC + receipt polling), so the legitimate +/// worst case is their sum — on the order of a couple of minutes. This timeout +/// sits comfortably above that and only fires if a dependency accepts the +/// connection but never responds, defeating the per-call timeouts. Critically, +/// the worker holds the pgmq transaction (and the row's `Running` lock and a +/// pooled DB connection) open for the whole `process_job` call, so an +/// unbounded hang would pin those resources and wedge the single worker +/// forever. On elapse the in-flight `process_job` future is cancelled (dropped) +/// and the job is rescheduled via its `JobGuard`, releasing the pinned +/// resources. Recovery is idempotent (chain-as-source-of-truth), so re-running +/// a job whose handler was cancelled mid-flight is safe. +pub(crate) const PROCESS_JOB_TIMEOUT: Duration = Duration::from_secs(300); + +/// Base backoff for a job rescheduled after hitting [`PROCESS_JOB_TIMEOUT`]. +const JOB_TIMEOUT_RETRY_BASE_DELAY: Duration = Duration::from_secs(30); + +/// Runs a `process_job` future under [`PROCESS_JOB_TIMEOUT`]. On elapse it +/// cancels the in-flight future and returns a retryable error so the worker +/// reschedules the job (via its `JobGuard`) rather than blocking indefinitely. +async fn run_job_with_timeout(timeout: Duration, fut: F) -> JobResult<()> +where + F: std::future::Future>, +{ + match tokio::time::timeout(timeout, fut).await { + Ok(res) => res, + Err(_elapsed) => Err(JobError::Retryable( + anyhow::anyhow!("job processing exceeded the {timeout:?} backstop timeout"), + JOB_TIMEOUT_RETRY_BASE_DELAY, + )), + } +} + /// What the worker should do after waiting for the next trigger. #[derive(Debug, PartialEq, Eq)] enum Tick { @@ -179,7 +214,7 @@ where }; let _span = tracing::debug_span!("process_job", job = %job.id()); - match process_job(&state, job.desc()).await { + match run_job_with_timeout(PROCESS_JOB_TIMEOUT, process_job(&state, job.desc())).await { Ok(..) => { if let Err(err) = job.remove().await { tracing::debug!(error=?err, "Failed to remove job from queue"); @@ -288,7 +323,7 @@ mod tests { use async_trait::async_trait; use tokio::sync::mpsc; - use super::{Tick, await_next_tick}; + use super::{JobError, Tick, await_next_tick, run_job_with_timeout}; use crate::worker::queue::JobNotifications; /// A listener stub whose `wait_for_notification` always errors. Models a @@ -357,4 +392,39 @@ mod tests { assert_eq!(tick, Tick::Poll); } + + /// A job that hangs past the timeout must be turned into a retryable error + /// (so it reschedules and the open transaction rolls back), not awaited + /// indefinitely. Bounded by a test-level timeout so a regression fails fast + /// instead of hanging. + #[tokio::test] + async fn hung_job_times_out_as_retryable() { + let hung = future::pending::>(); + + let result = tokio::time::timeout( + std::time::Duration::from_secs(2), + run_job_with_timeout(std::time::Duration::from_millis(50), hung), + ) + .await + .expect("run_job_with_timeout did not bound the hung job"); + + assert!( + matches!(result, Err(JobError::Retryable(..))), + "a timed-out job should be retryable, got {result:?}" + ); + } + + /// A job that completes within the timeout passes its result through + /// unchanged (including a Fatal classification). + #[tokio::test] + async fn fast_job_result_passes_through() { + let ok = run_job_with_timeout(std::time::Duration::from_secs(60), async { Ok(()) }).await; + assert!(ok.is_ok()); + + let fatal = run_job_with_timeout(std::time::Duration::from_secs(60), async { + Err(JobError::Fatal(anyhow::anyhow!("boom"))) + }) + .await; + assert!(matches!(fatal, Err(JobError::Fatal(..)))); + } } From d94b9498abdb7ff6cd2ccdc24b12c2b2fa0363b7 Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:07:47 +0000 Subject: [PATCH 4/7] feat(service): add worker liveness watermark and HTTP health endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The supervisor catches a worker that exits, but not one that is wedged — alive yet making no progress. Nothing external could observe that state: there was no health or readiness endpoint at all. Add a Liveness watermark the worker ticks every loop iteration (including idle polls), and a minimal HTTP health server (axum, reusing the hyper stack already linked by the RPC servers) that returns 503 once the watermark is older than a threshold, else 200. An orchestrator liveness probe can then restart a wedged process. The threshold defaults to 600s — above the worst-case gap between ticks (poll period + the 300s process_job backstop) so a legitimately slow job never trips it. Gated behind optional [health] config (listen_addr + threshold); no endpoint is started when unset. The server is spawned into the supervised task tree and stopped first in the graceful shutdown sequence. The staleness predicate and the 200/503 responses are covered by unit and TCP integration tests. --- Cargo.lock | 4 + bin/dipper-service/Cargo.toml | 1 + bin/dipper-service/src/config.rs | 24 ++ bin/dipper-service/src/health.rs | 288 +++++++++++++++++++++++ bin/dipper-service/src/main.rs | 50 ++++ bin/dipper-service/src/worker/context.rs | 4 + bin/dipper-service/src/worker/service.rs | 6 + k8s/configmap-example.yaml | 4 + k8s/deployment.yaml | 12 +- 9 files changed, 391 insertions(+), 2 deletions(-) create mode 100644 bin/dipper-service/src/health.rs diff --git a/Cargo.lock b/Cargo.lock index d58a7e4e..8429adb6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1429,6 +1429,8 @@ dependencies = [ "http 1.4.0", "http-body 1.0.1", "http-body-util", + "hyper 1.8.1", + "hyper-util", "itoa", "matchit", "memchr", @@ -1437,6 +1439,7 @@ dependencies = [ "pin-project-lite", "serde_core", "sync_wrapper 1.0.2", + "tokio", "tower", "tower-layer", "tower-service", @@ -2425,6 +2428,7 @@ dependencies = [ "anyhow", "async-signal", "async-trait", + "axum", "custom_debug", "dashmap", "dipper-core", diff --git a/bin/dipper-service/Cargo.toml b/bin/dipper-service/Cargo.toml index 423eefe7..e8b66809 100644 --- a/bin/dipper-service/Cargo.toml +++ b/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" } diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 101c35c7..87c11c17 100644 --- a/bin/dipper-service/src/config.rs +++ b/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, + /// The health endpoint configuration. When unset, no health server is + /// started. + #[serde(default)] + pub health: Option, +} + +/// 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 diff --git a/bin/dipper-service/src/health.rs b/bin/dipper-service/src/health.rs new file mode 100644 index 00000000..d6ef865c --- /dev/null +++ b/bin/dipper-service/src/health.rs @@ -0,0 +1,288 @@ +//! Worker liveness watermark and a minimal HTTP health endpoint. +//! +//! The supervisor ([`crate::supervisor`]) catches a worker that *exits* +//! (panic or error return). It cannot catch a worker that is *wedged* — alive +//! but making no progress (e.g. parked on an await that the per-call timeouts +//! somehow didn't bound). This module closes that gap: the worker ticks a +//! progress watermark every loop iteration, and a small health server reports +//! 503 once the watermark goes stale so an external orchestrator (k8s liveness +//! probe) can restart the wedged process. + +use std::{ + net::SocketAddr, + sync::{ + Arc, + atomic::{AtomicI64, Ordering}, + }, + time::Duration, +}; + +use axum::{Router, extract::State, http::StatusCode, response::IntoResponse, routing::get}; +use time::OffsetDateTime; +use tokio::{net::TcpListener, sync::mpsc}; + +/// Default staleness threshold after which the worker is considered wedged. +/// +/// Must exceed the worker's worst-case time between progress ticks: the poll +/// period plus a single job's backstop timeout +/// ([`crate::worker::service`]'s `PROCESS_JOB_TIMEOUT`, 300s). 600s leaves +/// comfortable headroom so a legitimately slow job never trips the probe. +pub const DEFAULT_HEALTH_THRESHOLD: Duration = Duration::from_secs(600); + +/// Shared liveness watermark: the unix-seconds timestamp at which the worker +/// last made progress. The worker ticks it via [`Liveness::record_progress`]; +/// the health server reads it via [`Liveness::is_healthy`]. +#[derive(Clone)] +pub struct Liveness { + last_progress: Arc, +} + +fn now_unix() -> i64 { + OffsetDateTime::now_utc().unix_timestamp() +} + +impl Liveness { + /// Creates a watermark seeded to now, so a freshly started worker is + /// considered live (startup grace). + pub fn new() -> Self { + Self { + last_progress: Arc::new(AtomicI64::new(now_unix())), + } + } + + /// Records that the worker just made progress. + pub fn record_progress(&self) { + self.last_progress.store(now_unix(), Ordering::Relaxed); + } + + /// Whether the worker made progress within `threshold` of `now_unix`. + /// + /// Pure in its inputs so it is unit testable without touching the clock. + pub fn is_healthy_at(&self, now_unix: i64, threshold: Duration) -> bool { + let age = now_unix.saturating_sub(self.last_progress.load(Ordering::Relaxed)); + age <= threshold.as_secs() as i64 + } + + /// [`Liveness::is_healthy_at`] against the current clock. + pub fn is_healthy(&self, threshold: Duration) -> bool { + self.is_healthy_at(now_unix(), threshold) + } +} + +impl Default for Liveness { + fn default() -> Self { + Self::new() + } +} + +/// Handle to stop the health server. +pub struct Handle { + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Stops the health server. + pub async fn stop(self) { + if self.tx_stop.is_closed() { + return; + } + let _ = self.tx_stop.send(()).await; + self.tx_stop.closed().await; + } +} + +/// Binds the health server and returns a stop handle plus its run future. +/// +/// The listener is bound eagerly so a bind failure surfaces at startup (and so +/// callers/tests can read the actual bound address). +pub async fn new( + addr: SocketAddr, + liveness: Liveness, + threshold: Duration, +) -> anyhow::Result<( + Handle, + SocketAddr, + impl std::future::Future>, +)> { + let listener = TcpListener::bind(addr).await?; + let local_addr = listener.local_addr()?; + let (tx_stop, rx_stop) = mpsc::channel(1); + let fut = serve(listener, liveness, threshold, rx_stop); + Ok((Handle { tx_stop }, local_addr, fut)) +} + +/// State shared with the [`health`] handler. +#[derive(Clone)] +struct HealthState { + liveness: Liveness, + threshold: Duration, +} + +/// `GET /health`: 200 while the worker watermark is fresh, 503 once it has gone +/// stale (the worker is wedged and the orchestrator should restart it). +async fn health(State(state): State) -> impl IntoResponse { + if state.liveness.is_healthy(state.threshold) { + (StatusCode::OK, "ok") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "worker stalled") + } +} + +/// Serves health requests until `stop_rx` fires. +/// +/// Backed by axum/hyper (already in the dependency tree via the RPC servers), +/// so request parsing, per-connection isolation, connection timeouts and +/// graceful shutdown are handled by the HTTP stack rather than by hand. +async fn serve( + listener: TcpListener, + liveness: Liveness, + threshold: Duration, + mut stop_rx: mpsc::Receiver<()>, +) -> anyhow::Result<()> { + tracing::info!(addr = %listener.local_addr()?, "health server listening"); + let app = Router::new() + .route("/health", get(health)) + .with_state(HealthState { + liveness, + threshold, + }); + axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = stop_rx.recv().await; + }) + .await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, + }; + + use super::*; + + #[test] + fn fresh_watermark_is_healthy() { + let liveness = Liveness::new(); + let now = now_unix(); + assert!(liveness.is_healthy_at(now, Duration::from_secs(600))); + } + + #[test] + fn stale_watermark_is_unhealthy() { + let liveness = Liveness::new(); + // Pretend "now" is well past the threshold since the watermark was set. + let now = now_unix() + 10_000; + assert!( + !liveness.is_healthy_at(now, Duration::from_secs(600)), + "a watermark older than the threshold must report unhealthy" + ); + } + + #[test] + fn boundary_is_inclusive_healthy() { + let liveness = Liveness::new(); + let base = now_unix(); + // Exactly at the threshold is still healthy; one second past is not. + assert!(liveness.is_healthy_at(base + 600, Duration::from_secs(600))); + assert!(!liveness.is_healthy_at(base + 601, Duration::from_secs(600))); + } + + /// Reads an HTTP response's status line from a fresh connection to the + /// server. + async fn probe(addr: SocketAddr) -> String { + let mut stream = TcpStream::connect(addr).await.unwrap(); + // `Connection: close` so the server closes the socket after responding + // and `read_to_end` returns rather than blocking on HTTP/1.1 keep-alive. + stream + .write_all(b"GET /health HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).await.unwrap(); + let text = String::from_utf8_lossy(&buf); + text.lines().next().unwrap_or_default().to_string() + } + + #[tokio::test] + async fn server_reports_503_when_stalled() { + let liveness = Liveness::new(); + // A zero threshold makes the (initially fresh) watermark immediately + // stale, so we exercise the unhealthy branch deterministically. + let (handle, addr, fut) = new("127.0.0.1:0".parse().unwrap(), liveness, Duration::ZERO) + .await + .unwrap(); + // Ensure the watermark is at least a second old so age > 0. + let server = tokio::spawn(fut); + + // Wait a moment so the seeded watermark is strictly in the past. + tokio::time::sleep(Duration::from_millis(1100)).await; + let status = probe(addr).await; + assert!( + status.contains("503"), + "expected 503 when stalled, got: {status:?}" + ); + + handle.stop().await; + let _ = server.await; + } + + #[tokio::test] + async fn server_reports_200_when_live() { + let liveness = Liveness::new(); + let (handle, addr, fut) = new( + "127.0.0.1:0".parse().unwrap(), + liveness, + Duration::from_secs(600), + ) + .await + .unwrap(); + let server = tokio::spawn(fut); + + let status = probe(addr).await; + assert!( + status.contains("200"), + "expected 200 when live, got: {status:?}" + ); + + handle.stop().await; + let _ = server.await; + } + + /// A client that connects but never sends must not wedge the server: other + /// probes are still answered promptly, and graceful shutdown still + /// completes. hyper isolates each connection, but this guards against a + /// future regression that serves connections without that isolation. + #[tokio::test] + async fn stalled_client_does_not_block_other_probes() { + let liveness = Liveness::new(); + let (handle, addr, fut) = new( + "127.0.0.1:0".parse().unwrap(), + liveness, + Duration::from_secs(600), + ) + .await + .unwrap(); + let server = tokio::spawn(fut); + + // Open a connection and never write to it; its server-side read blocks. + let _stalled = TcpStream::connect(addr).await.unwrap(); + + // A well-behaved probe must still get a response without waiting on the + // stalled client (well under CONN_TIMEOUT). + let status = tokio::time::timeout(Duration::from_secs(2), probe(addr)) + .await + .expect("a stalled client blocked an independent probe"); + assert!(status.contains("200"), "expected 200, got: {status:?}"); + + // Shutdown must not wait on the stalled connection either. + tokio::time::timeout(Duration::from_secs(2), handle.stop()) + .await + .expect("a stalled client blocked shutdown"); + let _ = server.await; + } +} diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index fcc02f96..b7ac62d2 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -16,6 +16,7 @@ mod admin_rpc_server; mod chain_client; mod config; mod db; +mod health; mod indexer_rpc_client; mod network; mod registry; @@ -303,6 +304,9 @@ pub async fn main() -> anyhow::Result<()> { // dispatched so it switches from 300s idle polling to 5s immediately. let chain_listener_notify = Arc::new(tokio::sync::Notify::new()); + //- Worker liveness watermark, shared with the health endpoint below. + let worker_liveness = health::Liveness::new(); + //- The worker service let (worker_handle, worker_service) = { let ctx = worker::Ctx { @@ -325,6 +329,7 @@ pub async fn main() -> anyhow::Result<()> { .map(|c| c.bypass_chain_clock_defenses) .unwrap_or(false), chain_listener_chain_id: conf.chain_listener.as_ref().map(|c| c.chain_id), + liveness: worker_liveness.clone(), }; worker::service::new(ctx) }; @@ -426,6 +431,34 @@ pub async fn main() -> anyhow::Result<()> { }; tracing::info!("initialized Admin RPC service"); + //- The health endpoint (optional). Reports 503 once the worker watermark + // goes stale so an orchestrator can restart a wedged process. + let health_handle = if let Some(health_conf) = conf.health.as_ref() { + // The worker only ticks the watermark once per loop iteration, so the + // worst-case gap between ticks is one job's backstop timeout. A + // threshold at or below that would trip the probe during a legitimately + // slow job and cause restart loops; warn rather than fail so a + // deliberate aggressive setting is still possible. + if health_conf.threshold <= worker::service::PROCESS_JOB_TIMEOUT { + tracing::warn!( + threshold_secs = health_conf.threshold.as_secs(), + process_job_timeout_secs = worker::service::PROCESS_JOB_TIMEOUT.as_secs(), + "health threshold is at or below the worker's per-job backstop timeout; a slow \ + job may trip the liveness probe and cause spurious restarts" + ); + } + let (handle, addr, service) = health::new( + health_conf.listen_addr, + worker_liveness.clone(), + health_conf.threshold, + ) + .await?; + tracing::info!(%addr, "initialized health endpoint"); + Some((handle, service)) + } else { + None + }; + // Construct the task tree let mut task_tree = JoinSet::new(); @@ -475,6 +508,15 @@ pub async fn main() -> anyhow::Result<()> { None }; + // Spawn the health endpoint if enabled + let health_stop_handle = if let Some((handle, service)) = health_handle { + let task_handle = task_tree.spawn(service); + tracing::debug!(task_id=%task_handle.id(), "Health endpoint started"); + Some(handle) + } else { + None + }; + let admin_rpc_task_handle = task_tree.spawn(admin_rpc_service); tracing::debug!(task_id=%admin_rpc_task_handle.id(), "Admin RPC service started"); @@ -502,6 +544,14 @@ pub async fn main() -> anyhow::Result<()> { // Stop all services in reverse dependency order, so a service is stopped before the // services it depends on. + + // Stop the health endpoint first; nothing depends on it. + if let Some(handle) = health_stop_handle { + tracing::trace!("stopping Health endpoint"); + handle.stop().await; + tracing::trace!("stopped Health endpoint"); + } + tracing::trace!("stopping Admin RPC service"); admin_rpc_handle.stop().await; tracing::trace!("stopped Admin RPC service"); diff --git a/bin/dipper-service/src/worker/context.rs b/bin/dipper-service/src/worker/context.rs index 614df1e3..f0f7bf05 100644 --- a/bin/dipper-service/src/worker/context.rs +++ b/bin/dipper-service/src/worker/context.rs @@ -109,6 +109,10 @@ pub struct Ctx { /// `last_processed_block_timestamp` when bypass is on. `None` /// when the chain_listener is not configured. pub chain_listener_chain_id: Option, + + /// Liveness watermark the worker ticks each loop iteration so the health + /// endpoint can detect a wedged worker. + pub liveness: crate::health::Liveness, } /// The inner worker context. diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index f4ed4430..9ca59e3b 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -148,6 +148,7 @@ where chain_listener_notify, bypass_chain_clock_defenses, chain_listener_chain_id, + liveness, } = state.into(); let (tx_stop, rx_stop) = mpsc::channel(1); @@ -185,6 +186,11 @@ where Tick::Poll => {} } + // Tick the liveness watermark on every iteration (including idle + // polls) so the health endpoint sees a live worker. A job that runs + // up to PROCESS_JOB_TIMEOUT is the longest gap between ticks. + liveness.record_progress(); + // If the listener degraded on a previous tick, try to re-establish // it. This is bounded to at most once per poll period and is // non-fatal: a failure keeps us in correct poll-only mode. diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 3f298375..ecd91d70 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -106,5 +106,9 @@ data: "gas_buffer_multiplier": 2.0, "gas_floor": 100000, "gas_max_addition": 200000 + }, + "health": { + "listen_addr": "0.0.0.0:8546", + "threshold": 600 } } diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 9b5c14b4..d8846441 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -31,15 +31,23 @@ spec: - name: indexer-rpc containerPort: 50051 protocol: TCP + - name: health + containerPort: 8546 + protocol: TCP securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL + # Probe the worker health endpoint, not just a TCP socket: it returns + # 503 once the worker watermark goes stale, so a wedged worker (alive + # but making no progress) is restarted. A plain admin-rpc TCP check + # stays up in that state and would never catch it. livenessProbe: - tcpSocket: - port: admin-rpc + httpGet: + path: /health + port: health initialDelaySeconds: 30 periodSeconds: 10 timeoutSeconds: 5 From c7a730c50af42c3f229204fbfd44e5edc66092d5 Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:11:27 +0000 Subject: [PATCH 5/7] refactor(service): supervise the RCA domain refresh instead of detaching it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The domain refresh was the one task started with a bare tokio::spawn: no stop handle, outside the task tree, unaffected by graceful shutdown, and invisible to the new supervisor. A panic in it would silently stop domain refreshes, letting dipper sign offers under a stale EIP-712 domain after a contract upgrade — an incorrect state with no signal. Extract the loop into chain_client::run_domain_refresh (generic over the refresh action, with a stop channel) and spawn it into the supervised task tree with the others. It now stops in the graceful shutdown sequence and an unexpected exit is caught by the supervisor. A test covers the stop wiring without a live chain. --- bin/dipper-service/src/chain_client.rs | 82 ++++++++++++++++++++++++++ bin/dipper-service/src/main.rs | 65 ++++++++++++-------- 2 files changed, 123 insertions(+), 24 deletions(-) diff --git a/bin/dipper-service/src/chain_client.rs b/bin/dipper-service/src/chain_client.rs index 40da889c..4870fb48 100644 --- a/bin/dipper-service/src/chain_client.rs +++ b/bin/dipper-service/src/chain_client.rs @@ -145,3 +145,85 @@ impl ChainClient for Arc { (**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( + interval: std::time::Duration, + mut stop_rx: tokio::sync::mpsc::Receiver<()>, + mut refresh: F, +) -> anyhow::Result<()> +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + 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" + ); + } +} diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index b7ac62d2..5f9ee2a8 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -97,31 +97,32 @@ pub async fn main() -> anyhow::Result<()> { // Background refresh so a running dipper follows an in-place contract upgrade // without a restart. Refresh failures keep the current domain and only warn. - if let Some(cfg) = conf.chain_client.as_ref().filter(|cfg| cfg.enabled) { - let cfg = cfg.clone(); - let rca_domain = Arc::clone(&rca_domain); - tokio::spawn(async move { - let mut ticker = tokio::time::interval(cfg.domain_refresh_interval); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - ticker.tick().await; // the first tick fires immediately; skip it - loop { - ticker.tick().await; - if let Err(err) = chain_client::refresh_rca_eip712_domain( - &cfg, - chain_id, - recurring_collector, - &rca_domain, - ) - .await - { - tracing::warn!( - error = %err, - "RCA EIP-712 domain refresh failed; keeping the current domain" - ); + // Built here but spawned into the supervised task tree below, with a stop + // handle, rather than detached — so it shares the same shutdown and + // supervision as every other long-running task. + let domain_refresh_handle = + if let Some(cfg) = conf.chain_client.as_ref().filter(|cfg| cfg.enabled) { + let cfg = cfg.clone(); + let rca_domain = Arc::clone(&rca_domain); + let interval = cfg.domain_refresh_interval; + let (tx_stop, rx_stop) = tokio::sync::mpsc::channel(1); + let service = chain_client::run_domain_refresh(interval, rx_stop, move || { + let cfg = cfg.clone(); + let rca_domain = Arc::clone(&rca_domain); + async move { + chain_client::refresh_rca_eip712_domain( + &cfg, + chain_id, + recurring_collector, + &rca_domain, + ) + .await } - } - }); - } + }); + Some((tx_stop, service)) + } else { + None + }; // Initialize the different components @@ -472,6 +473,15 @@ pub async fn main() -> anyhow::Result<()> { let worker_task_handle = task_tree.spawn(worker_service); tracing::debug!(task_id=%worker_task_handle.id(), "Worker service started"); + // Spawn the RCA domain refresh if the chain client is enabled + let domain_refresh_stop_handle = if let Some((tx_stop, service)) = domain_refresh_handle { + let task_handle = task_tree.spawn(service); + tracing::debug!(task_id=%task_handle.id(), "RCA domain refresh started"); + Some(tx_stop) + } else { + None + }; + // Spawn the reassignment service if enabled let reassignment_stop_handle = if let Some((handle, service)) = reassignment_handle { let task_handle = task_tree.spawn(service); @@ -591,6 +601,13 @@ pub async fn main() -> anyhow::Result<()> { tracing::trace!("stopped Entity count cache service"); } + // Stop the RCA domain refresh (a background helper to the chain client) + if let Some(tx_stop) = domain_refresh_stop_handle { + tracing::trace!("stopping RCA domain refresh"); + let _ = tx_stop.send(()).await; + tracing::trace!("stopped RCA domain refresh"); + } + tracing::trace!("stopping Worker service"); worker_handle.stop().await; tracing::trace!("stopped Worker service"); From c570c6a7b3a827a44e8b7b97028b388081953275 Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Tue, 16 Jun 2026 16:36:34 +0000 Subject: [PATCH 6/7] docs(worker): correct stale concurrency wording in process_job timeout comments --- bin/dipper-service/src/worker/service.rs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/bin/dipper-service/src/worker/service.rs b/bin/dipper-service/src/worker/service.rs index 9ca59e3b..6dffb61d 100644 --- a/bin/dipper-service/src/worker/service.rs +++ b/bin/dipper-service/src/worker/service.rs @@ -35,13 +35,15 @@ const DEFAULT_QUEUE_POLL_PERIOD: Duration = Duration::from_secs(1); /// worst case is their sum — on the order of a couple of minutes. This timeout /// sits comfortably above that and only fires if a dependency accepts the /// connection but never responds, defeating the per-call timeouts. Critically, -/// the worker holds the pgmq transaction (and the row's `Running` lock and a -/// pooled DB connection) open for the whole `process_job` call, so an -/// unbounded hang would pin those resources and wedge the single worker -/// forever. On elapse the in-flight `process_job` future is cancelled (dropped) -/// and the job is rescheduled via its `JobGuard`, releasing the pinned -/// resources. Recovery is idempotent (chain-as-source-of-truth), so re-running -/// a job whose handler was cancelled mid-flight is safe. +/// the worker processes jobs one at a time, and for the whole `process_job` +/// call the job's `JobGuard` holds the row's `Running` lock (and the pgmq +/// transaction behind it). An unbounded hang would therefore both wedge the +/// single worker and pin that row indefinitely. On elapse the in-flight +/// `process_job` future is cancelled (dropped), which unblocks the worker, and +/// the timeout is surfaced as a retryable error so the `JobGuard` reschedules +/// the row and releases its lock. Recovery is idempotent +/// (chain-as-source-of-truth), so re-running a job whose handler was cancelled +/// mid-flight is safe. pub(crate) const PROCESS_JOB_TIMEOUT: Duration = Duration::from_secs(300); /// Base backoff for a job rescheduled after hitting [`PROCESS_JOB_TIMEOUT`]. @@ -115,7 +117,8 @@ async fn await_next_tick( /// Create a new worker and a future that processes jobs from the queue. /// -/// The worker pulls jobs from the queue and processes them concurrently every 1 second. +/// The worker pulls jobs from the queue and processes them one at a time, woken +/// by a `LISTEN`/`NOTIFY` notification or, failing that, polled every 1 second. pub fn new(state: S) -> (Handle, impl Future>) where Q: Queue + Clone + Send + Sync, From 7f3afd07744a1b65df4d045200c69af9e663b4af Mon Sep 17 00:00:00 2001 From: Rembrandt Kuipers <50174308+RembrandtK@users.noreply.github.com> Date: Wed, 17 Jun 2026 06:09:45 +0000 Subject: [PATCH 7/7] fix(service): wake all parked shutdown waiters, not just one Shutdown::request used Notify::notify_one, which wakes a single parked waiter. requested_signal is a general-purpose, cloneable signal that multiple tasks may await concurrently, so a shutdown could leave all but one waiter hanging forever. Production wires a single awaiter today, so this was latent, but the abstraction is wrong as written. Switch to notify_waiters, which wakes every parked waiter. It stores no permit for future waiters, but none is needed: the register-then-check in requested_signal returns via the bool flag for any waiter that arrives after the request. Adds a deterministic regression test that parks two waiters and asserts a single request wakes both. --- bin/dipper-service/src/supervisor.rs | 44 +++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/bin/dipper-service/src/supervisor.rs b/bin/dipper-service/src/supervisor.rs index edf1ccee..ae205c05 100644 --- a/bin/dipper-service/src/supervisor.rs +++ b/bin/dipper-service/src/supervisor.rs @@ -39,7 +39,12 @@ impl Shutdown { /// [`Shutdown::requested_signal`]. Idempotent. pub fn request(&self) { self.requested.store(true, Ordering::SeqCst); - self.trigger.notify_one(); + // `notify_waiters`, not `notify_one`: multiple tasks may be parked in + // `requested_signal`, and shutdown must wake all of them. It stores no + // permit for future waiters, but none is needed — a waiter that arrives + // after this sees the flag via the register-then-check in + // `requested_signal` and returns without parking. + self.trigger.notify_waiters(); } /// Whether shutdown has been requested (by a signal or by an unexpected @@ -183,4 +188,41 @@ mod tests { "a deliberate shutdown must not be reported as fatal: {result:?}" ); } + + /// A single `request()` must wake *every* parked waiter, not just one. + /// + /// Both waiters are polled to their suspended (registered) state while the + /// flag is still false, then `request()` is called once. Regression test + /// for `notify_one`, which wakes only the first waiter and leaves the rest + /// hanging forever. + #[test] + fn request_wakes_all_concurrent_waiters() { + use std::{ + future::Future, + pin::pin, + task::{Context, Poll}, + }; + + let shutdown = Shutdown::new(); + let waker = std::task::Waker::noop(); + let mut cx = Context::from_waker(waker); + + let mut w1 = pin!(shutdown.requested_signal()); + let mut w2 = pin!(shutdown.requested_signal()); + + // With the flag still false, both register on the notify and suspend. + assert!(matches!(w1.as_mut().poll(&mut cx), Poll::Pending)); + assert!(matches!(w2.as_mut().poll(&mut cx), Poll::Pending)); + + shutdown.request(); + + assert!( + matches!(w1.as_mut().poll(&mut cx), Poll::Ready(())), + "the first parked waiter must be woken by request()" + ); + assert!( + matches!(w2.as_mut().poll(&mut cx), Poll::Ready(())), + "every parked waiter must be woken by request(), not just the first" + ); + } }