radixdb_executor/procedural/
job.rs

1use chrono::{DateTime, Utc};
2use radixdb_catalog::{
3    CatalogName, CatalogPayload, JobArgument, JobSchedule, ObjectId, ObjectKind, ResourcePolicy,
4};
5use radixdb_core::{Error, Result, Value};
6use radixdb_procedural::{
7    Diagnostic, DiagnosticCategory, DiagnosticKind, PrincipalContext, RuntimeValue,
8};
9use radixdb_sql::{CreateJobStatement, Expression, JobScheduleSyntax};
10use radixdb_storage::mvcc::persistence::{deserialize_value, serialize_value};
11
12use crate::catalog::BoundJobDefinition;
13use crate::context::ExecutionContext;
14use crate::Executor;
15
16use super::call::{bind_job_call, CallStatementStage};
17use super::error::map_executor_error;
18use super::{transaction_visible_catalog, ProceduralCallOutcome, ProceduralResultStage};
19
20const MAX_JOB_ATTEMPT: u32 = i32::MAX as u32;
21
22fn scheduler_may_retry(kind: DiagnosticKind) -> bool {
23    matches!(
24        kind,
25        DiagnosticKind::RuntimeConflict
26            | DiagnosticKind::ResourceDeadline
27            | DiagnosticKind::ResourceCancelled
28    )
29}
30
31fn job_attempt_failure(
32    job_id: ObjectId,
33    metadata: &JobAttemptMetadata,
34    cause: Diagnostic,
35) -> Diagnostic {
36    if cause.kind() == DiagnosticKind::JobAttemptFailed {
37        return cause;
38    }
39    let cause_kind = cause.kind();
40    let cause_category = cause.category();
41    let mut failure = Diagnostic::new(
42        DiagnosticKind::JobAttemptFailed,
43        "scheduled job attempt failed",
44    )
45    .with_cause(cause_kind)
46    .with_detail("job_id", job_id.to_string())
47    .with_detail(
48        "scheduled_at_unix_ns",
49        metadata.scheduled_at_unix_ns.to_string(),
50    )
51    .with_detail("attempt", metadata.attempt.to_string())
52    .with_detail("cause_kind", cause_kind.as_str())
53    .with_detail("cause_category", cause_category.as_str())
54    .with_detail(
55        "scheduler_retryable",
56        scheduler_may_retry(cause_kind).to_string(),
57    )
58    .with_primary_span(cause.primary_span().cloned());
59    for secondary in cause.secondary_spans() {
60        failure = failure.with_secondary_span(secondary.label.clone(), secondary.span.clone());
61    }
62    for frame in cause.frames() {
63        failure = failure.with_frame(frame.clone());
64    }
65    // Security errors deliberately keep only the stable class. Their message
66    // may contain an object name that the attempt principal must not disclose
67    // through scheduler history or a remote status surface.
68    if cause_category != DiagnosticCategory::Security {
69        failure = failure.with_detail("cause_message", cause.message());
70        for detail in cause.details() {
71            failure = failure.with_detail(format!("cause.{}", detail.key), detail.value.clone());
72        }
73    }
74    failure
75}
76
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct JobAttemptMetadata {
79    pub scheduled_at_unix_ns: i64,
80    pub attempt: u32,
81    pub idempotency_key: String,
82}
83
84#[derive(Debug, Clone, PartialEq)]
85pub struct JobAttemptOutcome {
86    pub job_id: ObjectId,
87    pub scheduled_at_unix_ns: i64,
88    pub attempt: u32,
89    pub idempotency_key: String,
90    pub call: ProceduralCallOutcome,
91}
92
93/// Immutable scheduler input captured from one catalog generation.
94#[doc(hidden)]
95#[derive(Debug, Clone, PartialEq, Eq)]
96pub struct ScheduledJobDefinition {
97    pub job_id: ObjectId,
98    pub definition_version: u32,
99    pub schedule: ScheduledJobSchedule,
100}
101
102#[doc(hidden)]
103#[derive(Debug, Clone, Copy, PartialEq, Eq)]
104pub enum ScheduledJobSchedule {
105    EveryNs(u64),
106    AtUnixNs(i64),
107}
108
109pub(crate) fn bind_job_definition(
110    executor: &Executor,
111    statement: &CreateJobStatement,
112    catalog: &radixdb_catalog::CatalogGeneration,
113    context: &ExecutionContext,
114) -> Result<BoundJobDefinition> {
115    let principal_name = statement
116        .principal
117        .components
118        .last()
119        .ok_or_else(|| Error::invalid_argument("job principal name is empty"))?;
120    if statement.principal.components.len() != 1 {
121        return Err(Error::invalid_argument(
122            "principals are global and cannot be namespace-qualified",
123        ));
124    }
125    let principal_name = CatalogName::new(principal_name.value.as_str())
126        .map_err(|error| Error::invalid_argument(error.to_string()))?;
127    let principal = catalog
128        .objects_of_kind(ObjectKind::Principal)
129        .find(|object| object.name().normalized() == principal_name.normalized())
130        .ok_or_else(|| {
131            Error::invalid_argument(format!(
132                "job principal '{}' does not exist",
133                statement.principal
134            ))
135        })?;
136    let schedule = bind_schedule(&statement.schedule)?;
137    let (procedure_id, arguments) = bind_job_call(
138        executor,
139        catalog,
140        &statement.procedure,
141        &statement.arguments,
142        context,
143    )?;
144    let arguments = arguments
145        .into_iter()
146        .map(|(data_type, value)| {
147            let encoded = if value.is_null() {
148                None
149            } else {
150                Some(serialize_value(&value)?)
151            };
152            JobArgument::new(None, data_type, encoded)
153                .map_err(|error| Error::invalid_argument(error.to_string()))
154        })
155        .collect::<Result<Vec<_>>>()?;
156    Ok(BoundJobDefinition {
157        procedure_id,
158        principal_id: principal.id(),
159        schedule,
160        arguments,
161        resource_policy: ResourcePolicy::default_job(),
162    })
163}
164
165fn bind_schedule(schedule: &JobScheduleSyntax) -> Result<JobSchedule> {
166    match schedule {
167        JobScheduleSyntax::Every(Expression::IntervalLiteral(interval)) => {
168            if interval.quantity <= 0 {
169                return Err(Error::invalid_argument(
170                    "job EVERY interval must be greater than zero",
171                ));
172            }
173            let nanos_per_unit = match interval.unit.as_str() {
174                "second" => 1_000_000_000_u64,
175                "minute" => 60 * 1_000_000_000,
176                "hour" => 60 * 60 * 1_000_000_000,
177                "day" => 24 * 60 * 60 * 1_000_000_000,
178                "week" => 7 * 24 * 60 * 60 * 1_000_000_000,
179                "month" | "year" => {
180                    return Err(Error::invalid_argument(
181                        "job EVERY does not accept calendar MONTH/YEAR intervals",
182                    ))
183                }
184                _ => return Err(Error::invalid_argument("unknown job interval unit")),
185            };
186            let quantity = u64::try_from(interval.quantity)
187                .map_err(|_| Error::invalid_argument("job interval is outside u64"))?;
188            let value = quantity
189                .checked_mul(nanos_per_unit)
190                .ok_or_else(|| Error::invalid_argument("job interval nanoseconds overflow"))?;
191            Ok(JobSchedule::EveryNs(value))
192        }
193        JobScheduleSyntax::Every(_) => Err(Error::invalid_argument(
194            "job EVERY schedule must be one INTERVAL literal",
195        )),
196        JobScheduleSyntax::At(expression) => {
197            let Expression::StringLiteral(literal) = expression else {
198                return Err(Error::invalid_argument(
199                    "job AT schedule must be one TIMESTAMP literal",
200                ));
201            };
202            if !literal.type_hint.as_deref().is_some_and(|hint| {
203                matches!(
204                    hint.to_ascii_uppercase().as_str(),
205                    "TIMESTAMP" | "TIMESTAMPTZ"
206                )
207            }) {
208                return Err(Error::invalid_argument(
209                    "job AT schedule requires TIMESTAMP or TIMESTAMPTZ",
210                ));
211            }
212            let timestamp = radixdb_core::value::parse_timestamp(literal.value.as_str())?;
213            let nanos = timestamp.timestamp_nanos_opt().ok_or_else(|| {
214                Error::invalid_argument("job timestamp is outside nanosecond range")
215            })?;
216            Ok(JobSchedule::AtUnixNs(nanos))
217        }
218    }
219}
220
221fn decode_arguments(payload: &radixdb_catalog::JobPayload) -> Result<Vec<RuntimeValue>> {
222    payload
223        .arguments()
224        .iter()
225        .map(|argument| {
226            let value = match argument.value() {
227                Some(encoded) => deserialize_value(encoded)?,
228                None => Value::null(argument.data_type().logical_type()),
229            };
230            if value.data_type() != argument.data_type().logical_type() {
231                return Err(Error::invalid_argument(
232                    "job argument payload type differs from its catalog descriptor",
233                ));
234            }
235            Ok(RuntimeValue::scalar(value))
236        })
237        .collect()
238}
239
240impl Executor {
241    #[doc(hidden)]
242    pub fn scheduled_jobs_snapshot(&self) -> Result<Vec<ScheduledJobDefinition>> {
243        let catalog = self.engine.pin_catalog()?;
244        let mut jobs = catalog
245            .objects_of_kind(ObjectKind::Job)
246            .filter_map(|object| {
247                let CatalogPayload::Job(payload) = object.payload() else {
248                    return None;
249                };
250                payload.enabled().then_some(ScheduledJobDefinition {
251                    job_id: object.id(),
252                    definition_version: payload.definition_version(),
253                    schedule: match payload.schedule() {
254                        JobSchedule::EveryNs(interval) => ScheduledJobSchedule::EveryNs(interval),
255                        JobSchedule::AtUnixNs(timestamp) => {
256                            ScheduledJobSchedule::AtUnixNs(timestamp)
257                        }
258                    },
259                })
260            })
261            .collect::<Vec<_>>();
262        jobs.sort_by_key(|job| job.job_id);
263        Ok(jobs)
264    }
265
266    pub fn execute_job_attempt(
267        &self,
268        job_id: ObjectId,
269        metadata: JobAttemptMetadata,
270        context: &ExecutionContext,
271    ) -> std::result::Result<JobAttemptOutcome, Diagnostic> {
272        self.execute_job_attempt_inner(job_id, metadata.clone(), context)
273            .map_err(|cause| job_attempt_failure(job_id, &metadata, cause))
274    }
275
276    fn execute_job_attempt_inner(
277        &self,
278        job_id: ObjectId,
279        metadata: JobAttemptMetadata,
280        context: &ExecutionContext,
281    ) -> std::result::Result<JobAttemptOutcome, Diagnostic> {
282        if !self.ddl_fence_already_held {
283            let _fence = self.engine.acquire_ddl_statement_fence(false);
284            return self
285                .fork_with_owned_ddl_fence()
286                .execute_job_attempt_inner(job_id, metadata, context);
287        }
288        if metadata.attempt == 0 || metadata.attempt > MAX_JOB_ATTEMPT {
289            return Err(map_executor_error(Error::invalid_argument(
290                "job attempt must be inside 1..=2147483647",
291            )));
292        }
293        if metadata.idempotency_key.is_empty() || metadata.idempotency_key.len() > 1024 {
294            return Err(map_executor_error(Error::invalid_argument(
295                "job idempotency key must contain 1..=1024 bytes",
296            )));
297        }
298        if self.has_active_transaction() {
299            return Err(map_executor_error(Error::invalid_argument(
300                "job attempts require a fresh executor transaction",
301            )));
302        }
303        let mut stage = CallStatementStage::default();
304        stage.begin()?;
305        let boundary = match self.begin_procedural_boundary() {
306            Ok(boundary) => boundary,
307            Err(error) => {
308                stage.discard();
309                return Err(map_executor_error(error));
310            }
311        };
312        let (catalog, _) = match transaction_visible_catalog(self) {
313            Ok(value) => value,
314            Err(error) => {
315                stage.discard();
316                return Err(self.abort_after_setup_error(&boundary, map_executor_error(error)));
317            }
318        };
319        let job = catalog
320            .object(job_id)
321            .filter(|object| object.kind() == ObjectKind::Job)
322            .ok_or_else(|| Error::invalid_argument(format!("job object {job_id} does not exist")));
323        let job = match job {
324            Ok(job) => job,
325            Err(error) => {
326                stage.discard();
327                return Err(self.abort_after_setup_error(&boundary, map_executor_error(error)));
328            }
329        };
330        let CatalogPayload::Job(payload) = job.payload() else {
331            unreachable!("catalog kind/payload invariant")
332        };
333        if !payload.enabled() {
334            stage.discard();
335            return Err(self.abort_after_setup_error(
336                &boundary,
337                map_executor_error(Error::invalid_argument(format!("job {job_id} is disabled"))),
338            ));
339        }
340        let principal = catalog
341            .object(payload.principal_id())
342            .filter(|object| object.kind() == ObjectKind::Principal)
343            .ok_or_else(|| Error::invalid_argument("job principal disappeared from the catalog"));
344        match principal {
345            Ok(_) => {}
346            Err(error) => {
347                stage.discard();
348                return Err(self.abort_after_setup_error(&boundary, map_executor_error(error)));
349            }
350        }
351        let procedure_id = payload.procedure_id();
352        let arguments = match decode_arguments(payload) {
353            Ok(arguments) => arguments,
354            Err(error) => {
355                stage.discard();
356                return Err(self.abort_after_setup_error(&boundary, map_executor_error(error)));
357            }
358        };
359        let policy = payload.resource_policy();
360        let principal_id = payload.principal_id();
361        drop(catalog);
362
363        let scheduled_at = DateTime::<Utc>::from_timestamp_nanos(metadata.scheduled_at_unix_ns);
364        let mut job_context = context.clone();
365        job_context = job_context.with_principal_id(principal_id);
366        job_context.set_timeout_ms(match context.timeout_ms() {
367            0 => policy.deadline_ms,
368            caller => caller.min(policy.deadline_ms),
369        });
370        job_context.set_job_context(
371            &metadata.idempotency_key,
372            job_id,
373            metadata.attempt,
374            scheduled_at,
375        );
376
377        let call = self.execute_procedure_inside_boundary(
378            procedure_id,
379            arguments,
380            &job_context,
381            PrincipalContext {
382                session_principal: principal_id,
383                invoker_principal: principal_id,
384                effective_principal: principal_id,
385            },
386            &mut stage,
387            &boundary,
388        )?;
389        Ok(JobAttemptOutcome {
390            job_id,
391            scheduled_at_unix_ns: metadata.scheduled_at_unix_ns,
392            attempt: metadata.attempt,
393            idempotency_key: metadata.idempotency_key,
394            call,
395        })
396    }
397}
radixdb_executor/procedural/job.rs

radixdb_executor/procedural/
job.rs
