chronon_runtime/
runtime.rs

1//! Assembled [`Chronon`] runtime and background loop dispatch.
2
3use std::sync::Arc;
4
5use chronon_core::store::SchedulerStore;
6use chronon_executor::Executor;
7use chronon_scheduler::Scheduler;
8use tokio::sync::Notify;
9
10use crate::builder::DeploymentShape;
11use crate::coordinator_service::CoordinatorService;
12use crate::coordinator::run_coordinator_loops;
13use crate::embedded::run_embedded_loops;
14use crate::events::spawn_event_handler;
15use crate::worker::run_worker_loop;
16
17/// Assembled Chronon runtime: store, scheduler, executor, and deployment loops.
18///
19/// Built via [`ChrononBuilder`](crate::ChrononBuilder). Hosts typically:
20///
21/// 1. Upsert jobs through [`Self::coordinator_service`] (or HTTP / remote client).
22/// 2. Call [`Self::run`] to start shape-specific loops (or [`Self::tick_once`] in tests).
23/// 3. Call [`Self::shutdown`] on exit.
24///
25/// | [`DeploymentShape`](crate::DeploymentShape) | [`Self::run`] behavior |
26/// |---------------------------------------------|------------------------|
27/// | `Embedded` | Tick + worker in this process |
28/// | `CoordinatorOnly` | Tick + partitions only |
29/// | `Worker(pool)` | Claim + execute for `pool` |
30/// | `RemoteClient(_)` | **Error** — use [`crate::RemoteCoordinatorClient`] |
31///
32/// # Examples
33///
34/// ```
35/// use std::sync::Arc;
36/// use chronon_backend_mem::InMemorySchedulerStore;
37/// use chronon_runtime::{ChrononBuilder, DeploymentShape};
38///
39/// let chronon = ChrononBuilder::new()
40///     .scheduler_store(Arc::new(InMemorySchedulerStore::new()))
41///     .embedded()
42///     .build()
43///     .unwrap();
44/// assert_eq!(chronon.deployment, DeploymentShape::Embedded);
45/// let _ = chronon.coordinator_service();
46/// ```
47pub struct Chronon {
48    /// Shared persistence for jobs, runs, partitions, and workers.
49    pub store: Arc<dyn SchedulerStore>,
50    /// Tick engine and partition assigner handle.
51    pub scheduler: Arc<Scheduler>,
52    /// Script registry and async dispatch.
53    pub executor: Arc<Executor>,
54    /// Job/run CRUD facade over [`Self::store`].
55    pub coordinator: CoordinatorService,
56    /// Deployment shape selected at build time.
57    pub deployment: DeploymentShape,
58    shutdown: Arc<Notify>,
59    event_rx: Option<tokio::sync::mpsc::UnboundedReceiver<chronon_executor::ExecutorEvent>>,
60}
61
62impl Chronon {
63    pub(crate) fn new(
64        store: Arc<dyn SchedulerStore>,
65        scheduler: Arc<Scheduler>,
66        executor: Arc<Executor>,
67        deployment: DeploymentShape,
68        shutdown: Arc<Notify>,
69        event_rx: tokio::sync::mpsc::UnboundedReceiver<chronon_executor::ExecutorEvent>,
70    ) -> Self {
71        let coordinator = CoordinatorService::new(store.clone());
72        Self {
73            store,
74            scheduler,
75            executor,
76            coordinator,
77            deployment,
78            shutdown,
79            event_rx: Some(event_rx),
80        }
81    }
82
83    /// Signal all runtime loops to stop.
84    pub fn shutdown(&self) {
85        self.shutdown.notify_waiters();
86    }
87
88    /// Shared shutdown signal for background [`Self::run`] tasks (testkit / host wiring).
89    pub fn shutdown_handle(&self) -> Arc<Notify> {
90        Arc::clone(&self.shutdown)
91    }
92
93    /// Run deployment loops until [`Self::shutdown`] is called.
94    ///
95    /// Dispatches on [`Self::deployment`]:
96    ///
97    /// - [`DeploymentShape::Embedded`](crate::DeploymentShape::Embedded) — tick + worker
98    /// - [`DeploymentShape::CoordinatorOnly`](crate::DeploymentShape::CoordinatorOnly) — tick only
99    /// - [`DeploymentShape::Worker`](crate::DeploymentShape::Worker) — claim + execute
100    /// - [`DeploymentShape::RemoteClient`](crate::DeploymentShape::RemoteClient) — returns
101    ///   [`ChrononError::Internal`](chronon_core::ChrononError::Internal); use
102    ///   [`crate::RemoteCoordinatorClient`] instead
103    ///
104    /// Call `scheduler.init_partitions().await` before [`Self::run`] on coordinator /
105    /// embedded shapes so partition ownership is ready.
106    ///
107    /// # Examples
108    ///
109    /// Embedded: init partitions, then run until shutdown:
110    ///
111    /// ```no_run
112    /// use std::sync::Arc;
113    /// use chronon_backend_mem::InMemorySchedulerStore;
114    /// use chronon_runtime::ChrononBuilder;
115    ///
116    /// # async fn demo() -> chronon_core::Result<()> {
117    /// let mut chronon = ChrononBuilder::new()
118    ///     .scheduler_store(Arc::new(InMemorySchedulerStore::new()))
119    ///     .embedded()
120    ///     .build()?;
121    /// chronon.scheduler.init_partitions().await;
122    /// // chronon.run().await?;  // blocks until chronon.shutdown()
123    /// # let _ = chronon;
124    /// # Ok(())
125    /// # }
126    /// ```
127    ///
128    /// Runnable daemons: `cargo run -p uf-chronon --example coordinator_daemon --features postgres,redis`
129    /// and `worker_daemon`.
130    pub async fn run(&mut self) -> chronon_core::Result<()> {
131        if let Some(rx) = self.event_rx.take() {
132            spawn_event_handler(Arc::clone(&self.store), rx);
133        }
134
135        let shutdown = Arc::clone(&self.shutdown);
136        let scheduler = Arc::clone(&self.scheduler);
137        let executor = Arc::clone(&self.executor);
138        let telemetry = scheduler.telemetry();
139
140        match &self.deployment {
141            DeploymentShape::RemoteClient(_) => {
142                return Err(chronon_core::ChrononError::Internal(
143                    "remote client does not run local scheduler loop".into(),
144                ));
145            }
146            DeploymentShape::Embedded => {
147                run_embedded_loops(scheduler, executor, telemetry, shutdown).await;
148            }
149            DeploymentShape::CoordinatorOnly => {
150                run_coordinator_loops(scheduler, telemetry, shutdown).await;
151            }
152            DeploymentShape::Worker(pool) => {
153                run_worker_loop(
154                    self.store.clone(),
155                    executor,
156                    telemetry,
157                    pool.clone(),
158                    scheduler.instance_id().to_string(),
159                    shutdown,
160                )
161                .await;
162            }
163        }
164        Ok(())
165    }
166
167    /// Advance the scheduler by one tick (tests and manual stepping).
168    pub async fn tick_once(&self) -> chronon_core::Result<chronon_scheduler::TickResult> {
169        self.scheduler.tick_once().await
170    }
171
172    /// Job and run CRUD for HTTP handlers and host integration.
173    pub fn coordinator_service(&self) -> &CoordinatorService {
174        &self.coordinator
175    }
176
177    /// Script executor (registry + dispatch).
178    pub fn executor(&self) -> &Executor {
179        &self.executor
180    }
181}
chronon_runtime/runtime.rs

chronon_runtime/
runtime.rs
