car_server_core/
command_scheduler.rs

1//! Daemon-native command scheduler (#72).
2//!
3//! A boot-spawned poller that fires **deterministic command tasks** on their
4//! interval cadence while `car-server` runs — the in-daemon alternative to the
5//! OS scheduler for consumers who'd rather car-server own the timer when it's up
6//! anyway. Only fires:
7//! - **command** tasks (a [`car_scheduler::CommandSpec`], not an LLM prompt),
8//! - with an **interval** trigger (cron cadences go to the OS scheduler; there's
9//!   no in-daemon cron evaluator),
10//! - that are **not** OS-installed (so a durable task isn't double-fired).
11//!
12//! Mirrors the evolution cadence: a single non-overlapping task, dies with the
13//! runtime. Runtime-added tasks are picked up on the next poll (the store is the
14//! source of truth), so no per-task re-arming is needed.
15
16use std::collections::HashSet;
17use std::time::Duration;
18
19use car_scheduler::os_schedule::LABEL_PREFIX;
20use car_scheduler::{parse_interval, run_command, TaskStore, TaskTrigger};
21
22/// How often the poller wakes to check for due command tasks. 30s bounds the
23/// worst-case firing lateness; the interval cadence itself can be any value.
24const POLL_SECS: u64 = 30;
25
26/// Retain at most this many execution records per task so a frequently-firing
27/// command can't grow its task file unboundedly.
28const MAX_EXECUTIONS: usize = 50;
29
30/// Spawn the daemon-native command scheduler. Idempotent-safe to call once at
31/// boot; the task lives for the process lifetime.
32pub fn spawn_command_scheduler() {
33    tokio::spawn(async move {
34        let mut ticker = tokio::time::interval(Duration::from_secs(POLL_SECS));
35        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
36        loop {
37            ticker.tick().await;
38            if let Err(e) = poll_once().await {
39                tracing::warn!(target: "car::scheduler", "[command-scheduler] {e}");
40            }
41        }
42    });
43}
44
45/// One poll pass: fire every due, enabled, daemon-native interval command task.
46async fn poll_once() -> Result<(), String> {
47    let store = TaskStore::new(&TaskStore::default_path());
48    // A transient store-read failure must not look like "no tasks"; surface it.
49    let tasks = store
50        .try_list()
51        .map_err(|e| format!("could not read task store: {e}"))?;
52    // If we can't read which tasks the OS scheduler owns, skip this tick rather
53    // than risk double-firing an OS-installed task (fail-safe, not fail-open).
54    let os_installed: HashSet<String> = car_scheduler::list_installed()
55        .map_err(|e| format!("could not read OS schedules (skipping tick): {e}"))?
56        .into_iter()
57        .collect();
58    let now = chrono::Utc::now();
59
60    for mut task in tasks {
61        if !task.enabled || !task.is_command() || task.trigger != TaskTrigger::Interval {
62            continue;
63        }
64        // Durable (OS-installed) tasks are fired by launchd/cron — don't double-fire.
65        let label = format!("{LABEL_PREFIX}{}", task.id);
66        if os_installed.contains(&label) {
67            continue;
68        }
69        let interval_secs = parse_interval(&task.schedule);
70        if interval_secs <= 0.0 {
71            continue;
72        }
73        let due = match task.last_run_at {
74            None => true, // never run → fire now
75            Some(last) => (now - last).num_seconds() as f64 >= interval_secs,
76        };
77        if !due {
78            continue;
79        }
80        let Some(cmd) = task.command.clone() else {
81            continue;
82        };
83        let exec = run_command(&cmd).await;
84        tracing::info!(
85            target: "car::scheduler",
86            task = %task.id, status = ?exec.status,
87            "[command-scheduler] fired command task"
88        );
89        task.last_run_at = Some(now);
90        task.run_count = task.run_count.saturating_add(1);
91        task.status = exec.status;
92        task.executions.push(exec);
93        if task.executions.len() > MAX_EXECUTIONS {
94            let drop = task.executions.len() - MAX_EXECUTIONS;
95            task.executions.drain(0..drop);
96        }
97        if let Err(e) = store.save(&task) {
98            tracing::warn!(target: "car::scheduler", "[command-scheduler] save {}: {e}", task.id);
99        }
100    }
101    Ok(())
102}
car_server_core/command_scheduler.rs

car_server_core/
command_scheduler.rs
