boson_runtime/builder.rs
1//! [`BosonBuilder`] — construct a [`Boson`] runtime with injected ports.
2
3use std::sync::Arc;
4
5use boson_core::{ActorJsonPolicy, ExecutionContextFactory, QueueBackend};
6use boson_telemetry::{ConsoleOpsLog, OpsLog};
7
8use crate::registry::TaskRegistry;
9use crate::Boson;
10
11/// Builder choice for enqueue [`ActorJsonPolicy`].
12#[derive(Default)]
13pub(crate) enum ActorPolicyChoice {
14 /// Install [`RejectExternalSystemActor`](boson_core::RejectExternalSystemActor).
15 #[default]
16 DefaultRejectExternalSystem,
17 /// No policy checks.
18 Disabled,
19 /// Host-supplied policy.
20 Custom(Arc<dyn ActorJsonPolicy>),
21}
22
23/// Builder for [`Boson`].
24///
25/// **Required:** a queue backend ([`queue_backend`](Self::queue_backend) or
26/// [`queue_backend_from_global`](Self::queue_backend_from_global)) and an
27/// [`execution_context_factory`](Self::execution_context_factory). For `#[task]` handlers on a
28/// **worker** process, call [`auto_registry`](Self::auto_registry) so inventory entries from linked
29/// crates are collected.
30///
31/// **Topology** (see the [`boson`](https://docs.rs/uf-boson) crate Getting started):
32///
33/// | Topology | Builder posture |
34/// |----------|-----------------|
35/// | Embedded | [`auto_registry`](Self::auto_registry) + [`build`](Self::build) (or [`build_manual`](Self::build_manual) in tests) |
36/// | Remote worker — enqueue host | [`auto_registry`](Self::auto_registry) + [`without_worker`](Self::without_worker) + [`build`](Self::build) + [`configure`](crate::configure) |
37/// | Remote worker — worker | [`worker_id`](Self::worker_id) + [`lease_ttl_secs`](Self::lease_ttl_secs) (`> 0`) + [`auto_registry`](Self::auto_registry) + [`build`](Self::build) |
38///
39/// **Optional:** [`ops_log`](Self::ops_log) / [`ops_log_console`](Self::ops_log_console) for
40/// telemetry ([`OpsLog`](boson_telemetry::OpsLog)).
41///
42/// # Examples
43///
44/// ## Embedded (enqueue + worker in one process)
45///
46/// After [`build`](Self::build), call [`configure`](crate::configure) if callers use macro
47/// `send_with` (not required when holding `Boson` and calling [`Boson::enqueue`] directly).
48///
49/// ```rust,no_run
50/// use std::sync::Arc;
51///
52/// use boson_backend_mem::MemQueueBackend;
53/// use boson_core::JsonExecutionContextFactory;
54/// use boson_runtime::{configure, Boson};
55///
56/// # fn main() -> boson_core::Result<()> {
57/// let boson = Boson::builder()
58/// .queue_backend(Arc::new(MemQueueBackend::new()))
59/// .execution_context_factory(JsonExecutionContextFactory)
60/// .auto_registry()
61/// .build()?;
62/// configure(boson);
63/// # Ok(())
64/// # }
65/// ```
66///
67/// ## Remote worker — enqueue-only host
68///
69/// ```rust,no_run
70/// use std::sync::Arc;
71///
72/// use boson_backend_mem::MemQueueBackend;
73/// use boson_core::JsonExecutionContextFactory;
74/// use boson_runtime::{configure, Boson};
75///
76/// # fn main() -> boson_core::Result<()> {
77/// // Production remote worker: use Sqlite/Postgres/Redis/NATS — mem cannot cross processes.
78/// let boson = Boson::builder()
79/// .queue_backend(Arc::new(MemQueueBackend::new()))
80/// .execution_context_factory(JsonExecutionContextFactory)
81/// .auto_registry()
82/// .without_worker()
83/// .build()?;
84/// configure(boson);
85/// # Ok(())
86/// # }
87/// ```
88#[derive(Default)]
89pub struct BosonBuilder {
90 pub(crate) queue_backend: Option<Arc<dyn QueueBackend>>,
91 pub(crate) execution_context_factory: Option<Arc<dyn ExecutionContextFactory>>,
92 pub(crate) ops_log: Option<Arc<dyn OpsLog>>,
93 pub(crate) registry: Option<Arc<TaskRegistry>>,
94 pub(crate) use_auto_registry: bool,
95 pub(crate) spawn_worker: bool,
96 pub(crate) worker_id: Option<String>,
97 pub(crate) lease_ttl_secs: Option<i64>,
98 pub(crate) runtime_label: Option<String>,
99 pub(crate) worker_pools: Option<Vec<String>>,
100 pub(crate) worker_poll_interval_ms: Option<u64>,
101 pub(crate) idempotency_mode: boson_core::IdempotencyMode,
102 pub(crate) actor_policy: ActorPolicyChoice,
103}
104
105impl BosonBuilder {
106 /// Worker identity for lease claims (default: `INSTANCE_ID` / `BOSON_WORKER_ID` / `boson-worker-1`).
107 ///
108 /// Required to be **unique per process** for remote workers (multiple workers sharing a backend). See
109 /// [`WorkerSettings`](crate::WorkerSettings) and the
110 /// [Remote worker](https://docs.rs/uf-boson/latest/boson/index.html#remote-worker-two-binaries)
111 /// section on the `boson` crate.
112 ///
113 /// # Example
114 ///
115 /// ```rust,no_run
116 /// use std::sync::Arc;
117 ///
118 /// use boson_backend_mem::MemQueueBackend;
119 /// use boson_core::JsonExecutionContextFactory;
120 /// use boson_runtime::Boson;
121 ///
122 /// # fn main() -> boson_core::Result<()> {
123 /// let _boson = Boson::builder()
124 /// .queue_backend(Arc::new(MemQueueBackend::new()))
125 /// .execution_context_factory(JsonExecutionContextFactory)
126 /// .worker_id("worker-a")
127 /// .lease_ttl_secs(30)
128 /// .auto_registry()
129 /// .build()?;
130 /// # Ok(())
131 /// # }
132 /// ```
133 #[must_use]
134 pub fn worker_id(mut self, worker_id: impl Into<String>) -> Self {
135 self.worker_id = Some(worker_id.into());
136 self
137 }
138
139 /// Run lease TTL in seconds; when `> 0`, claim path acquires leases before job claim.
140 ///
141 /// Use `0` (default) for embedded monoliths. Use a positive value for remote workers so
142 /// workers do not double-execute the same run. Env override: `BOSON_LEASE_TTL_SECS`.
143 ///
144 /// See [`WorkerSettings`](crate::WorkerSettings).
145 #[must_use]
146 pub const fn lease_ttl_secs(mut self, secs: i64) -> Self {
147 self.lease_ttl_secs = Some(secs);
148 self
149 }
150
151 /// Telemetry/runtime label (default `embedded`; bench uses topology slug).
152 #[must_use]
153 pub fn runtime_label(mut self, label: impl Into<String>) -> Self {
154 self.runtime_label = Some(label.into());
155 self
156 }
157
158 /// Restrict this worker to specific pools (comma-free list). Unset = poll all queued pools.
159 ///
160 /// Also available via `BOSON_WORKER_POOLS=pool-a,pool-b`. Pin workers to disjoint pool sets
161 /// for shared-nothing scaling (each worker skips `distinct_pools_queued` fan-out).
162 #[must_use]
163 pub fn worker_pools(mut self, pools: impl IntoIterator<Item = impl Into<String>>) -> Self {
164 self.worker_pools = Some(pools.into_iter().map(Into::into).collect());
165 self
166 }
167
168 /// Milliseconds between worker poll ticks (default 50; use 0 for bench drain tests).
169 #[must_use]
170 pub const fn worker_poll_interval_ms(mut self, ms: u64) -> Self {
171 self.worker_poll_interval_ms = Some(ms);
172 self
173 }
174
175 /// Default enqueue idempotency mode when a task does not override it.
176 ///
177 /// [`boson_core::IdempotencyMode::Lwt`] (default) is exactly-once under concurrent enqueue.
178 /// [`boson_core::IdempotencyMode::None`] is at-least-once and skips coordination (higher throughput).
179 #[must_use]
180 pub const fn idempotency_mode(mut self, mode: boson_core::IdempotencyMode) -> Self {
181 self.idempotency_mode = mode;
182 self
183 }
184
185 /// Inject queue persistence backend explicitly.
186 ///
187 /// Pick the backend for your topology: [`MemQueueBackend`](https://docs.rs/boson-backend-mem)
188 /// for embedded only; `SQLite`/Postgres/Redis/NATS when processes share a queue. See the
189 /// [`boson`](https://docs.rs/uf-boson) crate backend table.
190 #[must_use]
191 pub fn queue_backend(mut self, backend: Arc<dyn QueueBackend>) -> Self {
192 self.queue_backend = Some(backend);
193 self
194 }
195
196 /// Use global [`QueueRouter`](boson_core::QueueRouter) default backend.
197 #[must_use]
198 pub fn queue_backend_from_global(mut self) -> Self {
199 self.queue_backend = None;
200 self
201 }
202
203 /// Identity factory for handler dispatch.
204 ///
205 /// Maps stored `actor_json` to `Box<dyn ExecutionContext>` when a worker runs a job. For examples
206 /// and smoke tests, pass [`JsonExecutionContextFactory`](boson_core::JsonExecutionContextFactory);
207 /// production apps typically implement [`ExecutionContextFactory`](boson_core::ExecutionContextFactory).
208 ///
209 /// # Example
210 ///
211 /// ```rust,no_run
212 /// use std::sync::Arc;
213 ///
214 /// use boson_backend_mem::MemQueueBackend;
215 /// use boson_core::JsonExecutionContextFactory;
216 /// use boson_runtime::Boson;
217 ///
218 /// # fn main() -> boson_core::Result<()> {
219 /// let _boson = Boson::builder()
220 /// .queue_backend(Arc::new(MemQueueBackend::new()))
221 /// .execution_context_factory(JsonExecutionContextFactory)
222 /// .build()?;
223 /// # Ok(())
224 /// # }
225 /// ```
226 #[must_use]
227 pub fn execution_context_factory(
228 mut self,
229 factory: impl ExecutionContextFactory + 'static,
230 ) -> Self {
231 self.execution_context_factory = Some(Arc::new(factory));
232 self
233 }
234
235 /// Identity factory from existing `Arc`.
236 #[must_use]
237 pub fn execution_context_factory_arc(
238 mut self,
239 factory: Arc<dyn ExecutionContextFactory>,
240 ) -> Self {
241 self.execution_context_factory = Some(factory);
242 self
243 }
244
245 /// Install an [`ActorJsonPolicy`] for enqueue validation (default rejects System on External).
246 #[must_use]
247 pub fn actor_json_policy(mut self, policy: impl ActorJsonPolicy + 'static) -> Self {
248 self.actor_policy = ActorPolicyChoice::Custom(Arc::new(policy));
249 self
250 }
251
252 /// Disable actor JSON policy checks at enqueue.
253 #[must_use]
254 pub fn without_actor_json_policy(mut self) -> Self {
255 self.actor_policy = ActorPolicyChoice::Disabled;
256 self
257 }
258
259 /// Install ops log adapter (default [`boson_telemetry::NoOpsLog`]).
260 ///
261 /// Prefer this or [`ops_log_console`](Self::ops_log_console) over installing an [`OpsLog`]
262 /// ad hoc. See [`boson_telemetry::OpsLog`] for adapter choices.
263 #[must_use]
264 pub fn ops_log(mut self, log: impl OpsLog + 'static) -> Self {
265 self.ops_log = Some(Arc::new(log));
266 self
267 }
268
269 /// Use console stderr ops log ([`ConsoleOpsLog`](boson_telemetry::ConsoleOpsLog)).
270 #[must_use]
271 pub fn ops_log_console(mut self) -> Self {
272 self.ops_log = Some(Arc::new(ConsoleOpsLog));
273 self
274 }
275
276 /// Use an existing task registry (e.g. testkit manual registration).
277 #[must_use]
278 pub fn registry(mut self, registry: Arc<TaskRegistry>) -> Self {
279 self.registry = Some(registry);
280 self.use_auto_registry = false;
281 self
282 }
283
284 /// Discover tasks registered via Quark inventory (for example `#[boson::task]`).
285 ///
286 /// **Worker processes** need this (or [`registry`](Self::registry)) so handlers are available
287 /// to the claim loop. **Enqueue hosts** also need it (or a manual registry) because
288 /// `send_with` / [`Boson::enqueue`] resolve task descriptors for priority, pool, and policies
289 /// — they do not run handlers.
290 ///
291 /// The binary must link every crate that defines inventory submissions; otherwise tasks
292 /// defined in library crates will not appear in the registry. Add the task-owning crate as a
293 /// dependency (for example `use my_worker as _;`).
294 ///
295 /// Getting started:
296 /// [Embedded](https://docs.rs/uf-boson/latest/boson/index.html#embedded-one-binary) /
297 /// [Remote worker](https://docs.rs/uf-boson/latest/boson/index.html#remote-worker-two-binaries).
298 ///
299 /// # Example
300 ///
301 /// ```rust,no_run
302 /// use std::sync::Arc;
303 ///
304 /// use boson_backend_mem::MemQueueBackend;
305 /// use boson_core::{ExecutionContext, JsonExecutionContextFactory};
306 /// use boson_macros::task;
307 /// use boson_runtime::{configure, Boson};
308 ///
309 /// #[task(name = "ping")]
310 /// async fn ping(_ctx: Box<dyn ExecutionContext>) -> boson_core::Result<()> {
311 /// Ok(())
312 /// }
313 ///
314 /// // When handlers live in a library crate, link it from `main`:
315 /// // use my_worker as _;
316 ///
317 /// # fn main() -> boson_core::Result<()> {
318 /// let boson = Boson::builder()
319 /// .queue_backend(Arc::new(MemQueueBackend::new()))
320 /// .execution_context_factory(JsonExecutionContextFactory)
321 /// .auto_registry()
322 /// .build()?;
323 /// configure(boson);
324 /// # Ok(())
325 /// # }
326 /// ```
327 #[must_use]
328 pub const fn auto_registry(mut self) -> Self {
329 self.use_auto_registry = true;
330 self
331 }
332
333 /// Do not spawn the background worker loop.
334 ///
335 /// Use for:
336 /// - **Remote-worker enqueue hosts** — this process only [`configure`](crate::configure)s and
337 /// calls `send_with`; a separate worker binary drains the shared backend
338 /// - **Tests** — pair with [`build_manual`](Self::build_manual) and
339 /// [`ManualWorker`](crate::ManualWorker)
340 ///
341 /// Getting started:
342 /// [Enqueue binary](https://docs.rs/uf-boson/latest/boson/index.html#enqueue-binary).
343 ///
344 /// # Example — enqueue-only process
345 ///
346 /// ```rust,no_run
347 /// use std::sync::Arc;
348 ///
349 /// use boson_backend_mem::MemQueueBackend;
350 /// use boson_core::JsonExecutionContextFactory;
351 /// use boson_runtime::{configure, Boson};
352 ///
353 /// # fn main() -> boson_core::Result<()> {
354 /// let boson = Boson::builder()
355 /// .queue_backend(Arc::new(MemQueueBackend::new()))
356 /// .execution_context_factory(JsonExecutionContextFactory)
357 /// .auto_registry()
358 /// .without_worker()
359 /// .build()?;
360 /// configure(boson);
361 /// # Ok(())
362 /// # }
363 /// ```
364 #[must_use]
365 pub const fn without_worker(mut self) -> Self {
366 self.spawn_worker = false;
367 self
368 }
369}
370
371impl Boson {
372 /// Create a new builder.
373 #[must_use]
374 pub fn builder() -> BosonBuilder {
375 BosonBuilder {
376 spawn_worker: true,
377 ..Default::default()
378 }
379 }
380}