chronon_executor/
lib.rs

1//! Script registry lookup, context build, and async run lifecycle.
2//!
3//! Resolves registered script handlers, builds execution context from stored actor JSON,
4//! and dispatches async runs with lifecycle events back to the runtime.
5//!
6//! # Documentation map
7//!
8//! - **Register handlers** — [`ScriptRegistry`], link-time inventory via `#[chronon::script]`
9//! - **Dispatch runs** — [`Executor::spawn_run`], [`execute_script`]
10//! - **Observe lifecycle** — [`ExecutorEvent`]
11//!
12//! # Notes
13//!
14//! [`Executor::spawn_run`] uses run-level `params_json`, not job defaults. Missing scripts
15//! surface as [`ChrononError::ScriptNotFound`](chronon_core::ChrononError::ScriptNotFound).
16
17mod descriptor;
18mod invoke;
19mod registry;
20
21pub use descriptor::{InvokeFn, ScriptDescriptor};
22pub use invoke::{execute_script, ExecuteScriptRequest};
23pub use registry::{ScriptDescriptorRef, ScriptRegistry};
24
25use std::sync::Arc;
26
27use chrono::Utc;
28use chronon_core::{ContextFactory, Job, Run};
29use chronon_telemetry::TelemetrySink;
30use tokio::sync::mpsc;
31
32/// Event sent from the executor to the runtime for run status updates.
33///
34/// Consumed by `chronon-runtime` to persist run state and forward metrics.
35#[derive(Debug, Clone)]
36pub enum ExecutorEvent {
37    /// A run task was spawned and execution has begun.
38    RunStarted {
39        /// Run identifier matching [`Run::run_id`](chronon_core::Run::run_id).
40        run_id: String,
41    },
42    /// Handler returned successfully.
43    RunCompleted {
44        /// Run identifier matching [`Run::run_id`](chronon_core::Run::run_id).
45        run_id: String,
46        /// Wall-clock duration from spawn to handler completion, in milliseconds.
47        duration_ms: i64,
48    },
49    /// Handler returned an error or context build failed.
50    RunFailed {
51        /// Run identifier matching [`Run::run_id`](chronon_core::Run::run_id).
52        run_id: String,
53        /// Display-formatted error message for logs and persistence.
54        error: String,
55    },
56}
57
58/// Executor for running registered scripts against scheduled jobs.
59///
60/// Constructed by `ChrononBuilder` in `chronon-runtime` and called when workers claim runs.
61pub struct Executor {
62    /// Script catalog used to resolve handler functions by name.
63    pub registry: Arc<ScriptRegistry>,
64    /// Rebuilds [`ScriptContext`](chronon_core::ScriptContext) from job `actor_json`.
65    pub context_factory: Arc<dyn ContextFactory>,
66    /// Metrics and structured error events for invoke phases.
67    pub telemetry: Arc<dyn TelemetrySink>,
68    event_tx: mpsc::UnboundedSender<ExecutorEvent>,
69}
70
71impl Executor {
72    /// Builds an executor wired to the given registry, factory, telemetry, and event channel.
73    ///
74    /// The runtime typically clones [`Self::event_sender`] before passing `event_tx` so both
75    /// sides can send lifecycle updates.
76    pub fn new(
77        registry: Arc<ScriptRegistry>,
78        context_factory: Arc<dyn ContextFactory>,
79        telemetry: Arc<dyn TelemetrySink>,
80        event_tx: mpsc::UnboundedSender<ExecutorEvent>,
81    ) -> Self {
82        Self {
83            registry,
84            context_factory,
85            telemetry,
86            event_tx,
87        }
88    }
89
90    /// Clones the unbounded sender for [`ExecutorEvent`] lifecycle updates.
91    ///
92    /// Used by the runtime to subscribe without holding an [`Executor`] reference.
93    pub fn event_sender(&self) -> mpsc::UnboundedSender<ExecutorEvent> {
94        self.event_tx.clone()
95    }
96
97    /// Returns the number of scripts currently registered.
98    pub fn script_count(&self) -> usize {
99        self.registry.len()
100    }
101
102    /// Spawn asynchronous execution for one run of the given job.
103    ///
104    /// Emits [`ExecutorEvent::RunStarted`] immediately, then invokes the script via
105    /// [`execute_script`]. Run `params_json` takes precedence over job defaults.
106    pub fn spawn_run(&self, job: &Job, run: Run) {
107        let registry = Arc::clone(&self.registry);
108        let context_factory = Arc::clone(&self.context_factory);
109        let telemetry = Arc::clone(&self.telemetry);
110        let event_tx = self.event_tx.clone();
111
112        let script_name = job.script_name.clone();
113        let job_name = job.job_name.clone();
114        let params_json = run.params_json.clone();
115        let actor_json = job.actor_json.clone();
116        let run_id = run.run_id;
117
118        tokio::spawn(async move {
119            let _ = event_tx.send(ExecutorEvent::RunStarted {
120                run_id: run_id.clone(),
121            });
122            telemetry.record_counter(
123                "chronon_runs_started",
124                &[("script", script_name.as_str()), ("job", job_name.as_str())],
125                1,
126            );
127
128            let started = Utc::now();
129            let result = invoke::execute_script(invoke::ExecuteScriptRequest {
130                registry: &registry,
131                context_factory: &context_factory,
132                telemetry: &telemetry,
133                script_name: &script_name,
134                actor_json: &actor_json,
135                params_json,
136                job_name: &job_name,
137                run_id: &run_id,
138            })
139            .await;
140
141            let duration_ms = (Utc::now() - started).num_milliseconds();
142            match result {
143                Ok(()) => {
144                    let _ = event_tx.send(ExecutorEvent::RunCompleted {
145                        run_id: run_id.clone(),
146                        duration_ms,
147                    });
148                    telemetry.record_counter(
149                        "chronon_runs_completed",
150                        &[("script", script_name.as_str()), ("job", job_name.as_str())],
151                        1,
152                    );
153                }
154                Err(e) => {
155                    let error_msg = e.to_string();
156                    let _ = event_tx.send(ExecutorEvent::RunFailed {
157                        run_id: run_id.clone(),
158                        error: error_msg.clone(),
159                    });
160                    telemetry.record_counter(
161                        "chronon_runs_failed",
162                        &[("script", script_name.as_str()), ("job", job_name.as_str())],
163                        1,
164                    );
165                    telemetry.log_event(
166                        "chronon_run_failed",
167                        &[
168                            ("run_id", run_id.as_str()),
169                            ("job", job_name.as_str()),
170                            ("error", error_msg.as_str()),
171                        ],
172                    );
173                }
174            }
175        });
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use chronon_core::{NoOpContextFactory, Result, ScriptContext};
183    use serde_json::{json, Value};
184    use std::future::Future;
185    use std::pin::Pin;
186    use std::sync::Mutex;
187
188    static LAST_PARAMS: Mutex<Option<Value>> = Mutex::new(None);
189
190    fn param_probe(
191        _ctx: Box<dyn ScriptContext>,
192        params: Value,
193    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
194        Box::pin(async move {
195            *LAST_PARAMS.lock().unwrap() = Some(params);
196            Ok(())
197        })
198    }
199
200    #[tokio::test]
201    async fn spawn_run_uses_run_params() {
202        *LAST_PARAMS.lock().unwrap() = None;
203        let registry = Arc::new({
204            let mut r = ScriptRegistry::new();
205            r.register(ScriptDescriptor::new("probe", param_probe));
206            r
207        });
208        let (tx, mut rx) = mpsc::unbounded_channel();
209        let executor = Executor::new(
210            registry,
211            Arc::new(NoOpContextFactory),
212            Arc::new(chronon_telemetry::NoOpSink),
213            tx,
214        );
215
216        let mut job = Job::new("job", "probe");
217        let mut run = chronon_core::Run::for_job(&job.job_id, "probe", Utc::now());
218        run.params_json = json!({ "source": "run" });
219        job.params_json = json!({ "source": "job" });
220
221        executor.spawn_run(&job, run);
222
223        for _ in 0..20 {
224            if let Some(ExecutorEvent::RunCompleted { .. }) = rx.recv().await {
225                break;
226            }
227        }
228        assert_eq!(
229            *LAST_PARAMS.lock().unwrap(),
230            Some(json!({ "source": "run" }))
231        );
232    }
233}
chronon_executor/lib.rs

chronon_executor/
lib.rs
