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