boson_runtime/
builder.rs

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

boson_runtime/
builder.rs
