chronon_runtime/
builder.rs

1//! [`ChrononBuilder`] and [`DeploymentShape`] — runtime configuration and assembly.
2
3use std::sync::Arc;
4
5use chronon_core::context::{ContextFactory, NoOpContextFactory};
6use chronon_core::error::{ChrononError, Result};
7use chronon_core::store::SchedulerStore;
8use chronon_executor::{
9    event_channel_capacity_from_env, executor_concurrency_from_env, Executor, ScriptRegistry,
10};
11use chronon_scheduler::{Scheduler, SchedulerConfig};
12use chronon_telemetry::{NoOpSink, TelemetrySink};
13use tokio::sync::{mpsc, Notify};
14
15/// Named deployment assembly — process topology shape.
16///
17/// Selected via [`ChrononBuilder`] fluent methods; drives which loops
18/// [`crate::Chronon::run`] starts. Maps to the public crate getting-started topologies:
19///
20/// | Shape | Builder | Local loops | When to use |
21/// |-------|---------|-------------|-------------|
22/// | [`Self::Embedded`] | [`.embedded()`](ChrononBuilder::embedded) | Tick **and** worker | Embedded — one process |
23/// | [`Self::CoordinatorOnly`] | [`.coordinator_only()`](ChrononBuilder::coordinator_only) | Tick only | Coordinator–worker coordinator binary |
24/// | [`Self::Worker`] | [`.worker(pool)`](ChrononBuilder::worker) | Claim + execute | Coordinator–worker worker binary(ies) |
25/// | [`Self::RemoteClient`] | [`.remote_coordinator(url)`](ChrononBuilder::remote_coordinator) | **None** | Remote HTTP client — schedule via HTTP |
26///
27/// Coordinator–worker and remote HTTP client need a shared durable store (Postgres, usually +
28/// Redis) on the host that runs Chronon loops. See [`crate::RemoteCoordinatorClient`] for the
29/// HTTP client path.
30///
31/// # Examples
32///
33/// ```
34/// use chronon_runtime::DeploymentShape;
35///
36/// assert_eq!(DeploymentShape::default(), DeploymentShape::Embedded);
37/// let worker = DeploymentShape::Worker("general".into());
38/// assert!(matches!(worker, DeploymentShape::Worker(_)));
39/// ```
40#[derive(Debug, Clone, PartialEq, Eq, Default)]
41pub enum DeploymentShape {
42    /// Coordinator tick loop + worker in one process (embedded default).
43    #[default]
44    Embedded,
45    /// Scheduler tick and partition assigner only; no script execution (coordinator binary).
46    CoordinatorOnly,
47    /// Worker loop for `pool_id`; claims runs from the shared store (worker binary).
48    Worker(String),
49    /// No local loops; host uses [`crate::RemoteCoordinatorClient`] against `base_url`.
50    RemoteClient(String),
51}
52
53/// Builds a [`crate::Chronon`] runtime with explicit adapter injection.
54///
55/// Hosts call fluent setters then [`Self::build`]. Missing store is the only hard requirement;
56/// context factory, telemetry, and registry fall back to no-op defaults.
57///
58/// Choose topology with [`.embedded()`](Self::embedded) / [`.coordinator_only()`](Self::coordinator_only) /
59/// [`.worker()`](Self::worker) / [`.remote_coordinator()`](Self::remote_coordinator) — see
60/// [`DeploymentShape`]. Use [`.auto_registry()`](Self::auto_registry) to pick up `#[chronon::script]`
61/// handlers linked into this binary (required on coordinator–worker **workers**).
62///
63/// # Examples
64///
65/// Embedded with an empty registry:
66///
67/// ```
68/// use std::sync::Arc;
69/// use chronon_backend_mem::InMemorySchedulerStore;
70/// use chronon_runtime::{ChrononBuilder, DeploymentShape};
71///
72/// let store = Arc::new(InMemorySchedulerStore::new());
73/// let chronon = ChrononBuilder::new()
74///     .scheduler_store(store)
75///     .embedded()
76///     .build()
77///     .unwrap();
78/// assert_eq!(chronon.deployment, DeploymentShape::Embedded);
79/// assert_eq!(chronon.executor().script_count(), 0);
80/// ```
81///
82/// Worker shape (scripts must be registered on this binary):
83///
84/// ```
85/// use std::sync::Arc;
86/// use chronon_backend_mem::InMemorySchedulerStore;
87/// use chronon_runtime::{ChrononBuilder, DeploymentShape};
88///
89/// let chronon = ChrononBuilder::new()
90///     .scheduler_store(Arc::new(InMemorySchedulerStore::new()))
91///     .instance_id("worker-a")
92///     .worker("general")
93///     .build()
94///     .unwrap();
95/// assert!(matches!(chronon.deployment, DeploymentShape::Worker(_)));
96/// ```
97pub struct ChrononBuilder {
98    store: Option<Arc<dyn SchedulerStore>>,
99    context_factory: Option<Arc<dyn ContextFactory>>,
100    telemetry: Option<Arc<dyn TelemetrySink>>,
101    registry: Option<Arc<ScriptRegistry>>,
102    deployment: DeploymentShape,
103    auto_registry: bool,
104    tick_interval_ms: u64,
105    instance_id: Option<String>,
106}
107
108impl ChrononBuilder {
109    /// Empty builder: embedded deployment, env-default tick interval, no store.
110    pub fn new() -> Self {
111        Self {
112            store: None,
113            context_factory: None,
114            telemetry: None,
115            registry: None,
116            deployment: DeploymentShape::Embedded,
117            auto_registry: false,
118            tick_interval_ms: chronon_scheduler::tick_interval_ms_from_env(),
119            instance_id: None,
120        }
121    }
122
123    /// Required unless [`Self::scheduler_store_from_global`] is used.
124    pub fn scheduler_store(mut self, store: Arc<dyn SchedulerStore>) -> Self {
125        self.store = Some(store);
126        self
127    }
128
129    /// Installs the process-global default store (e.g. mem backend); errors if unset.
130    pub fn scheduler_store_from_global(mut self) -> Result<Self> {
131        self.store = Some(chronon_core::default_store_from_global()?);
132        Ok(self)
133    }
134
135    /// Factory used when executing scripts; defaults to [`NoOpContextFactory`].
136    pub fn context_factory(mut self, factory: Arc<dyn ContextFactory>) -> Self {
137        self.context_factory = Some(factory);
138        self
139    }
140
141    /// Metrics sink shared by scheduler and executor; defaults to [`NoOpSink`].
142    pub fn telemetry_sink(mut self, sink: Arc<dyn TelemetrySink>) -> Self {
143        self.telemetry = Some(sink);
144        self
145    }
146
147    /// Script registry for the executor; use [`Self::auto_registry`] to populate from inventory.
148    pub fn script_registry(mut self, registry: Arc<ScriptRegistry>) -> Self {
149        self.registry = Some(registry);
150        self
151    }
152
153    /// Stable id for scheduler leader election and worker rows; random UUID if omitted.
154    pub fn instance_id(mut self, id: impl Into<String>) -> Self {
155        self.instance_id = Some(id.into());
156        self
157    }
158
159    /// Embedded coordinator + worker loops in one process.
160    pub fn embedded(mut self) -> Self {
161        self.deployment = DeploymentShape::Embedded;
162        self
163    }
164
165    /// Coordinator-only: tick and partition assigner, no worker slots.
166    pub fn coordinator_only(mut self) -> Self {
167        self.deployment = DeploymentShape::CoordinatorOnly;
168        self
169    }
170
171    /// Worker-only: claim and execute runs for `pool_id`.
172    pub fn worker(mut self, pool_id: impl Into<String>) -> Self {
173        self.deployment = DeploymentShape::Worker(pool_id.into());
174        self
175    }
176
177    /// Remote client shape: no local loops; pair with [`crate::RemoteCoordinatorClient`].
178    ///
179    /// [`crate::Chronon::run`] returns an error for this shape — schedule via the HTTP client.
180    pub fn remote_coordinator(mut self, base_url: impl Into<String>) -> Self {
181        self.deployment = DeploymentShape::RemoteClient(base_url.into());
182        self
183    }
184
185    /// Populate registry from `inventory` (`#[chronon::script]` link-time registration).
186    ///
187    /// In a coordinator–worker split, call this on **worker** binaries (that is where scripts
188    /// execute).
189    pub fn auto_registry(mut self) -> Self {
190        self.auto_registry = true;
191        self
192    }
193
194    /// Scheduler tick period in milliseconds; overrides `CHRONON_TICK_INTERVAL_MS` when set.
195    pub fn tick_interval_ms(mut self, ms: u64) -> Self {
196        self.tick_interval_ms = ms;
197        self
198    }
199
200    /// Assemble [`crate::Chronon`]; returns [`ChrononError::Internal`] if store was not configured.
201    pub fn build(self) -> Result<super::Chronon> {
202        let store = self
203            .store
204            .ok_or_else(|| ChrononError::Internal("scheduler_store is required".into()))?;
205        let context_factory = self
206            .context_factory
207            .unwrap_or_else(|| Arc::new(NoOpContextFactory));
208        let telemetry = self
209            .telemetry
210            .unwrap_or_else(|| Arc::new(NoOpSink) as Arc<dyn TelemetrySink>);
211        let registry = match self.registry {
212            Some(registry) => registry,
213            None if self.auto_registry => Arc::new(ScriptRegistry::from_inventory()),
214            None => Arc::new(ScriptRegistry::new()),
215        };
216
217        let embedded_partitions = matches!(self.deployment, DeploymentShape::Embedded);
218        let instance_id = self
219            .instance_id
220            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
221
222        let scheduler = Arc::new(Scheduler::new(
223            SchedulerConfig {
224                tick_interval_ms: self.tick_interval_ms,
225                instance_id,
226                num_partitions: chronon_scheduler::num_partitions_from_env(),
227                embedded: embedded_partitions,
228            },
229            store.clone(),
230            telemetry.clone(),
231        ));
232
233        let (event_tx, event_rx) = mpsc::channel(event_channel_capacity_from_env());
234        let executor = Arc::new(Executor::new(
235            registry,
236            context_factory,
237            telemetry,
238            event_tx,
239            executor_concurrency_from_env(),
240        ));
241
242        Ok(super::Chronon::new(
243            store,
244            scheduler,
245            executor,
246            self.deployment,
247            Arc::new(Notify::new()),
248            event_rx,
249        ))
250    }
251}
252
253impl Default for ChrononBuilder {
254    fn default() -> Self {
255        Self::new()
256    }
257}
258
259/// Shorthand for [`ChrononBuilder::new`].
260pub fn builder() -> ChrononBuilder {
261    ChrononBuilder::new()
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267    use chronon_backend_mem::InMemorySchedulerStore;
268    use chronon_telemetry::ConsoleSink;
269
270    #[tokio::test]
271    async fn builder_embedded_compiles() {
272        let store = Arc::new(InMemorySchedulerStore::new());
273        let chronon = ChrononBuilder::new()
274            .scheduler_store(store)
275            .telemetry_sink(Arc::new(ConsoleSink))
276            .embedded()
277            .build()
278            .expect("build");
279        assert_eq!(chronon.deployment, DeploymentShape::Embedded);
280    }
281
282    #[tokio::test]
283    async fn builder_auto_registry_from_inventory() {
284        let store = Arc::new(InMemorySchedulerStore::new());
285        let chronon = ChrononBuilder::new()
286            .scheduler_store(store)
287            .embedded()
288            .auto_registry()
289            .build()
290            .expect("build");
291        let _ = chronon.executor().script_count();
292    }
293
294    #[tokio::test]
295    async fn builder_scheduler_store_from_global() {
296        let _installed = chronon_backend_mem::install_default_mem_store();
297        let chronon = ChrononBuilder::new()
298            .scheduler_store_from_global()
299            .expect("global store")
300            .embedded()
301            .build()
302            .expect("build");
303        assert_eq!(chronon.deployment, DeploymentShape::Embedded);
304    }
305}
chronon_runtime/builder.rs

chronon_runtime/
builder.rs
