roboticus_db/
cron.rs

1use rusqlite::OptionalExtension;
2
3use crate::{Database, DbResultExt};
4use roboticus_core::Result;
5
6#[derive(Debug, Clone)]
7pub struct CronJob {
8    pub id: String,
9    pub name: String,
10    pub description: Option<String>,
11    pub enabled: bool,
12    pub schedule_kind: String,
13    pub schedule_expr: Option<String>,
14    pub schedule_every_ms: Option<i64>,
15    pub schedule_tz: Option<String>,
16    pub agent_id: String,
17    pub session_target: String,
18    pub payload_json: String,
19    pub delivery_mode: Option<String>,
20    pub delivery_channel: Option<String>,
21    pub last_run_at: Option<String>,
22    pub last_status: Option<String>,
23    pub last_duration_ms: Option<i64>,
24    pub consecutive_errors: i64,
25    pub next_run_at: Option<String>,
26    pub last_error: Option<String>,
27    pub lease_holder: Option<String>,
28    pub lease_expires_at: Option<String>,
29}
30
31#[derive(Debug, Clone)]
32pub struct CronRun {
33    pub id: String,
34    pub job_id: String,
35    pub status: String,
36    pub duration_ms: Option<i64>,
37    pub error: Option<String>,
38    pub output_text: Option<String>,
39    pub created_at: String,
40}
41
42pub fn create_job(
43    db: &Database,
44    name: &str,
45    agent_id: &str,
46    schedule_kind: &str,
47    schedule_expr: Option<&str>,
48    payload_json: &str,
49) -> Result<String> {
50    let conn = db.conn();
51    let id = uuid::Uuid::new_v4().to_string();
52    conn.execute(
53        "INSERT INTO cron_jobs (id, name, agent_id, schedule_kind, schedule_expr, payload_json) \
54         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
55        rusqlite::params![
56            id,
57            name,
58            agent_id,
59            schedule_kind,
60            schedule_expr,
61            payload_json
62        ],
63    )
64    .db_err()?;
65    Ok(id)
66}
67
68pub fn list_jobs(db: &Database) -> Result<Vec<CronJob>> {
69    let conn = db.conn();
70    let mut stmt = conn
71        .prepare(
72            "SELECT id, name, description, enabled, schedule_kind, schedule_expr, \
73             schedule_every_ms, schedule_tz, agent_id, session_target, payload_json, \
74             delivery_mode, delivery_channel, last_run_at, last_status, last_duration_ms, \
75             consecutive_errors, next_run_at, last_error, lease_holder, lease_expires_at \
76             FROM cron_jobs ORDER BY name ASC",
77        )
78        .db_err()?;
79
80    let rows = stmt
81        .query_map([], |row| {
82            Ok(CronJob {
83                id: row.get(0)?,
84                name: row.get(1)?,
85                description: row.get(2)?,
86                enabled: row.get::<_, i32>(3)? != 0,
87                schedule_kind: row.get(4)?,
88                schedule_expr: row.get(5)?,
89                schedule_every_ms: row.get(6)?,
90                schedule_tz: row.get(7)?,
91                agent_id: row.get(8)?,
92                session_target: row.get(9)?,
93                payload_json: row.get(10)?,
94                delivery_mode: row.get(11)?,
95                delivery_channel: row.get(12)?,
96                last_run_at: row.get(13)?,
97                last_status: row.get(14)?,
98                last_duration_ms: row.get(15)?,
99                consecutive_errors: row.get(16)?,
100                next_run_at: row.get(17)?,
101                last_error: row.get(18)?,
102                lease_holder: row.get(19)?,
103                lease_expires_at: row.get(20)?,
104            })
105        })
106        .db_err()?;
107
108    rows.collect::<std::result::Result<Vec<_>, _>>().db_err()
109}
110
111pub fn get_job(db: &Database, id: &str) -> Result<Option<CronJob>> {
112    let conn = db.conn();
113    conn.query_row(
114        "SELECT id, name, description, enabled, schedule_kind, schedule_expr, \
115         schedule_every_ms, schedule_tz, agent_id, session_target, payload_json, \
116         delivery_mode, delivery_channel, last_run_at, last_status, last_duration_ms, \
117         consecutive_errors, next_run_at, last_error, lease_holder, lease_expires_at \
118         FROM cron_jobs WHERE id = ?1",
119        [id],
120        |row| {
121            Ok(CronJob {
122                id: row.get(0)?,
123                name: row.get(1)?,
124                description: row.get(2)?,
125                enabled: row.get::<_, i32>(3)? != 0,
126                schedule_kind: row.get(4)?,
127                schedule_expr: row.get(5)?,
128                schedule_every_ms: row.get(6)?,
129                schedule_tz: row.get(7)?,
130                agent_id: row.get(8)?,
131                session_target: row.get(9)?,
132                payload_json: row.get(10)?,
133                delivery_mode: row.get(11)?,
134                delivery_channel: row.get(12)?,
135                last_run_at: row.get(13)?,
136                last_status: row.get(14)?,
137                last_duration_ms: row.get(15)?,
138                consecutive_errors: row.get(16)?,
139                next_run_at: row.get(17)?,
140                last_error: row.get(18)?,
141                lease_holder: row.get(19)?,
142                lease_expires_at: row.get(20)?,
143            })
144        },
145    )
146    .optional()
147    .db_err()
148}
149
150pub fn delete_job(db: &Database, id: &str) -> Result<bool> {
151    fn quote_ident(s: &str) -> String {
152        format!("\"{}\"", s.replace('"', "\"\""))
153    }
154
155    let conn = db.conn();
156    let tx = conn.unchecked_transaction().db_err()?;
157
158    // Delete dependent rows from every table that has an FK to cron_jobs.
159    // This keeps delete robust across schema versions/custom local tables.
160    let mut table_stmt = tx
161        .prepare(
162            "SELECT name FROM sqlite_master \
163             WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
164        )
165        .db_err()?;
166    let table_names = table_stmt
167        .query_map([], |row| row.get::<_, String>(0))
168        .db_err()?
169        .collect::<std::result::Result<Vec<_>, _>>()
170        .db_err()?;
171    drop(table_stmt);
172
173    for table in table_names {
174        let pragma_sql = format!("PRAGMA foreign_key_list({})", quote_ident(&table));
175        let mut fk_stmt = tx.prepare(&pragma_sql).db_err()?;
176        let fk_cols = fk_stmt
177            .query_map([], |row| {
178                let ref_table: String = row.get(2)?;
179                let from_col: String = row.get(3)?;
180                Ok((ref_table, from_col))
181            })
182            .db_err()?
183            .collect::<std::result::Result<Vec<_>, _>>()
184            .db_err()?;
185        drop(fk_stmt);
186
187        for (ref_table, from_col) in fk_cols {
188            if ref_table == "cron_jobs" {
189                let delete_sql = format!(
190                    "DELETE FROM {} WHERE {} = ?1",
191                    quote_ident(&table),
192                    quote_ident(&from_col)
193                );
194                tx.execute(&delete_sql, [id]).db_err()?;
195            }
196        }
197    }
198
199    let changed = tx
200        .execute("DELETE FROM cron_jobs WHERE id = ?1", [id])
201        .db_err()?;
202    tx.commit().db_err()?;
203    Ok(changed > 0)
204}
205
206pub fn update_job(
207    db: &Database,
208    id: &str,
209    name: Option<&str>,
210    schedule_kind: Option<&str>,
211    schedule_expr: Option<&str>,
212    enabled: Option<bool>,
213) -> Result<bool> {
214    let conn = db.conn();
215    let mut sets = Vec::new();
216    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
217
218    if let Some(v) = name {
219        sets.push("name = ?");
220        params.push(Box::new(v.to_string()));
221    }
222    if let Some(v) = schedule_kind {
223        sets.push("schedule_kind = ?");
224        params.push(Box::new(v.to_string()));
225    }
226    if let Some(v) = schedule_expr {
227        sets.push("schedule_expr = ?");
228        params.push(Box::new(v.to_string()));
229    }
230    if let Some(v) = enabled {
231        sets.push("enabled = ?");
232        params.push(Box::new(v as i32));
233    }
234
235    if sets.is_empty() {
236        return Ok(false);
237    }
238
239    // Renumber placeholders: ?1, ?2, ... ?N, id = ?N+1
240    let numbered: Vec<String> = sets
241        .iter()
242        .enumerate()
243        .map(|(i, s)| s.replace('?', &format!("?{}", i + 1)))
244        .collect();
245    let id_param = params.len() + 1;
246    let sql = format!(
247        "UPDATE cron_jobs SET {} WHERE id = ?{id_param}",
248        numbered.join(", ")
249    );
250    params.push(Box::new(id.to_string()));
251
252    let param_refs: Vec<&dyn rusqlite::types::ToSql> = params.iter().map(|b| b.as_ref()).collect();
253    let changed = conn.execute(&sql, param_refs.as_slice()).db_err()?;
254    Ok(changed > 0)
255}
256
257pub fn update_job_description(db: &Database, id: &str, description: Option<&str>) -> Result<bool> {
258    let conn = db.conn();
259    let changed = conn
260        .execute(
261            "UPDATE cron_jobs SET description = ?1 WHERE id = ?2",
262            rusqlite::params![description, id],
263        )
264        .db_err()?;
265    Ok(changed > 0)
266}
267
268/// Attempts to acquire a 60-second lease for `instance_id` on the given job.
269/// Returns `true` if the lease was acquired (no existing valid lease or expired).
270pub fn acquire_lease(db: &Database, job_id: &str, instance_id: &str) -> Result<bool> {
271    let conn = db.conn();
272    let changed = conn
273        .execute(
274            "UPDATE cron_jobs SET lease_holder = ?1, lease_expires_at = datetime('now', '+60 seconds') \
275             WHERE id = ?2 AND (lease_holder IS NULL OR lease_expires_at < datetime('now'))",
276            rusqlite::params![instance_id, job_id],
277        )
278        .db_err()?;
279    Ok(changed > 0)
280}
281
282pub fn release_lease(db: &Database, job_id: &str, lease_holder: &str) -> Result<()> {
283    let conn = db.conn();
284    conn.execute(
285        "UPDATE cron_jobs SET lease_holder = NULL, lease_expires_at = NULL \
286         WHERE id = ?1 AND lease_holder = ?2",
287        rusqlite::params![job_id, lease_holder],
288    )
289    .db_err()?;
290    Ok(())
291}
292
293/// Update the `next_run_at` field for a cron job.
294pub fn update_next_run_at(db: &Database, job_id: &str, next_run_at: Option<&str>) -> Result<()> {
295    let conn = db.conn();
296    conn.execute(
297        "UPDATE cron_jobs SET next_run_at = ?1 WHERE id = ?2",
298        rusqlite::params![next_run_at, job_id],
299    )
300    .db_err()?;
301    Ok(())
302}
303
304pub fn record_run(
305    db: &Database,
306    job_id: &str,
307    status: &str,
308    duration_ms: Option<i64>,
309    error: Option<&str>,
310    output_text: Option<&str>,
311) -> Result<String> {
312    let conn = db.conn();
313    let tx = conn.unchecked_transaction().db_err()?;
314
315    let id = uuid::Uuid::new_v4().to_string();
316    tx.execute(
317        "INSERT INTO cron_runs (id, job_id, status, duration_ms, error, output_text) \
318         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
319        rusqlite::params![id, job_id, status, duration_ms, error, output_text],
320    )
321    .db_err()?;
322
323    if status == "success" {
324        tx.execute(
325            "UPDATE cron_jobs SET last_run_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'), last_status = ?1, \
326             last_duration_ms = ?2, consecutive_errors = 0, last_error = NULL WHERE id = ?3",
327            rusqlite::params![status, duration_ms, job_id],
328        )
329        .db_err()?;
330    } else {
331        tx.execute(
332            "UPDATE cron_jobs SET last_run_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'), last_status = ?1, \
333             last_duration_ms = ?2, consecutive_errors = consecutive_errors + 1, \
334             last_error = ?3 WHERE id = ?4",
335            rusqlite::params![status, duration_ms, error, job_id],
336        )
337        .db_err()?;
338    }
339
340    tx.commit().db_err()?;
341    Ok(id)
342}
343
344pub fn list_runs(
345    db: &Database,
346    from: Option<&str>,
347    to: Option<&str>,
348    job_id: Option<&str>,
349    limit: i64,
350) -> Result<Vec<CronRun>> {
351    let conn = db.conn();
352    let sql = "SELECT id, job_id, status, duration_ms, error, output_text, created_at
353               FROM cron_runs
354               WHERE (?1 IS NULL OR created_at >= ?1)
355                 AND (?2 IS NULL OR created_at <= ?2)
356                 AND (?3 IS NULL OR job_id = ?3)
357               ORDER BY created_at DESC
358               LIMIT ?4";
359    let mut stmt = conn.prepare(sql).db_err()?;
360    let rows = stmt
361        .query_map(rusqlite::params![from, to, job_id, limit], |row| {
362            Ok(CronRun {
363                id: row.get(0)?,
364                job_id: row.get(1)?,
365                status: row.get(2)?,
366                duration_ms: row.get(3)?,
367                error: row.get(4)?,
368                output_text: row.get(5)?,
369                created_at: row.get(6)?,
370            })
371        })
372        .db_err()?;
373    rows.collect::<std::result::Result<Vec<_>, _>>().db_err()
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379
380    fn test_db() -> Database {
381        Database::new(":memory:").unwrap()
382    }
383
384    #[test]
385    fn create_and_list_jobs() {
386        let db = test_db();
387        create_job(
388            &db,
389            "heartbeat",
390            "agent-1",
391            "every",
392            None,
393            r#"{"action":"ping"}"#,
394        )
395        .unwrap();
396        create_job(
397            &db,
398            "daily-report",
399            "agent-1",
400            "cron",
401            Some("0 9 * * *"),
402            r#"{"action":"report"}"#,
403        )
404        .unwrap();
405
406        let jobs = list_jobs(&db).unwrap();
407        assert_eq!(jobs.len(), 2);
408        assert_eq!(jobs[0].name, "daily-report");
409        assert_eq!(jobs[1].name, "heartbeat");
410    }
411
412    #[test]
413    fn lease_acquisition_and_release() {
414        let db = test_db();
415        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
416
417        assert!(acquire_lease(&db, &job_id, "instance-1").unwrap());
418        // Second acquire by a different instance should fail (lease not expired)
419        assert!(!acquire_lease(&db, &job_id, "instance-2").unwrap());
420
421        release_lease(&db, &job_id, "instance-1").unwrap();
422        assert!(acquire_lease(&db, &job_id, "instance-2").unwrap());
423    }
424
425    #[test]
426    fn get_and_delete_job() {
427        let db = test_db();
428        let job_id = create_job(&db, "to-delete", "a1", "every", None, "{}").unwrap();
429
430        let job = get_job(&db, &job_id).unwrap().expect("job should exist");
431        assert_eq!(job.name, "to-delete");
432        assert_eq!(job.agent_id, "a1");
433
434        assert!(delete_job(&db, &job_id).unwrap());
435        assert!(get_job(&db, &job_id).unwrap().is_none());
436        assert!(!delete_job(&db, &job_id).unwrap());
437    }
438
439    #[test]
440    fn record_run_updates_job() {
441        let db = test_db();
442        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
443
444        record_run(&db, &job_id, "success", Some(150), None, None).unwrap();
445        let jobs = list_jobs(&db).unwrap();
446        assert_eq!(jobs[0].last_status.as_deref(), Some("success"));
447        assert_eq!(jobs[0].consecutive_errors, 0);
448
449        record_run(&db, &job_id, "error", Some(50), Some("timeout"), None).unwrap();
450        let jobs = list_jobs(&db).unwrap();
451        assert_eq!(jobs[0].consecutive_errors, 1);
452        assert_eq!(jobs[0].last_error.as_deref(), Some("timeout"));
453    }
454
455    #[test]
456    fn get_job_nonexistent_returns_none() {
457        let db = test_db();
458        assert!(get_job(&db, "does-not-exist").unwrap().is_none());
459    }
460
461    #[test]
462    fn list_jobs_empty_db() {
463        let db = test_db();
464        let jobs = list_jobs(&db).unwrap();
465        assert!(jobs.is_empty());
466    }
467
468    #[test]
469    fn create_job_defaults() {
470        let db = test_db();
471        let id = create_job(&db, "j1", "a1", "every", None, "{}").unwrap();
472        let job = get_job(&db, &id).unwrap().unwrap();
473        assert!(job.enabled);
474        assert!(job.last_run_at.is_none());
475        assert!(job.last_status.is_none());
476        assert_eq!(job.consecutive_errors, 0);
477        assert!(job.last_error.is_none());
478        assert!(job.lease_holder.is_none());
479    }
480
481    #[test]
482    fn create_job_with_schedule_expr() {
483        let db = test_db();
484        let id = create_job(
485            &db,
486            "cron-job",
487            "a1",
488            "cron",
489            Some("0 */5 * * *"),
490            r#"{"a":1}"#,
491        )
492        .unwrap();
493        let job = get_job(&db, &id).unwrap().unwrap();
494        assert_eq!(job.schedule_kind, "cron");
495        assert_eq!(job.schedule_expr.as_deref(), Some("0 */5 * * *"));
496    }
497
498    #[test]
499    fn record_run_success_clears_last_error() {
500        let db = test_db();
501        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
502
503        record_run(&db, &job_id, "error", Some(10), Some("oops"), None).unwrap();
504        let job = get_job(&db, &job_id).unwrap().unwrap();
505        assert_eq!(job.consecutive_errors, 1);
506        assert_eq!(job.last_error.as_deref(), Some("oops"));
507
508        record_run(&db, &job_id, "success", Some(20), None, None).unwrap();
509        let job = get_job(&db, &job_id).unwrap().unwrap();
510        assert_eq!(job.consecutive_errors, 0);
511        assert!(job.last_error.is_none());
512    }
513
514    #[test]
515    fn record_run_with_none_duration() {
516        let db = test_db();
517        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
518        let run_id = record_run(&db, &job_id, "error", None, Some("crash"), None).unwrap();
519        assert!(!run_id.is_empty());
520        let job = get_job(&db, &job_id).unwrap().unwrap();
521        assert!(job.last_duration_ms.is_none());
522    }
523
524    #[test]
525    fn consecutive_errors_compound() {
526        let db = test_db();
527        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
528        for i in 1..=5 {
529            record_run(
530                &db,
531                &job_id,
532                "error",
533                Some(10),
534                Some(&format!("err-{i}")),
535                None,
536            )
537            .unwrap();
538            let job = get_job(&db, &job_id).unwrap().unwrap();
539            assert_eq!(job.consecutive_errors, i);
540        }
541    }
542
543    #[test]
544    fn acquire_lease_nonexistent_job() {
545        let db = test_db();
546        let acquired = acquire_lease(&db, "no-such-job", "inst-1").unwrap();
547        assert!(!acquired);
548    }
549
550    #[test]
551    fn release_lease_nonexistent_job() {
552        let db = test_db();
553        release_lease(&db, "no-such-job", "inst-1").unwrap();
554    }
555
556    #[test]
557    fn record_run_persists_output_text() {
558        let db = test_db();
559        let job_id = create_job(
560            &db,
561            "job-output",
562            "agent-1",
563            "cron",
564            Some("0 * * * *"),
565            "{}",
566        )
567        .unwrap();
568        let run_id =
569            record_run(&db, &job_id, "success", Some(5), None, Some("hello world")).unwrap();
570        let runs = list_runs(&db, None, None, Some(&job_id), 10).unwrap();
571        assert_eq!(runs[0].id, run_id);
572        assert_eq!(runs[0].output_text.as_deref(), Some("hello world"));
573    }
574
575    #[test]
576    fn record_run_returns_unique_ids() {
577        let db = test_db();
578        let job_id = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
579        let r1 = record_run(&db, &job_id, "success", Some(10), None, None).unwrap();
580        let r2 = record_run(&db, &job_id, "success", Some(20), None, None).unwrap();
581        assert_ne!(r1, r2);
582    }
583
584    #[test]
585    fn update_job_name_only() {
586        let db = test_db();
587        let id = create_job(&db, "old-name", "a1", "every", None, "{}").unwrap();
588        let changed = update_job(&db, &id, Some("new-name"), None, None, None).unwrap();
589        assert!(changed);
590        let job = get_job(&db, &id).unwrap().unwrap();
591        assert_eq!(job.name, "new-name");
592    }
593
594    #[test]
595    fn update_job_schedule() {
596        let db = test_db();
597        let id = create_job(&db, "j", "a1", "every", None, "{}").unwrap();
598        let changed = update_job(&db, &id, None, Some("cron"), Some("0 9 * * *"), None).unwrap();
599        assert!(changed);
600        let job = get_job(&db, &id).unwrap().unwrap();
601        assert_eq!(job.schedule_kind, "cron");
602        assert_eq!(job.schedule_expr.as_deref(), Some("0 9 * * *"));
603    }
604
605    #[test]
606    fn update_job_enabled_flag() {
607        let db = test_db();
608        let id = create_job(&db, "j", "a1", "every", None, "{}").unwrap();
609
610        let changed = update_job(&db, &id, None, None, None, Some(false)).unwrap();
611        assert!(changed);
612        let job = get_job(&db, &id).unwrap().unwrap();
613        assert!(!job.enabled);
614
615        let changed = update_job(&db, &id, None, None, None, Some(true)).unwrap();
616        assert!(changed);
617        let job = get_job(&db, &id).unwrap().unwrap();
618        assert!(job.enabled);
619    }
620
621    #[test]
622    fn update_job_empty_returns_false() {
623        let db = test_db();
624        let id = create_job(&db, "j", "a1", "every", None, "{}").unwrap();
625        let changed = update_job(&db, &id, None, None, None, None).unwrap();
626        assert!(!changed);
627    }
628
629    #[test]
630    fn update_job_nonexistent_returns_false() {
631        let db = test_db();
632        let changed = update_job(&db, "no-such-id", Some("new-name"), None, None, None).unwrap();
633        assert!(!changed);
634    }
635
636    #[test]
637    fn update_job_all_fields() {
638        let db = test_db();
639        let id = create_job(&db, "j", "a1", "every", None, "{}").unwrap();
640        let changed = update_job(
641            &db,
642            &id,
643            Some("updated"),
644            Some("cron"),
645            Some("*/5 * * * *"),
646            Some(false),
647        )
648        .unwrap();
649        assert!(changed);
650        let job = get_job(&db, &id).unwrap().unwrap();
651        assert_eq!(job.name, "updated");
652        assert_eq!(job.schedule_kind, "cron");
653        assert_eq!(job.schedule_expr.as_deref(), Some("*/5 * * * *"));
654        assert!(!job.enabled);
655    }
656
657    #[test]
658    fn list_runs_empty() {
659        let db = test_db();
660        let runs = list_runs(&db, None, None, None, 100).unwrap();
661        assert!(runs.is_empty());
662    }
663
664    #[test]
665    fn list_runs_all() {
666        let db = test_db();
667        let jid = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
668        record_run(&db, &jid, "success", Some(100), None, None).unwrap();
669        record_run(&db, &jid, "error", Some(50), Some("boom"), None).unwrap();
670        record_run(&db, &jid, "success", Some(200), None, None).unwrap();
671
672        let runs = list_runs(&db, None, None, None, 100).unwrap();
673        assert_eq!(runs.len(), 3);
674    }
675
676    #[test]
677    fn list_runs_by_job_id() {
678        let db = test_db();
679        let j1 = create_job(&db, "job1", "a1", "every", None, "{}").unwrap();
680        let j2 = create_job(&db, "job2", "a1", "every", None, "{}").unwrap();
681        record_run(&db, &j1, "success", Some(10), None, None).unwrap();
682        record_run(&db, &j1, "success", Some(20), None, None).unwrap();
683        record_run(&db, &j2, "success", Some(30), None, None).unwrap();
684
685        let runs = list_runs(&db, None, None, Some(&j1), 100).unwrap();
686        assert_eq!(runs.len(), 2);
687        for run in &runs {
688            assert_eq!(run.job_id, j1);
689        }
690    }
691
692    #[test]
693    fn list_runs_respects_limit() {
694        let db = test_db();
695        let jid = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
696        for _ in 0..10 {
697            record_run(&db, &jid, "success", Some(10), None, None).unwrap();
698        }
699        let runs = list_runs(&db, None, None, None, 3).unwrap();
700        assert_eq!(runs.len(), 3);
701    }
702
703    #[test]
704    fn list_runs_fields_populated() {
705        let db = test_db();
706        let jid = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
707        let run_id = record_run(&db, &jid, "error", Some(42), Some("timeout"), None).unwrap();
708        let runs = list_runs(&db, None, None, None, 10).unwrap();
709        assert_eq!(runs.len(), 1);
710        assert_eq!(runs[0].id, run_id);
711        assert_eq!(runs[0].job_id, jid);
712        assert_eq!(runs[0].status, "error");
713        assert_eq!(runs[0].duration_ms, Some(42));
714        assert_eq!(runs[0].error.as_deref(), Some("timeout"));
715        assert!(!runs[0].created_at.is_empty());
716    }
717
718    #[test]
719    fn delete_job_cascades_cron_runs() {
720        let db = test_db();
721        let jid = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
722        record_run(&db, &jid, "success", Some(10), None, None).unwrap();
723        record_run(&db, &jid, "error", Some(5), Some("oops"), None).unwrap();
724
725        assert!(delete_job(&db, &jid).unwrap());
726
727        // Verify runs were cascade-deleted
728        let runs = list_runs(&db, None, None, Some(&jid), 100).unwrap();
729        assert!(runs.is_empty());
730    }
731
732    #[test]
733    fn release_lease_wrong_holder_is_noop() {
734        let db = test_db();
735        let jid = create_job(&db, "task", "a1", "every", None, "{}").unwrap();
736        acquire_lease(&db, &jid, "inst-1").unwrap();
737
738        // Releasing with wrong holder should not clear the lease
739        release_lease(&db, &jid, "inst-2").unwrap();
740        let job = get_job(&db, &jid).unwrap().unwrap();
741        assert_eq!(job.lease_holder.as_deref(), Some("inst-1"));
742    }
743}
roboticus_db/cron.rs

roboticus_db/
cron.rs
