openzeppelin_relayer/bootstrap/
runtime.rs

1//! Pipeline runtime configuration.
2//!
3//! Resolves the worker-thread budget for the two runtimes the relayer runs:
4//! the actix HTTP server (`actix_workers`) and the explicit multi-thread tokio
5//! runtime that hosts the background transaction pipeline (`tokio_worker_threads`).
6//!
7//! Per `contracts/runtime-config.md`, both counts are pinned from the environment
8//! and SHOULD sum to no more than the container vCPU quota. Auto-detection via
9//! [`std::thread::available_parallelism`] returns the cgroup-aware core count on
10//! Rust 1.91 for cgroup v1/v2 hard quotas, but NOT for AWS Fargate's `cpu.shares`
11//! — so the env-var pins are the authoritative control on all targets.
12
13use std::env;
14use std::thread::available_parallelism;
15use std::time::Duration;
16
17use tracing::{info, warn};
18
19use crate::queues::WorkerHandle;
20
21const ENV_TOKIO_WORKER_THREADS: &str = "TOKIO_WORKER_THREADS";
22const ENV_ACTIX_WORKERS: &str = "ACTIX_WORKERS";
23
24/// Resolved worker-thread budget for the relayer's runtimes.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub struct RuntimeConfig {
27    /// Detected/assumed container vCPU quota (best-effort; host cores on Fargate).
28    pub vcpu: usize,
29    /// HTTP server worker count (pins actix `.workers()`).
30    pub actix_workers: usize,
31    /// Pipeline runtime worker-thread count.
32    pub tokio_worker_threads: usize,
33}
34
35impl RuntimeConfig {
36    /// Resolve from the environment, detecting the vCPU quota for the defaults.
37    pub fn from_env() -> Self {
38        Self::resolve(
39            detect_vcpu(),
40            read_usize_env(ENV_ACTIX_WORKERS),
41            read_usize_env(ENV_TOKIO_WORKER_THREADS),
42        )
43    }
44
45    /// Pure budget resolution (separated from env/IO for testability).
46    ///
47    /// Defaults (per the runtime-config contract):
48    /// - `actix_workers` = `max(1, vcpu / 2)`
49    /// - `tokio_worker_threads` = `max(1, vcpu - actix_workers)`
50    ///
51    /// Explicit overrides of `0` are ignored (treated as unset) so a misconfigured
52    /// `=0` can never produce a zero-thread runtime.
53    fn resolve(vcpu: usize, actix_override: Option<usize>, tokio_override: Option<usize>) -> Self {
54        let vcpu = vcpu.max(1);
55        let actix_workers = actix_override
56            .filter(|n| *n > 0)
57            .unwrap_or_else(|| (vcpu / 2).max(1));
58        let tokio_worker_threads = tokio_override
59            .filter(|n| *n > 0)
60            .unwrap_or_else(|| vcpu.saturating_sub(actix_workers).max(1));
61        Self {
62            vcpu,
63            actix_workers,
64            tokio_worker_threads,
65        }
66    }
67
68    /// Total threads requested across both runtimes.
69    pub fn total_threads(&self) -> usize {
70        self.actix_workers + self.tokio_worker_threads
71    }
72
73    /// Whether the requested budget exceeds the detected vCPU quota.
74    pub fn over_budget(&self) -> bool {
75        self.total_threads() > self.vcpu
76    }
77
78    /// Log the resolved budget at startup and WARN when it exceeds the vCPU quota.
79    pub fn log_startup(&self) {
80        info!(
81            vcpu = self.vcpu,
82            actix_workers = self.actix_workers,
83            tokio_worker_threads = self.tokio_worker_threads,
84            "resolved runtime worker budget"
85        );
86        if self.over_budget() {
87            warn!(
88                vcpu = self.vcpu,
89                actix_workers = self.actix_workers,
90                tokio_worker_threads = self.tokio_worker_threads,
91                total = self.total_threads(),
92                "worker budget exceeds vCPU quota; threads will contend / CFS-throttle. \
93                 Pin TOKIO_WORKER_THREADS and ACTIX_WORKERS to fit the container CPU quota."
94            );
95        }
96    }
97}
98
99/// Join the pipeline worker handles so in-flight work drains before the runtime is
100/// torn down on shutdown (D7). Bounded by `timeout`; if it elapses, remaining work is
101/// abandoned (at-least-once redelivery is the backstop). Returns `true` if the drain
102/// completed within the window, `false` on timeout.
103///
104/// `WorkerHandle::Apalis` handles (if any) are dropped — the apalis Monitor shuts down
105/// via its own signal handling.
106pub async fn drain_worker_handles(handles: Vec<WorkerHandle>, timeout: Duration) -> bool {
107    let drain = async {
108        for handle in handles {
109            if let WorkerHandle::Tokio(join_handle) = handle {
110                if let Err(e) = join_handle.await {
111                    warn!(error = %e, "pipeline worker did not join cleanly");
112                }
113            }
114        }
115    };
116    tokio::time::timeout(timeout, drain).await.is_ok()
117}
118
119fn detect_vcpu() -> usize {
120    available_parallelism().map(|n| n.get()).unwrap_or(1)
121}
122
123fn read_usize_env(key: &str) -> Option<usize> {
124    env::var(key)
125        .ok()
126        .and_then(|v| v.trim().parse::<usize>().ok())
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[test]
134    fn defaults_split_vcpu_between_runtimes() {
135        let cfg = RuntimeConfig::resolve(8, None, None);
136        assert_eq!(cfg.vcpu, 8);
137        assert_eq!(cfg.actix_workers, 4); // max(1, 8/2)
138        assert_eq!(cfg.tokio_worker_threads, 4); // max(1, 8-4)
139        assert!(!cfg.over_budget());
140    }
141
142    #[test]
143    fn single_vcpu_never_yields_zero_threads() {
144        let cfg = RuntimeConfig::resolve(1, None, None);
145        assert_eq!(cfg.actix_workers, 1); // max(1, 0)
146        assert_eq!(cfg.tokio_worker_threads, 1); // max(1, 1-1)
147                                                 // 1 + 1 > 1: a single-vCPU box is inherently over budget (expected; warns).
148        assert!(cfg.over_budget());
149    }
150
151    #[test]
152    fn explicit_overrides_are_respected() {
153        let cfg = RuntimeConfig::resolve(8, Some(2), Some(4));
154        assert_eq!(cfg.actix_workers, 2);
155        assert_eq!(cfg.tokio_worker_threads, 4);
156        assert_eq!(cfg.total_threads(), 6);
157        assert!(!cfg.over_budget());
158    }
159
160    #[test]
161    fn tokio_default_accounts_for_actix_override() {
162        // Only actix pinned: tokio fills the remainder of the quota.
163        let cfg = RuntimeConfig::resolve(8, Some(6), None);
164        assert_eq!(cfg.actix_workers, 6);
165        assert_eq!(cfg.tokio_worker_threads, 2); // max(1, 8-6)
166    }
167
168    #[test]
169    fn zero_override_is_ignored() {
170        let cfg = RuntimeConfig::resolve(4, Some(0), Some(0));
171        assert_eq!(cfg.actix_workers, 2); // falls back to default
172        assert_eq!(cfg.tokio_worker_threads, 2);
173    }
174
175    #[test]
176    fn over_budget_detected_when_sum_exceeds_quota() {
177        let cfg = RuntimeConfig::resolve(4, Some(4), Some(4));
178        assert_eq!(cfg.total_threads(), 8);
179        assert!(cfg.over_budget());
180    }
181
182    #[tokio::test]
183    async fn drain_completes_when_workers_finish_within_window() {
184        // Two in-flight workers that each finish after a brief delay (well within the window).
185        let handles: Vec<WorkerHandle> = (0..2)
186            .map(|_| {
187                WorkerHandle::Tokio(tokio::spawn(async {
188                    tokio::time::sleep(Duration::from_millis(20)).await;
189                }))
190            })
191            .collect();
192
193        let drained = drain_worker_handles(handles, Duration::from_secs(5)).await;
194        assert!(drained, "workers should drain within the window");
195    }
196
197    #[tokio::test]
198    async fn drain_times_out_when_a_worker_hangs() {
199        // A worker that never completes — drain must give up at the (short) deadline.
200        let handles = vec![WorkerHandle::Tokio(tokio::spawn(async {
201            futures::future::pending::<()>().await;
202        }))];
203
204        let drained = drain_worker_handles(handles, Duration::from_millis(150)).await;
205        assert!(!drained, "drain must time out when a worker hangs");
206    }
207}
openzeppelin_relayer/bootstrap/runtime.rs

openzeppelin_relayer/bootstrap/
runtime.rs
