chronon_executor/
invoke.rs

1//! Synchronous script invocation (registry lookup, context build, invoke).
2
3use std::sync::Arc;
4
5use chronon_core::{ChrononError, ContextFactory, Result};
6use chronon_telemetry::TelemetrySink;
7use serde_json::Value;
8
9use crate::registry::ScriptRegistry;
10
11/// Inputs for a single script execution attempt.
12pub struct ExecuteScriptRequest<'a> {
13    /// Script registry containing the target handler.
14    pub registry: &'a ScriptRegistry,
15    /// Factory that rebuilds [`ScriptContext`](chronon_core::ScriptContext) from stored actor JSON.
16    pub context_factory: &'a Arc<dyn ContextFactory>,
17    /// Sink for executor metrics and error events.
18    pub telemetry: &'a Arc<dyn TelemetrySink>,
19    /// Registered script name to invoke.
20    pub script_name: &'a str,
21    /// Actor JSON persisted on the job at schedule time.
22    pub actor_json: &'a Value,
23    /// Run-specific parameters JSON.
24    pub params_json: Value,
25    /// Human-readable job name for telemetry.
26    pub job_name: &'a str,
27    /// Run identifier for telemetry correlation.
28    pub run_id: &'a str,
29}
30
31fn record_executor_error(
32    telemetry: &Arc<dyn TelemetrySink>,
33    job_name: &str,
34    run_id: &str,
35    script_name: &str,
36    phase: &str,
37    message: &str,
38) {
39    telemetry.log_event(
40        "chronon_executor_error",
41        &[
42            ("job_name", job_name),
43            ("run_id", run_id),
44            ("script_name", script_name),
45            ("phase", phase),
46            ("message", message),
47        ],
48    );
49}
50
51/// Execute a script synchronously.
52pub async fn execute_script(req: ExecuteScriptRequest<'_>) -> Result<()> {
53    let ExecuteScriptRequest {
54        registry,
55        context_factory,
56        telemetry,
57        script_name,
58        actor_json,
59        params_json,
60        job_name,
61        run_id,
62    } = req;
63
64    let descriptor = registry.get_or_err(script_name).inspect_err(|e| {
65        record_executor_error(
66            telemetry,
67            job_name,
68            run_id,
69            script_name,
70            "registry_lookup",
71            &e.to_string(),
72        );
73    })?;
74
75    let ctx = context_factory.build(actor_json).map_err(|e| {
76        let msg = format!("Failed to build script context: {e}");
77        record_executor_error(
78            telemetry,
79            job_name,
80            run_id,
81            script_name,
82            "context_build",
83            &msg,
84        );
85        ChrononError::Internal(msg)
86    })?;
87
88    (descriptor.invoke)(ctx, params_json)
89        .await
90        .map_err(|e| {
91            record_executor_error(
92                telemetry,
93                job_name,
94                run_id,
95                script_name,
96                "script_invoke",
97                &e.to_string(),
98            );
99            map_invoke_error(e.to_string())
100        })
101}
102
103fn map_invoke_error(message: String) -> ChrononError {
104    if is_likely_param_error(&message) {
105        ChrononError::ParamError(message)
106    } else {
107        ChrononError::Internal(message)
108    }
109}
110
111fn is_likely_param_error(message: &str) -> bool {
112    const PARAM_ERROR_HINTS: [&str; 6] = [
113        "missing field",
114        "invalid type",
115        "expected",
116        "unknown field",
117        "parameter error",
118        "deserializing",
119    ];
120    let lower = message.to_ascii_lowercase();
121    PARAM_ERROR_HINTS.iter().any(|h| lower.contains(h))
122}
123
124#[cfg(test)]
125mod tests {
126    use super::*;
127    use crate::descriptor::ScriptDescriptor;
128    use chronon_core::{NoOpContextFactory, Result, ScriptContext};
129    use serde_json::Value;
130    use std::future::Future;
131    use std::pin::Pin;
132
133    fn noop_invoke(
134        _ctx: Box<dyn ScriptContext>,
135        _params: Value,
136    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> {
137        Box::pin(async { Ok(()) })
138    }
139
140    #[tokio::test]
141    async fn execute_registered_script() {
142        let mut registry = ScriptRegistry::new();
143        registry.register(ScriptDescriptor::new("test_script", noop_invoke));
144        let factory: Arc<dyn chronon_core::ContextFactory> = Arc::new(NoOpContextFactory);
145        let telemetry: Arc<dyn chronon_telemetry::TelemetrySink> =
146            Arc::new(chronon_telemetry::NoOpSink);
147        let result = execute_script(ExecuteScriptRequest {
148            registry: &registry,
149            context_factory: &factory,
150            telemetry: &telemetry,
151            script_name: "test_script",
152            actor_json: &Value::Null,
153            params_json: Value::Object(serde_json::Map::default()),
154            job_name: "job",
155            run_id: "run-1",
156        })
157        .await;
158        assert!(result.is_ok());
159    }
160}
chronon_executor/invoke.rs

chronon_executor/
invoke.rs
