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 spawn
20/// [`Chronon::run`] on a background task and call [`Chronon::shutdown`] on exit.
21pub struct Chronon {
22    /// Shared persistence for jobs, runs, partitions, and workers.
23    pub store: Arc<dyn SchedulerStore>,
24    /// Tick engine and partition assigner handle.
25    pub scheduler: Arc<Scheduler>,
26    /// Script registry and async dispatch.
27    pub executor: Arc<Executor>,
28    /// Job/run CRUD facade over [`Self::store`].
29    pub coordinator: CoordinatorService,
30    /// Deployment shape selected at build time.
31    pub deployment: DeploymentShape,
32    shutdown: Arc<Notify>,
33    event_rx: Option<tokio::sync::mpsc::UnboundedReceiver<chronon_executor::ExecutorEvent>>,
34}
35
36impl Chronon {
37    pub(crate) fn new(
38        store: Arc<dyn SchedulerStore>,
39        scheduler: Arc<Scheduler>,
40        executor: Arc<Executor>,
41        deployment: DeploymentShape,
42        shutdown: Arc<Notify>,
43        event_rx: tokio::sync::mpsc::UnboundedReceiver<chronon_executor::ExecutorEvent>,
44    ) -> Self {
45        let coordinator = CoordinatorService::new(store.clone());
46        Self {
47            store,
48            scheduler,
49            executor,
50            coordinator,
51            deployment,
52            shutdown,
53            event_rx: Some(event_rx),
54        }
55    }
56
57    /// Signal all runtime loops to stop.
58    pub fn shutdown(&self) {
59        self.shutdown.notify_waiters();
60    }
61
62    /// Shared shutdown signal for background [`Self::run`] tasks (testkit / host wiring).
63    pub fn shutdown_handle(&self) -> Arc<Notify> {
64        Arc::clone(&self.shutdown)
65    }
66
67    /// Run deployment loops until [`Self::shutdown`] is called.
68    ///
69    /// Returns an error for [`DeploymentShape::RemoteClient`] because no local loops apply.
70    pub async fn run(&mut self) -> chronon_core::Result<()> {
71        if let Some(rx) = self.event_rx.take() {
72            spawn_event_handler(Arc::clone(&self.store), rx);
73        }
74
75        let shutdown = Arc::clone(&self.shutdown);
76        let scheduler = Arc::clone(&self.scheduler);
77        let executor = Arc::clone(&self.executor);
78        let telemetry = scheduler.telemetry();
79
80        match &self.deployment {
81            DeploymentShape::RemoteClient(_) => {
82                return Err(chronon_core::ChrononError::Internal(
83                    "remote client does not run local scheduler loop".into(),
84                ));
85            }
86            DeploymentShape::Embedded => {
87                run_embedded_loops(scheduler, executor, telemetry, shutdown).await;
88            }
89            DeploymentShape::CoordinatorOnly => {
90                run_coordinator_loops(scheduler, telemetry, shutdown).await;
91            }
92            DeploymentShape::Worker(pool) => {
93                run_worker_loop(
94                    self.store.clone(),
95                    executor,
96                    telemetry,
97                    pool.clone(),
98                    scheduler.instance_id().to_string(),
99                    shutdown,
100                )
101                .await;
102            }
103        }
104        Ok(())
105    }
106
107    /// Advance the scheduler by one tick (tests and manual stepping).
108    pub async fn tick_once(&self) -> chronon_core::Result<chronon_scheduler::TickResult> {
109        self.scheduler.tick_once().await
110    }
111
112    /// Job and run CRUD for HTTP handlers and host integration.
113    pub fn coordinator_service(&self) -> &CoordinatorService {
114        &self.coordinator
115    }
116
117    /// Script executor (registry + dispatch).
118    pub fn executor(&self) -> &Executor {
119        &self.executor
120    }
121}
chronon_runtime/runtime.rs

chronon_runtime/
runtime.rs
