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, runs at most `max_concurrent_tasks` handlers, and keeps `idle_streams_per_queue`
18/// waiting streams on each queue. An omitted task limit uses the CPU count; an omitted idle
19/// stream count uses one. An omitted drain timeout uses 30 seconds.
20///
21/// ```no_run
22/// use std::sync::Arc;
23/// use dex_sdk::{BlobCache, BlobCacheConfig, Client, ClientOptions, PullWorkerOptions,
24/// Registry, Worker, WorkerOptions};
25///
26/// let cache = Arc::new(BlobCache::open(BlobCacheConfig::new(
27/// "/tmp/dex-pull-blobs", 64 * 1024 * 1024, 0,
28/// )?)?);
29/// let registry = Registry::new();
30/// let worker = Worker::try_new(registry.clone(), cache.clone(),
31/// WorkerOptions::new().pull(PullWorkerOptions::new().max_concurrent_tasks(8)))?;
32/// let client = Client::try_new(registry, cache,
33/// ClientOptions::new())?;
34/// // Run worker.start() on a dedicated thread; call worker.stop() during shutdown.
35/// # Ok::<(), Box<dyn std::error::Error>>(())
36/// ```
37#[derive(Clone, Debug, Default)]
38pub struct PullWorkerOptions {
39 pub(crate) max_concurrent_tasks: Option<usize>,
40 pub(crate) reserved_rpc_tasks: Option<usize>,
41 pub(crate) idle_streams_per_queue: Option<usize>,
42 pub(crate) reserve_pressure_linger: Option<Duration>,
43 pub(crate) app_worker_id: Option<String>,
44 pub(crate) stop_drain_timeout: Option<Duration>,
45 pub(crate) stop_on_permanent_stream_rejection: bool,
46}
47
48impl PullWorkerOptions {
49 /// Creates pull options with CPU-count concurrency, one idle stream per queue, and a 30-second drain.
50 ///
51 /// The app Worker ID is generated from process identity and eight random base32 characters.
52 pub fn new() -> Self {
53 Self::default()
54 }
55
56 /// Sets the maximum number of concurrent Step and RPC handlers.
57 ///
58 /// The default is the available CPU count. Zero returns [`crate::SdkError::FlowDefinition`]
59 /// from [`crate::Worker::try_new`]. Additional assignments are rejected at capacity.
60 pub fn max_concurrent_tasks(mut self, value: usize) -> Self {
61 self.max_concurrent_tasks = Some(value);
62 self
63 }
64
65 /// Reserves capacity to protect synchronous RPC latency from long-running Steps.
66 ///
67 /// Unset reserves one slot when resolved capacity is at least two, otherwise zero.
68 /// With the CPU-count default, Steps get one fewer slot than before. Set zero to opt out.
69 /// With one slot and no explicit reserve, start warns once that RPCs may wait behind Steps.
70 /// Flows must stay responsive because RPCs share Worker task capacity with Steps.
71 /// WaitFor and Execute tasks, including local attempts, are declined when
72 /// total busy handlers reach the resolved task limit minus this reserve. RPCs use the full limit.
73 /// API-direct and transactional RPCs both use the full limit. Declines reassign without consuming an attempt.
74 /// If all Workers remain at the Step limit through the dispatch wait, NO_WORKER_AVAILABLE
75 /// consumes an attempt and follows the Step retry policy. Local attempts wait the local dispatch
76 /// wait, then fall back according to their policy. Readiness is unaffected.
77 /// Declines emit a tracing debug event and increment [`crate::Worker::pull_reserved_rpc_declines`].
78 /// A queue keeps one extra idle stream for 30 seconds after its last reserve decline.
79 /// The Worker does not open it while all task slots are busy.
80 ///
81 /// `value` must be less than the task limit after its CPU-count default is resolved.
82 /// [`crate::Worker::try_new`] returns [`crate::SdkError::FlowDefinition`] otherwise.
83 /// Returns these options with the reserve applied.
84 ///
85 /// ```
86 /// use dex_sdk::PullWorkerOptions;
87 /// let pull = PullWorkerOptions::new().max_concurrent_tasks(8).reserved_rpc_tasks(2);
88 /// ```
89 pub fn reserved_rpc_tasks(mut self, value: usize) -> Self {
90 self.reserved_rpc_tasks = Some(value);
91 self
92 }
93
94 /// Sets the number of waiting streams per queue.
95 ///
96 /// The default is one. Zero returns [`crate::SdkError::FlowDefinition`] from
97 /// [`crate::Worker::try_new`]. The Worker also retains busy streams until their tasks finish.
98 /// A queue keeps one additional idle stream after a recent reserve decline. Use at least two for heavy Step backlogs.
99 pub fn idle_streams_per_queue(mut self, value: usize) -> Self {
100 self.idle_streams_per_queue = Some(value);
101 self
102 }
103
104 /// Overrides the process Worker ID used for sticky routing.
105 ///
106 /// Use a new ID after each process restart. It must contain 1–63 ASCII letters, digits,
107 /// dots, underscores, or hyphens and must differ from every registered Flow type.
108 /// Invalid values return [`crate::SdkError::FlowDefinition`] from [`crate::Worker::try_new`].
109 /// If omitted, the Worker generates a fresh ID during construction.
110 pub fn app_worker_id(mut self, value: impl Into<String>) -> Self {
111 self.app_worker_id = Some(value.into());
112 self
113 }
114
115 /// Stops when a permanent rejection leaves a queue without any registered stream.
116 ///
117 /// Defaults to false, retrying indefinitely. When enabled, [`crate::Worker::start`] drains
118 /// within the configured timeout, cancels remaining handlers, and waits for their return.
119 /// It returns a service error exposing [`crate::SdkError::pull_registration_error`] before or after initial readiness.
120 /// Calling [`crate::Worker::stop`] afterward remains safe.
121 pub fn stop_on_permanent_stream_rejection(mut self, value: bool) -> Self {
122 self.stop_on_permanent_stream_rejection = value;
123 self
124 }
125
126 /// Bounds graceful drain before active handlers are cancelled.
127 ///
128 /// The default is 30 seconds. Zero returns [`crate::SdkError::FlowDefinition`] from
129 /// [`crate::Worker::try_new`].
130 /// A busy server-directed Reconnect uses this duration to wait for server EOF after the final frame.
131 /// During Stop, half-closed streams use the remaining drain deadline.
132 /// [`crate::Worker::start`] waits for cancelled handlers to return before it exits.
133 pub fn stop_drain_timeout(mut self, value: Duration) -> Self {
134 self.stop_drain_timeout = Some(value);
135 self
136 }
137}
138
139#[derive(Clone, Debug)]
140/// Configures a pull [`crate::Worker`] and its outbound Dex connection.
141///
142/// Defaults connect to Dex at `127.0.0.1:8801`, use [`PullWorkerOptions::default`], and allow
143/// two minutes for Attribute-index synchronization. Workers need only an outbound connection.
144pub struct WorkerOptions {
145 server_address: String,
146 pull: PullWorkerOptions,
147 attribute_index_sync_timeout: Duration,
148 credential: Option<Arc<dyn Credential>>,
149}
150
151impl WorkerOptions {
152 /// Creates Worker options with local development defaults.
153 pub fn new() -> Self {
154 Self {
155 server_address: "127.0.0.1:8801".to_string(),
156 pull: PullWorkerOptions::default(),
157 attribute_index_sync_timeout: Duration::from_secs(120),
158 credential: None,
159 }
160 }
161
162 /// Sets the plaintext Dex FlowService `host:port` used by the Worker.
163 pub fn server_address(mut self, value: impl Into<String>) -> Self {
164 self.server_address = value.into();
165 self
166 }
167
168 /// Sends a bearer token on FlowService calls and pull Worker task streams.
169 ///
170 /// The default sends no authorization metadata. The credential may be called concurrently;
171 /// errors fail calls locally. Pull streams evaluate it at opening and again on reconnect.
172 /// The token uses Dex's plaintext connection, so protect the network path.
173 pub fn credential(mut self, value: Arc<dyn Credential>) -> Self {
174 self.credential = Some(value);
175 self
176 }
177
178 /// Configures pull capacity, queue registration, and graceful shutdown.
179 ///
180 /// Every Worker uses pull; this builder overrides [`PullWorkerOptions::default`].
181 /// Takes ownership of `value` and returns these Worker options with its settings applied.
182 /// [`crate::Worker::try_new`] validates capacity, identity, and timeout settings.
183 /// See [`PullWorkerOptions`] for defaults and a complete Worker and Client example.
184 pub fn pull(mut self, value: PullWorkerOptions) -> Self {
185 self.pull = value;
186 self
187 }
188
189 /// Sets the startup Indexed Attribute synchronization deadline.
190 ///
191 /// The default is two minutes. [`Worker::try_new`](crate::Worker::try_new)
192 /// rejects [`Duration::ZERO`].
193 pub fn attribute_index_sync_timeout(mut self, value: Duration) -> Self {
194 self.attribute_index_sync_timeout = value;
195 self
196 }
197
198 pub(crate) fn pull_value(&self) -> &PullWorkerOptions {
199 &self.pull
200 }
201
202 pub(crate) fn server_address_value(&self) -> &str {
203 &self.server_address
204 }
205
206 pub(crate) fn attribute_index_sync_timeout_value(&self) -> Duration {
207 self.attribute_index_sync_timeout
208 }
209
210 pub(crate) fn credential_value(&self) -> Option<Arc<dyn Credential>> {
211 self.credential.clone()
212 }
213}
214
215impl Default for WorkerOptions {
216 fn default() -> Self {
217 Self::new()
218 }
219}
220
221pub(crate) struct WorkerEnvironmentConfig {
222 pub(crate) identity: String,
223}
224
225impl WorkerEnvironmentConfig {
226 pub(crate) fn from_environment() -> Self {
227 let identity = ["DEX_WORKER_INSTANCE", "POD_NAME", "HOSTNAME"]
228 .into_iter()
229 .filter_map(std::env::var_os)
230 .map(|value| value.to_string_lossy().into_owned())
231 .find(|value| !value.trim().is_empty())
232 .unwrap_or_else(|| gethostname::gethostname().to_string_lossy().into_owned());
233 Self { identity }
234 }
235}