dex_sdk/worker_options.rs
1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
8
9use std::sync::Arc;
10use std::time::Duration;
11
12use crate::Credential;
13
14/// Configures outbound polling for a Worker.
15///
16/// Pass this to [`WorkerOptions::pull`]. The Worker polls its sticky queue and each registered
17/// Flow type through exactly `max_concurrent_tasks` long-lived slot streams. An omitted task limit
18/// uses the CPU count. An omitted drain timeout uses 30 seconds.
19///
20/// ```no_run
21/// use std::sync::Arc;
22/// use dex_sdk::{BlobCache, BlobCacheConfig, Client, ClientOptions, PullWorkerOptions,
23/// Registry, Worker, WorkerOptions};
24///
25/// let cache = Arc::new(BlobCache::open(BlobCacheConfig::new(
26/// "/tmp/dex-pull-blobs", 64 * 1024 * 1024, 0,
27/// )?)?);
28/// let registry = Registry::new();
29/// let worker = Worker::try_new(registry.clone(), cache.clone(),
30/// WorkerOptions::new().pull(PullWorkerOptions::new().max_concurrent_tasks(8)))?;
31/// let client = Client::try_new(registry, cache,
32/// ClientOptions::new())?;
33/// // Run worker.start() on a dedicated thread; call worker.stop() during shutdown.
34/// # Ok::<(), Box<dyn std::error::Error>>(())
35/// ```
36#[derive(Clone, Debug, Default)]
37pub struct PullWorkerOptions {
38 pub(crate) max_concurrent_tasks: Option<usize>,
39 pub(crate) reserved_rpc_tasks: Option<usize>,
40 pub(crate) idle_streams_per_queue: Option<usize>,
41 pub(crate) app_worker_id: Option<String>,
42 pub(crate) stop_drain_timeout: Option<Duration>,
43 pub(crate) stop_on_permanent_stream_rejection: bool,
44}
45
46impl PullWorkerOptions {
47 /// Creates pull options with CPU-count slot concurrency and a 30-second drain.
48 ///
49 /// The app Worker ID is generated from process identity and eight random base32 characters.
50 pub fn new() -> Self {
51 Self::default()
52 }
53
54 /// Sets the maximum number of concurrent Step and RPC handlers.
55 ///
56 /// The default is the available CPU count. Zero returns [`crate::SdkError::FlowDefinition`]
57 /// from [`crate::Worker::try_new`]. Each slot owns one long-lived poll stream and executes
58 /// every task assigned to it. [`crate::Worker::start`] fails if this value exceeds the
59 /// Server's reported per-Worker limit, configured by `matching.maxConcurrentTasksCap`.
60 pub fn max_concurrent_tasks(mut self, value: usize) -> Self {
61 self.max_concurrent_tasks = Some(value);
62 self
63 }
64
65 /// Reserves slots that accept only RPCs, protecting RPC latency from long-running Steps.
66 ///
67 /// The default reserves one slot when resolved concurrency is at least two, otherwise zero.
68 /// The remaining slots accept Steps and RPCs. The Server selects eligible slots; the Worker
69 /// never declines a Step for capacity. Set zero to allow Steps on every slot. A one-slot
70 /// Worker with no explicit reservation warns that RPCs may wait behind Steps.
71 /// Readiness requires a registered Step-and-RPC slot and, with a reservation, an RPC-only slot.
72 /// `value` must be less than resolved `max_concurrent_tasks`; otherwise
73 /// [`crate::Worker::try_new`] returns [`crate::SdkError::FlowDefinition`].
74 /// Returns these options with the reservation applied.
75 ///
76 /// ```
77 /// use dex_sdk::PullWorkerOptions;
78 /// let pull = PullWorkerOptions::new().max_concurrent_tasks(8).reserved_rpc_tasks(2);
79 /// ```
80 pub fn reserved_rpc_tasks(mut self, value: usize) -> Self {
81 self.reserved_rpc_tasks = Some(value);
82 self
83 }
84
85 /// Retains the former per-queue idle-stream setting for source compatibility.
86 ///
87 /// Deprecated: `value` is accepted, including zero, and has no effect. Every execution slot
88 /// now owns exactly one long-lived stream. Use [`Self::max_concurrent_tasks`] to select the
89 /// total stream count. Returns these options with the compatibility value retained.
90 #[deprecated(note = "one long-lived stream per execution slot; this setting has no effect")]
91 pub fn idle_streams_per_queue(mut self, value: usize) -> Self {
92 self.idle_streams_per_queue = Some(value);
93 self
94 }
95
96 /// Overrides the process Worker ID used for sticky routing.
97 ///
98 /// Use a new ID after each process restart. It must contain 1–63 ASCII letters, digits,
99 /// dots, underscores, or hyphens and must differ from every registered Flow type.
100 /// Invalid values return [`crate::SdkError::FlowDefinition`] from [`crate::Worker::try_new`].
101 /// If omitted, the Worker generates a fresh ID during construction.
102 pub fn app_worker_id(mut self, value: impl Into<String>) -> Self {
103 self.app_worker_id = Some(value.into());
104 self
105 }
106
107 /// Stops when a permanent rejection leaves a required slot kind without any registered stream.
108 ///
109 /// Defaults to false, retrying indefinitely. When enabled, [`crate::Worker::start`] drains
110 /// within the configured timeout, cancels remaining handlers, and waits for their return.
111 /// It returns a service error exposing [`crate::SdkError::pull_registration_error`] before or after initial readiness.
112 /// Calling [`crate::Worker::stop`] afterward remains safe.
113 pub fn stop_on_permanent_stream_rejection(mut self, value: bool) -> Self {
114 self.stop_on_permanent_stream_rejection = value;
115 self
116 }
117
118 /// Bounds graceful drain before active handlers are cancelled.
119 ///
120 /// The default is 30 seconds. Zero returns [`crate::SdkError::FlowDefinition`] from
121 /// [`crate::Worker::try_new`].
122 /// A busy server-directed Reconnect uses this duration to wait for server EOF after the final frame.
123 /// During Stop, idle streams half-close immediately. Busy streams finish and send their final
124 /// output before half-closing. Later task frames are dropped without invoking handlers, and
125 /// half-closed streams keep reading until the Server ends them or the drain deadline expires.
126 /// At the deadline, a stream with a running handler is reset instead of half-closed.
127 /// [`crate::Worker::start`] waits for cancelled handlers to return before it exits.
128 /// Every other stream still waiting for the Server to end it is reset at the deadline.
129 /// A stream whose last final output Dex has not yet acknowledged with a new task waits up to
130 /// five seconds past this deadline for the Server to end it, then is reset.
131 /// Stop can therefore return up to five seconds after this deadline.
132 pub fn stop_drain_timeout(mut self, value: Duration) -> Self {
133 self.stop_drain_timeout = Some(value);
134 self
135 }
136}
137
138#[derive(Clone, Debug)]
139/// Configures a pull [`crate::Worker`] and its outbound Dex connection.
140///
141/// Defaults connect to Dex at `127.0.0.1:8801`, use [`PullWorkerOptions::default`], and allow
142/// two minutes for Attribute-index synchronization. Workers need only an outbound connection.
143pub struct WorkerOptions {
144 server_address: String,
145 pull: PullWorkerOptions,
146 attribute_index_sync_timeout: Duration,
147 credential: Option<Arc<dyn Credential>>,
148}
149
150impl WorkerOptions {
151 /// Creates Worker options with local development defaults.
152 pub fn new() -> Self {
153 Self {
154 server_address: "127.0.0.1:8801".to_string(),
155 pull: PullWorkerOptions::default(),
156 attribute_index_sync_timeout: Duration::from_secs(120),
157 credential: None,
158 }
159 }
160
161 /// Sets the plaintext Dex FlowService `host:port` used by the Worker.
162 pub fn server_address(mut self, value: impl Into<String>) -> Self {
163 self.server_address = value.into();
164 self
165 }
166
167 /// Sends a bearer token on FlowService calls and pull Worker task streams.
168 ///
169 /// The default sends no authorization metadata. The credential may be called concurrently;
170 /// errors fail calls locally. Pull streams evaluate it at opening and again on reconnect.
171 /// The token uses Dex's plaintext connection, so protect the network path.
172 pub fn credential(mut self, value: Arc<dyn Credential>) -> Self {
173 self.credential = Some(value);
174 self
175 }
176
177 /// Configures pull capacity, queue registration, and graceful shutdown.
178 ///
179 /// Every Worker uses pull; this builder overrides [`PullWorkerOptions::default`].
180 /// Takes ownership of `value` and returns these Worker options with its settings applied.
181 /// [`crate::Worker::try_new`] validates capacity, identity, and timeout settings.
182 /// See [`PullWorkerOptions`] for defaults and a complete Worker and Client example.
183 pub fn pull(mut self, value: PullWorkerOptions) -> Self {
184 self.pull = value;
185 self
186 }
187
188 /// Sets the startup Indexed Attribute synchronization deadline.
189 ///
190 /// The default is two minutes. [`Worker::try_new`](crate::Worker::try_new)
191 /// rejects [`Duration::ZERO`].
192 pub fn attribute_index_sync_timeout(mut self, value: Duration) -> Self {
193 self.attribute_index_sync_timeout = value;
194 self
195 }
196
197 pub(crate) fn pull_value(&self) -> &PullWorkerOptions {
198 &self.pull
199 }
200
201 pub(crate) fn server_address_value(&self) -> &str {
202 &self.server_address
203 }
204
205 pub(crate) fn attribute_index_sync_timeout_value(&self) -> Duration {
206 self.attribute_index_sync_timeout
207 }
208
209 pub(crate) fn credential_value(&self) -> Option<Arc<dyn Credential>> {
210 self.credential.clone()
211 }
212}
213
214impl Default for WorkerOptions {
215 fn default() -> Self {
216 Self::new()
217 }
218}
219
220pub(crate) struct WorkerEnvironmentConfig {
221 pub(crate) identity: String,
222}
223
224impl WorkerEnvironmentConfig {
225 pub(crate) fn from_environment() -> Self {
226 let identity = ["DEX_WORKER_INSTANCE", "POD_NAME", "HOSTNAME"]
227 .into_iter()
228 .filter_map(std::env::var_os)
229 .map(|value| value.to_string_lossy().into_owned())
230 .find(|value| !value.trim().is_empty())
231 .unwrap_or_else(|| gethostname::gethostname().to_string_lossy().into_owned());
232 Self { identity }
233 }
234}