chronon_core/models/job.rs
1//! Job model - represents a scheduled job configuration.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7/// How a [`Job`] becomes due for enqueue.
8///
9/// | Variant | Due via tick? | Required fields | Typical trigger |
10/// |---------|---------------|-----------------|-----------------|
11/// | [`Self::Cron`] | Yes | `cron_expr` (+ optional `timezone`) | Recurring schedules |
12/// | [`Self::RunOnce`] | Yes | `next_run_at` or `run_once_at` | One-shot deferred work |
13/// | [`Self::Manual`] | **No** | — | `CoordinatorService::run_now` or HTTP `POST /jobs/run_now` |
14///
15/// Prefer Chronon `JobBuilder` (`chronon_scheduler::JobBuilder`, re-exported from the
16/// `chronon` / `uf-chronon` facade) to set schedule kind — `.cron` / `.run_once_at` /
17/// `.manual` — rather than assigning these variants by hand. Persist via coordinator upsert;
18/// cron next-fire is computed there when `schedule_kind == Cron`. See the `chronon` crate
19/// getting-started §5.
20///
21/// # Examples
22///
23/// Persisted shape after a builder (or low-level field assignment):
24///
25/// ```
26/// use chronon_core::ScheduleKind;
27///
28/// assert_eq!(ScheduleKind::Cron, ScheduleKind::default());
29/// assert_ne!(ScheduleKind::Manual, ScheduleKind::RunOnce);
30/// ```
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
32#[serde(rename_all = "snake_case")]
33pub enum ScheduleKind {
34 /// Recurring cron schedule — set [`Job::cron_expr`] and optional [`Job::timezone`].
35 #[default]
36 Cron,
37 /// One-time execution when [`Job::next_run_at`] / [`Job::run_once_at`] is due.
38 RunOnce,
39 /// Never due for the tick loop — enqueue only via `run_now` (in-process or HTTP).
40 Manual,
41}
42
43/// Maximum concurrent runs accepted for a single job (API / model clamp).
44pub const MAX_JOB_CONCURRENCY: i32 = 1_000;
45
46/// Maximum per-run timeout in milliseconds (24 hours).
47pub const MAX_TIMEOUT_MS: i64 = 86_400_000;
48
49/// Maximum additional retry attempts after the first execution.
50pub const MAX_RETRY_ATTEMPTS: u32 = 100;
51
52/// Maximum retry delay (base or cap) in milliseconds (24 hours).
53pub const MAX_RETRY_DELAY_MS: u64 = 86_400_000;
54
55/// Maximum page size for list APIs (jobs and runs).
56pub const MAX_LIST_LIMIT: usize = 1_000;
57
58/// Retry policy for failed or timed-out runs.
59///
60/// `max_attempts` is the number of **additional** retries after the first attempt
61/// (attempt 1). `0` means no retries.
62#[derive(Debug, Clone, Serialize, Deserialize)]
63pub struct RetryPolicy {
64 /// Maximum number of retry attempts after the first execution.
65 pub max_attempts: u32,
66
67 /// Base delay between retries in milliseconds.
68 pub base_delay_ms: u64,
69
70 /// Exponential backoff multiplier (`1.0` = no backoff).
71 pub backoff_multiplier: f64,
72
73 /// Cap on delay in milliseconds; `0` means uncapped.
74 pub max_delay_ms: u64,
75}
76
77impl Default for RetryPolicy {
78 fn default() -> Self {
79 Self {
80 max_attempts: 0,
81 base_delay_ms: 0,
82 backoff_multiplier: 1.0,
83 max_delay_ms: 0,
84 }
85 }
86}
87
88impl RetryPolicy {
89 /// Whether a failed attempt with this `attempt` number should schedule another run.
90 pub fn should_retry(&self, attempt: i32) -> bool {
91 attempt > 0 && (attempt as u32) <= self.max_attempts
92 }
93
94 /// Delay before the next attempt after `failed_attempt` finishes.
95 pub fn delay_ms_after(&self, failed_attempt: i32) -> u64 {
96 let exp = failed_attempt.saturating_sub(1).max(0);
97 let raw = (self.base_delay_ms as f64) * self.backoff_multiplier.powi(exp);
98 let ms = if raw.is_finite() && raw > 0.0 {
99 raw.min(u64::MAX as f64) as u64
100 } else {
101 0
102 };
103 if self.max_delay_ms == 0 {
104 ms
105 } else {
106 ms.min(self.max_delay_ms)
107 }
108 }
109
110 /// Clamp retry knobs to production-safe ceilings ([`MAX_RETRY_ATTEMPTS`], [`MAX_RETRY_DELAY_MS`]).
111 #[must_use]
112 pub fn clamp_security_bounds(mut self) -> Self {
113 self.max_attempts = self.max_attempts.min(MAX_RETRY_ATTEMPTS);
114 self.base_delay_ms = self.base_delay_ms.min(MAX_RETRY_DELAY_MS);
115 if self.max_delay_ms != 0 {
116 self.max_delay_ms = self.max_delay_ms.min(MAX_RETRY_DELAY_MS);
117 }
118 if !self.backoff_multiplier.is_finite() || self.backoff_multiplier < 1.0 {
119 self.backoff_multiplier = 1.0;
120 }
121 self
122 }
123}
124
125/// Policy for handling missed scheduled fires at tick time.
126///
127/// When `max_misfire_window_secs == 0`, misfire gating is disabled (legacy: always enqueue).
128#[derive(Debug, Clone, Default, Serialize, Deserialize)]
129pub struct MisfirePolicy {
130 /// When true and within the misfire window, enqueue one coalesced run for the miss.
131 pub run_immediately: bool,
132
133 /// Max lateness (seconds) that still qualifies for misfire recovery; `0` disables gating.
134 pub max_misfire_window_secs: u64,
135}
136
137/// Scheduled work unit: binds a **script name** to a [`ScheduleKind`] and params.
138///
139/// **Preferred construction:** Chronon `JobBuilder` (`chronon_scheduler::JobBuilder`,
140/// re-exported from the `chronon` / `uf-chronon` facade) from a [`crate::ScriptHandle`],
141/// then persist with `CoordinatorService::upsert_job` (or HTTP upsert /
142/// `RemoteCoordinatorClient`).
143///
144/// [`Job::new`] / [`crate::ScriptHandle::job`] / [`crate::ScriptHandle::job_with_params`]
145/// only seed baseline rows — a low-level alternate to the fluent builder.
146///
147/// | Field group | Purpose |
148/// |-------------|---------|
149/// | `script_name` / `params_json` | What to run and with which args |
150/// | `schedule_kind` + cron / run-once fields | When it becomes due |
151/// | `actor_json` | Identity snapshotted onto each run at enqueue; dispatch uses the **run** snapshot |
152/// | `pool` / placement | Worker pool targeting in split deployments |
153///
154/// # Examples
155///
156/// Baseline row (no schedule yet). Prefer `JobBuilder` for cron / run-once / manual:
157///
158/// ```
159/// use chronon_core::Job;
160///
161/// let job = Job::new("nightly-cleanup", "nightly_cleanup");
162/// assert!(job.enabled);
163/// assert_eq!(job.script_name, "nightly_cleanup");
164/// assert_eq!(job.schedule_kind, chronon_core::ScheduleKind::Cron);
165/// ```
166///
167/// Runnable: `cargo run -p uf-chronon --example script_handle_job --features mem`.
168#[derive(Debug, Clone, Serialize, Deserialize)]
169pub struct Job {
170 /// Unique identifier (UUID).
171 pub job_id: String,
172
173 /// Human-readable name, unique in the deployment.
174 pub job_name: String,
175
176 /// Name of the script to execute.
177 pub script_name: String,
178
179 /// Signature hash at job creation (for validation).
180 pub script_sig_hash: String,
181
182 /// Whether the job is enabled.
183 pub enabled: bool,
184
185 /// How the job is scheduled.
186 pub schedule_kind: ScheduleKind,
187
188 /// Cron expression (when schedule_kind is Cron).
189 pub cron_expr: Option<String>,
190
191 /// Timezone for cron evaluation (e.g., "America/New_York").
192 pub timezone: Option<String>,
193
194 /// One-time execution timestamp (when schedule_kind is RunOnce).
195 pub run_once_at: Option<DateTime<Utc>>,
196
197 /// When a coordinator claimed this run-once job for enqueue (distributed safety).
198 pub run_once_claimed_at: Option<DateTime<Utc>>,
199
200 /// Coordinator instance id that holds the claim (`coordinator_instance_id`).
201 pub run_once_claimed_by: Option<String>,
202
203 /// Set after a scheduled run-once execution is successfully enqueued (persisted run row).
204 pub run_once_completed_at: Option<DateTime<Utc>>,
205
206 /// Claim lease expiry; after this, another coordinator may reclaim if not completed.
207 pub run_once_claim_expires_at: Option<DateTime<Utc>>,
208
209 /// Partition hash for coordinator sharding (distributed mode).
210 pub partition_hash: Option<i64>,
211
212 /// Coordinator tick-claim holder id.
213 pub claim_lease_id: Option<String>,
214
215 /// Coordinator tick-claim lease expiry.
216 pub claim_lease_until: Option<DateTime<Utc>>,
217
218 // Distributed-mode fields (stored but ignored in local mode)
219 /// Execution pool (e.g., "global", "region/us-west").
220 pub pool: Option<String>,
221
222 /// Target region for execution.
223 pub region: Option<String>,
224
225 /// Additional placement constraints as JSON.
226 pub placement_json: Option<Value>,
227
228 // Identity for permission reconstruction
229 /// Serialized actor identity. Copied onto each run at enqueue; workers rebuild context
230 /// from the **run** snapshot so later job updates cannot change queued identity.
231 pub actor_json: Value,
232
233 /// Parameters to pass to the script.
234 pub params_json: Value,
235
236 /// Maximum concurrent runs allowed.
237 pub concurrency: i32,
238
239 /// Execution timeout in milliseconds.
240 pub timeout_ms: Option<i64>,
241
242 /// Retry policy JSON ([`RetryPolicy`]).
243 pub retry_policy_json: Value,
244
245 /// Misfire policy JSON ([`MisfirePolicy`]).
246 pub misfire_policy_json: Value,
247
248 /// Limits for parent/child runs.
249 pub parent_limits_json: Option<Value>,
250
251 /// When the job should next run.
252 pub next_run_at: Option<DateTime<Utc>>,
253
254 /// Current revision number.
255 pub current_revision: i32,
256
257 /// Last modification timestamp.
258 pub updated_at: DateTime<Utc>,
259
260 /// Creation timestamp.
261 pub created_at: DateTime<Utc>,
262}
263
264impl Job {
265 /// Baseline job with generated `job_id`, `enabled = true`, and
266 /// [`ScheduleKind::Cron`] (no expression yet).
267 ///
268 /// Populate schedule fields via Chronon `JobBuilder` (`chronon_scheduler::JobBuilder`,
269 /// preferred), or set `schedule_kind`, cron / run-once fields, `params_json`, and
270 /// `actor_json` before upsert. [`crate::ScriptHandle::job`] /
271 /// [`crate::ScriptHandle::job_with_params`] only seed baseline rows.
272 ///
273 /// # Examples
274 ///
275 /// ```
276 /// use chronon_core::Job;
277 ///
278 /// let job = Job::new("demo", "noop");
279 /// assert!(!job.job_id.is_empty());
280 /// assert_eq!(job.job_name, "demo");
281 /// assert_eq!(job.script_name, "noop");
282 /// ```
283 pub fn new(job_name: impl Into<String>, script_name: impl Into<String>) -> Self {
284 let now = Utc::now();
285 Self {
286 job_id: uuid::Uuid::new_v4().to_string(),
287 job_name: job_name.into(),
288 script_name: script_name.into(),
289 script_sig_hash: String::new(),
290 enabled: true,
291 schedule_kind: ScheduleKind::default(),
292 cron_expr: None,
293 timezone: None,
294 run_once_at: None,
295 run_once_claimed_at: None,
296 run_once_claimed_by: None,
297 run_once_completed_at: None,
298 run_once_claim_expires_at: None,
299 partition_hash: None,
300 claim_lease_id: None,
301 claim_lease_until: None,
302 pool: None,
303 region: None,
304 placement_json: None,
305 actor_json: Value::Null,
306 params_json: Value::Object(serde_json::Map::default()),
307 concurrency: 1,
308 timeout_ms: None,
309 retry_policy_json: serde_json::to_value(RetryPolicy::default()).unwrap_or_default(),
310 misfire_policy_json: serde_json::to_value(MisfirePolicy::default()).unwrap_or_default(),
311 parent_limits_json: None,
312 next_run_at: None,
313 current_revision: 1,
314 updated_at: now,
315 created_at: now,
316 }
317 }
318
319 /// Decode [`RetryPolicy`] from [`Self::retry_policy_json`], or default on null/invalid.
320 pub fn retry_policy(&self) -> RetryPolicy {
321 serde_json::from_value(self.retry_policy_json.clone()).unwrap_or_default()
322 }
323
324 /// Decode [`MisfirePolicy`] from [`Self::misfire_policy_json`], or default on null/invalid.
325 pub fn misfire_policy(&self) -> MisfirePolicy {
326 serde_json::from_value(self.misfire_policy_json.clone()).unwrap_or_default()
327 }
328
329 /// Persist a typed retry policy into [`Self::retry_policy_json`].
330 pub fn set_retry_policy(&mut self, policy: &RetryPolicy) {
331 self.retry_policy_json = serde_json::to_value(policy).unwrap_or_default();
332 }
333
334 /// Persist a typed misfire policy into [`Self::misfire_policy_json`].
335 pub fn set_misfire_policy(&mut self, policy: &MisfirePolicy) {
336 self.misfire_policy_json = serde_json::to_value(policy).unwrap_or_default();
337 }
338
339 /// Clamp concurrency, timeout, and retry policy to production-safe ceilings.
340 ///
341 /// Applied by the HTTP upsert path and available to hosts before persist.
342 pub fn clamp_security_bounds(&mut self) {
343 self.concurrency = self.concurrency.clamp(1, MAX_JOB_CONCURRENCY);
344 if let Some(ms) = self.timeout_ms {
345 if ms <= 0 {
346 self.timeout_ms = None;
347 } else {
348 self.timeout_ms = Some(ms.min(MAX_TIMEOUT_MS));
349 }
350 }
351 let clamped = self.retry_policy().clamp_security_bounds();
352 self.set_retry_policy(&clamped);
353 }
354}
355
356#[cfg(test)]
357mod policy_tests {
358 use super::*;
359
360 #[test]
361 fn retry_default_does_not_retry() {
362 let p = RetryPolicy::default();
363 assert!(!p.should_retry(1));
364 assert_eq!(p.delay_ms_after(1), 0);
365 }
366
367 #[test]
368 fn retry_backoff_and_cap() {
369 let p = RetryPolicy {
370 max_attempts: 3,
371 base_delay_ms: 100,
372 backoff_multiplier: 2.0,
373 max_delay_ms: 250,
374 };
375 assert!(p.should_retry(1));
376 assert!(p.should_retry(3));
377 assert!(!p.should_retry(4));
378 assert_eq!(p.delay_ms_after(1), 100);
379 assert_eq!(p.delay_ms_after(2), 200);
380 assert_eq!(p.delay_ms_after(3), 250);
381 }
382
383 #[test]
384 fn job_policy_roundtrip() {
385 let mut job = Job::new("n", "s");
386 let retry = RetryPolicy {
387 max_attempts: 2,
388 base_delay_ms: 50,
389 backoff_multiplier: 1.5,
390 max_delay_ms: 500,
391 };
392 let misfire = MisfirePolicy {
393 run_immediately: true,
394 max_misfire_window_secs: 3600,
395 };
396 job.set_retry_policy(&retry);
397 job.set_misfire_policy(&misfire);
398 assert_eq!(job.retry_policy().max_attempts, 2);
399 assert!(job.misfire_policy().run_immediately);
400 assert_eq!(job.misfire_policy().max_misfire_window_secs, 3600);
401 }
402
403 #[test]
404 fn clamp_security_bounds_caps_extremes() {
405 let mut job = Job::new("n", "s");
406 job.concurrency = i32::MAX;
407 job.timeout_ms = Some(i64::MAX);
408 job.set_retry_policy(&RetryPolicy {
409 max_attempts: MAX_RETRY_ATTEMPTS.saturating_mul(10),
410 base_delay_ms: MAX_RETRY_DELAY_MS.saturating_mul(2),
411 backoff_multiplier: 0.5,
412 max_delay_ms: MAX_RETRY_DELAY_MS.saturating_mul(2),
413 });
414 job.clamp_security_bounds();
415 assert_eq!(job.concurrency, MAX_JOB_CONCURRENCY);
416 assert_eq!(job.timeout_ms, Some(MAX_TIMEOUT_MS));
417 let retry = job.retry_policy();
418 assert_eq!(retry.max_attempts, MAX_RETRY_ATTEMPTS);
419 assert_eq!(retry.base_delay_ms, MAX_RETRY_DELAY_MS);
420 assert_eq!(retry.max_delay_ms, MAX_RETRY_DELAY_MS);
421 assert!((retry.backoff_multiplier - 1.0).abs() < f64::EPSILON);
422 }
423
424 #[test]
425 fn clamp_security_bounds_clears_non_positive_timeout() {
426 let mut job = Job::new("n", "s");
427 job.timeout_ms = Some(0);
428 job.clamp_security_bounds();
429 assert_eq!(job.timeout_ms, None);
430 }
431
432 #[test]
433 fn clamp_security_bounds_on_raw_extreme_json() {
434 let mut job = Job::new("n", "s");
435 job.retry_policy_json = serde_json::json!({
436 "max_attempts": 9_999_999_u64,
437 "base_delay_ms": 9_999_999_999_u64,
438 "backoff_multiplier": 2.0,
439 "max_delay_ms": 9_999_999_999_u64
440 });
441 job.clamp_security_bounds();
442 let retry = job.retry_policy();
443 assert_eq!(retry.max_attempts, MAX_RETRY_ATTEMPTS);
444 assert_eq!(retry.base_delay_ms, MAX_RETRY_DELAY_MS);
445 assert_eq!(retry.max_delay_ms, MAX_RETRY_DELAY_MS);
446 }
447}