aion_worker/
config.rs

1//! `WorkerConfig` endpoint, task queue, identity, concurrency, and TLS/credentials passthrough.
2
3use std::fmt;
4use std::time::Duration;
5
6/// Opaque credentials forwarded to the worker transport layer.
7///
8/// The worker SDK does not interpret this value or define an authentication
9/// scheme. It exists only so operators can pass transport-specific credentials
10/// to the session implementation that knows how to apply them.
11#[derive(Clone, PartialEq, Eq)]
12pub struct TransportCredentials {
13    secret: Vec<u8>,
14}
15
16impl TransportCredentials {
17    /// Creates opaque transport credentials from caller-supplied bytes.
18    #[must_use]
19    pub fn new(secret: impl Into<Vec<u8>>) -> Self {
20        Self {
21            secret: secret.into(),
22        }
23    }
24
25    /// Returns the opaque credential bytes for transport-specific forwarding.
26    #[must_use]
27    pub fn secret(&self) -> &[u8] {
28        &self.secret
29    }
30}
31
32impl fmt::Debug for TransportCredentials {
33    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
34        formatter
35            .debug_struct("TransportCredentials")
36            .field("secret", &"<redacted>")
37            .finish()
38    }
39}
40
41/// Operator-supplied reconnect backoff settings.
42///
43/// The settings govern both session establishment and the run loop's
44/// cumulative mid-run session-drop budget.
45///
46/// **Budget reset:** the cumulative drop budget resets to zero once an
47/// established session proves healthy — it served at least one task, or it
48/// stayed connected longer than `max_backoff` (measured monotonically from
49/// successful registration to the moment the stream ended or dropped;
50/// post-drop draining of in-flight activities never extends it). The cap is
51/// the policy's own definition of the longest pause, so a session outliving
52/// it is demonstrably past the flapping regime, and a served task proves
53/// end-to-end health. A genuinely flapping server — no session ever serves
54/// a task or outlives `max_backoff` — exhausts the budget after exactly
55/// `max_attempts` drops.
56///
57/// **Drains and clean closes:** a server-announced drain (the wire
58/// `DrainRequest` frame) is an unbudgeted drop — the worker finishes
59/// in-flight work and redials after `initial_backoff`; the drain
60/// classification latches for the session, so even an abrupt end after the
61/// frame stays drain-class. An *unannounced* clean stream close remains a
62/// budgeted retryable drop: the worker redials through the same budgeted,
63/// backed-off cycle, and only a persistent unannounced clean-close loop
64/// exhausts the budget (surfacing
65/// [`crate::error::WorkerError::CleanCloseExhausted`]).
66///
67/// **Shutdown during a drop backoff:** every SDK races the backoff sleep
68/// against the shutdown signal and returns promptly, and the run outcome is
69/// aligned across the Rust, Python, and TypeScript workers: a pending
70/// drain-class or clean-close drop ends the run cleanly, while a pending
71/// error-class drop surfaces its error — a supervisor sees "this worker was
72/// mid-fault" distinctly from "this worker drained cleanly".
73#[derive(Clone, Debug, PartialEq, Eq)]
74pub struct ReconnectConfig {
75    /// Initial reconnect backoff delay. Must be non-zero before reconnecting.
76    pub initial_backoff: Duration,
77    /// Maximum reconnect backoff delay cap. Must be non-zero before
78    /// reconnecting. Doubles as the session-health threshold for the
79    /// drop-budget reset described on this type.
80    pub max_backoff: Duration,
81    /// Maximum reconnect attempts before surfacing the last connection error.
82    pub max_attempts: usize,
83}
84
85impl ReconnectConfig {
86    /// Creates reconnect settings with every field supplied explicitly.
87    #[must_use]
88    pub const fn new(
89        initial_backoff: Duration,
90        max_backoff: Duration,
91        max_attempts: usize,
92    ) -> Self {
93        Self {
94            initial_backoff,
95            max_backoff,
96            max_attempts,
97        }
98    }
99}
100
101/// Operator-supplied worker connection and serving configuration.
102///
103/// Tunable fields remain caller-supplied. Namespace authorization metadata
104/// defaults to `default`/`worker` so development workers can register against
105/// the default namespace without an explicit auth setup.
106///
107/// `namespaces`, `task_queue`, and `node` are disjoint routing dimensions.
108/// `namespaces` is the SET of correctness/isolation boundaries the worker is
109/// authorized for — the same set carried (comma-joined) in the
110/// `x-aion-namespaces` auth metadata and in the registration scope, so a worker
111/// registers into exactly the namespaces it is authorized for. `task_queue` is
112/// the pool/flavour selector *within* each namespace. `node` is an OPTIONAL
113/// locality affinity the worker advertises (default = machine hostname).
114#[derive(Clone, Debug, PartialEq, Eq)]
115pub struct WorkerConfig {
116    /// Set of correctness/isolation boundaries this worker registers into.
117    /// Advertised both in `x-aion-namespaces` worker stream metadata (comma
118    /// joined) and as the registration namespace set, so authorization and
119    /// registration scope agree. Must be non-empty.
120    pub namespaces: Vec<String>,
121    /// Subject advertised in `x-aion-subject` worker stream metadata.
122    pub subject: String,
123    /// Engine worker endpoint URI.
124    pub endpoint: String,
125    /// Pool/flavour selector within each namespace, sent as the registration's
126    /// `task_queue`. The worker-pool address is `(namespace, task_queue)`.
127    pub task_queue: String,
128    /// Locality affinity advertised at registration. Defaults to the machine
129    /// hostname; a dispatch pinned to this node reaches this worker.
130    pub node: String,
131    /// Worker identity used by operators and future wire metadata.
132    pub identity: String,
133    /// Maximum concurrent activities this worker may serve.
134    pub max_concurrency: usize,
135    /// Operator-supplied reconnect settings.
136    pub reconnect: ReconnectConfig,
137    /// Opaque credentials for the transport implementation.
138    pub transport_credentials: Option<TransportCredentials>,
139}
140
141const DEFAULT_WORKER_NAMESPACE: &str = "default";
142const DEFAULT_WORKER_SUBJECT: &str = "worker";
143
144/// Fallback node id when the machine hostname cannot be resolved. A worker must
145/// never fail to start over a missing hostname, so it advertises this documented
146/// default instead of panicking.
147const DEFAULT_WORKER_NODE: &str = "localhost";
148
149/// Resolve the machine hostname for use as the default node locality affinity.
150///
151/// Falls back to [`DEFAULT_WORKER_NODE`] (never panics) when the OS hostname is
152/// unavailable or is not valid UTF-8.
153#[must_use]
154pub fn default_node() -> String {
155    std::env::var("HOSTNAME")
156        .ok()
157        .filter(|hostname| !hostname.is_empty())
158        .or_else(|| {
159            hostname::get()
160                .ok()
161                .and_then(|raw| raw.into_string().ok())
162                .filter(|hostname| !hostname.is_empty())
163        })
164        .unwrap_or_else(|| String::from(DEFAULT_WORKER_NODE))
165}
166
167impl WorkerConfig {
168    /// Starts an explicit builder. The caller must provide every required field
169    /// before calling [`WorkerConfigBuilder::build`].
170    #[must_use]
171    pub const fn builder() -> WorkerConfigBuilder {
172        WorkerConfigBuilder::new()
173    }
174
175    /// Creates a worker config with default authorization metadata.
176    #[must_use]
177    pub fn new(
178        endpoint: impl Into<String>,
179        task_queue: impl Into<String>,
180        identity: impl Into<String>,
181        max_concurrency: usize,
182        reconnect: ReconnectConfig,
183        transport_credentials: Option<TransportCredentials>,
184    ) -> Self {
185        Self {
186            namespaces: vec![String::from(DEFAULT_WORKER_NAMESPACE)],
187            subject: String::from(DEFAULT_WORKER_SUBJECT),
188            endpoint: endpoint.into(),
189            task_queue: task_queue.into(),
190            node: default_node(),
191            identity: identity.into(),
192            max_concurrency,
193            reconnect,
194            transport_credentials,
195        }
196    }
197}
198
199/// Builder for [`WorkerConfig`] with auth metadata defaults and explicit required fields.
200#[derive(Clone, Debug, Default)]
201pub struct WorkerConfigBuilder {
202    namespaces: Option<Vec<String>>,
203    subject: Option<String>,
204    endpoint: Option<String>,
205    task_queue: Option<String>,
206    node: Option<String>,
207    identity: Option<String>,
208    max_concurrency: Option<usize>,
209    reconnect_initial_backoff: Option<Duration>,
210    reconnect_max_backoff: Option<Duration>,
211    reconnect_max_attempts: Option<usize>,
212    transport_credentials: Option<TransportCredentials>,
213}
214
215impl WorkerConfigBuilder {
216    /// Creates an empty config builder.
217    #[must_use]
218    pub const fn new() -> Self {
219        Self {
220            namespaces: None,
221            subject: None,
222            endpoint: None,
223            task_queue: None,
224            node: None,
225            identity: None,
226            max_concurrency: None,
227            reconnect_initial_backoff: None,
228            reconnect_max_backoff: None,
229            reconnect_max_attempts: None,
230            transport_credentials: None,
231        }
232    }
233
234    /// Sets the SET of namespaces advertised in worker stream authorization
235    /// metadata and the registration scope. Replaces any previously set value.
236    #[must_use]
237    pub fn namespaces(mut self, namespaces: impl IntoIterator<Item = String>) -> Self {
238        self.namespaces = Some(namespaces.into_iter().collect());
239        self
240    }
241
242    /// Sets a single namespace, replacing any previously set namespace set.
243    /// Convenience over [`Self::namespaces`] for the common one-namespace case.
244    #[must_use]
245    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
246        self.namespaces = Some(vec![namespace.into()]);
247        self
248    }
249
250    /// Sets the locality affinity (node) advertised at registration.
251    #[must_use]
252    pub fn node(mut self, node: impl Into<String>) -> Self {
253        self.node = Some(node.into());
254        self
255    }
256
257    /// Sets the subject advertised in worker stream authorization metadata.
258    #[must_use]
259    pub fn subject(mut self, subject: impl Into<String>) -> Self {
260        self.subject = Some(subject.into());
261        self
262    }
263
264    /// Sets the engine worker endpoint URI.
265    #[must_use]
266    pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
267        self.endpoint = Some(endpoint.into());
268        self
269    }
270
271    /// Sets the task queue advertised to the engine.
272    #[must_use]
273    pub fn task_queue(mut self, task_queue: impl Into<String>) -> Self {
274        self.task_queue = Some(task_queue.into());
275        self
276    }
277
278    /// Sets the worker identity.
279    #[must_use]
280    pub fn identity(mut self, identity: impl Into<String>) -> Self {
281        self.identity = Some(identity.into());
282        self
283    }
284
285    /// Sets the operator-configured maximum concurrency.
286    #[must_use]
287    pub const fn max_concurrency(mut self, max_concurrency: usize) -> Self {
288        self.max_concurrency = Some(max_concurrency);
289        self
290    }
291
292    /// Sets the operator-configured initial reconnect backoff delay.
293    #[must_use]
294    pub const fn reconnect_initial_backoff(mut self, delay: Duration) -> Self {
295        self.reconnect_initial_backoff = Some(delay);
296        self
297    }
298
299    /// Sets the operator-configured reconnect backoff cap.
300    #[must_use]
301    pub const fn reconnect_max_backoff(mut self, delay: Duration) -> Self {
302        self.reconnect_max_backoff = Some(delay);
303        self
304    }
305
306    /// Sets the operator-configured maximum reconnect attempts.
307    #[must_use]
308    pub const fn reconnect_max_attempts(mut self, attempts: usize) -> Self {
309        self.reconnect_max_attempts = Some(attempts);
310        self
311    }
312
313    /// Sets optional opaque transport credentials.
314    #[must_use]
315    pub fn transport_credentials(mut self, credentials: TransportCredentials) -> Self {
316        self.transport_credentials = Some(credentials);
317        self
318    }
319
320    /// Builds a [`WorkerConfig`] when every required field has been supplied.
321    ///
322    /// # Errors
323    ///
324    /// Returns [`WorkerConfigBuildError`] naming the missing required field.
325    pub fn build(self) -> Result<WorkerConfig, WorkerConfigBuildError> {
326        let namespaces = self
327            .namespaces
328            .filter(|namespaces| !namespaces.is_empty())
329            .unwrap_or_else(|| vec![String::from(DEFAULT_WORKER_NAMESPACE)]);
330        Ok(WorkerConfig {
331            namespaces,
332            subject: self
333                .subject
334                .unwrap_or_else(|| String::from(DEFAULT_WORKER_SUBJECT)),
335            endpoint: self
336                .endpoint
337                .ok_or(WorkerConfigBuildError::MissingEndpoint)?,
338            task_queue: self
339                .task_queue
340                .ok_or(WorkerConfigBuildError::MissingTaskQueue)?,
341            node: self.node.unwrap_or_else(default_node),
342            identity: self
343                .identity
344                .ok_or(WorkerConfigBuildError::MissingIdentity)?,
345            max_concurrency: self
346                .max_concurrency
347                .ok_or(WorkerConfigBuildError::MissingMaxConcurrency)?,
348            reconnect: ReconnectConfig {
349                initial_backoff: self
350                    .reconnect_initial_backoff
351                    .ok_or(WorkerConfigBuildError::MissingReconnectInitialBackoff)?,
352                max_backoff: self
353                    .reconnect_max_backoff
354                    .ok_or(WorkerConfigBuildError::MissingReconnectMaxBackoff)?,
355                max_attempts: self
356                    .reconnect_max_attempts
357                    .ok_or(WorkerConfigBuildError::MissingReconnectMaxAttempts)?,
358            },
359            transport_credentials: self.transport_credentials,
360        })
361    }
362}
363
364/// Errors produced while building [`WorkerConfig`].
365#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
366pub enum WorkerConfigBuildError {
367    /// The endpoint was not supplied.
368    #[error("worker endpoint is required")]
369    MissingEndpoint,
370    /// The task queue was not supplied.
371    #[error("worker task queue is required")]
372    MissingTaskQueue,
373    /// The worker identity was not supplied.
374    #[error("worker identity is required")]
375    MissingIdentity,
376    /// The max concurrency was not supplied.
377    #[error("worker max_concurrency is required")]
378    MissingMaxConcurrency,
379    /// The reconnect initial backoff was not supplied.
380    #[error("worker reconnect_initial_backoff is required")]
381    MissingReconnectInitialBackoff,
382    /// The reconnect max backoff was not supplied.
383    #[error("worker reconnect_max_backoff is required")]
384    MissingReconnectMaxBackoff,
385    /// The reconnect max attempts value was not supplied.
386    #[error("worker reconnect_max_attempts is required")]
387    MissingReconnectMaxAttempts,
388}
389
390#[cfg(test)]
391mod tests {
392    use std::time::Duration;
393
394    use super::{TransportCredentials, WorkerConfig};
395
396    #[test]
397    fn worker_config_builder_round_trips_fields() -> Result<(), Box<dyn std::error::Error>> {
398        let credentials = TransportCredentials::new(b"secret-token".to_vec());
399        let config = WorkerConfig::builder()
400            .endpoint("http://127.0.0.1:50051")
401            .task_queue("payments")
402            .identity("worker-a")
403            .max_concurrency(7)
404            .reconnect_initial_backoff(Duration::from_millis(10))
405            .reconnect_max_backoff(Duration::from_millis(100))
406            .reconnect_max_attempts(3)
407            .namespace("payments")
408            .subject("worker-a")
409            .transport_credentials(credentials.clone())
410            .build()?;
411
412        assert_eq!(config.namespaces, vec![String::from("payments")]);
413        assert_eq!(config.subject, "worker-a");
414        assert_eq!(config.endpoint, "http://127.0.0.1:50051");
415        assert_eq!(config.task_queue, "payments");
416        assert!(!config.node.is_empty(), "node defaults to the hostname");
417        assert_eq!(config.identity, "worker-a");
418        assert_eq!(config.max_concurrency, 7);
419        assert_eq!(config.reconnect.initial_backoff, Duration::from_millis(10));
420        assert_eq!(config.reconnect.max_backoff, Duration::from_millis(100));
421        assert_eq!(config.reconnect.max_attempts, 3);
422        assert_eq!(config.transport_credentials, Some(credentials));
423        assert!(!format!("{config:?}").contains("secret-token"));
424
425        Ok(())
426    }
427
428    #[test]
429    fn worker_config_new_uses_auth_metadata_defaults() {
430        let config = WorkerConfig::new(
431            "http://127.0.0.1:50051",
432            "default",
433            "worker-a",
434            4,
435            super::ReconnectConfig::new(Duration::from_millis(10), Duration::from_millis(100), 3),
436            None,
437        );
438
439        assert_eq!(config.namespaces, vec![String::from("default")]);
440        assert_eq!(config.subject, "worker");
441        assert!(!config.node.is_empty(), "node defaults to the hostname");
442    }
443
444    #[test]
445    fn worker_config_builder_carries_namespace_set_and_node()
446    -> Result<(), Box<dyn std::error::Error>> {
447        let config = WorkerConfig::builder()
448            .endpoint("http://127.0.0.1:50051")
449            .task_queue("default")
450            .identity("worker-a")
451            .max_concurrency(1)
452            .reconnect_initial_backoff(Duration::from_millis(5))
453            .reconnect_max_backoff(Duration::from_millis(20))
454            .reconnect_max_attempts(3)
455            .namespaces([String::from("a"), String::from("b")])
456            .node("host-7")
457            .build()?;
458
459        assert_eq!(
460            config.namespaces,
461            vec![String::from("a"), String::from("b")]
462        );
463        assert_eq!(config.node, "host-7");
464        Ok(())
465    }
466}
aion_worker/config.rs

aion_worker/
config.rs
