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

chronon_runtime/
builder.rs
