clawmates/crates/cm-api/src/mission_schedule.rs

//! Launching missions that are due.
//!
//! `missions.schedule` has carried a cron since `0047_missions.sql` — the
//! wizard collects it, the API persists it — and until this module nothing ever
//! read it back. The only due-work enumerator in the codebase was
//! `routines::claim_due`, so **every scheduled mission ever created sat in
//! `draft` forever** while the UI reported it was on a schedule. Measured
//! before this was written: a mission with `* * * * *` did not move for four
//! minutes and started no runs.
//!
//! The shape here is deliberately `cm-scheduler`'s, not a second invention:
//!
//! - **Claim atomically** (`FOR UPDATE SKIP LOCKED`) so replicas fire once.
//! - **Advance the clock before dispatching**, so a failing launch cannot stall
//!   the schedule.
//! - **Record the claim first** in `mission_fires`, keyed by the occurrence's
//!   own timestamp, so a crash between those two is retried rather than
//!   silently dropped — and a slot already launched is never launched twice.
//! - **Cap the fan-out**, because a backlog would otherwise start one container
//!   per missed occurrence.
//!
//! The one thing it does NOT share with routines is the launch itself: a due
//! mission goes through `mission_orchestrator::on_launch` and
//! `missions::set_status`, exactly as the draft→running transition in
//! `routes::missions::set_status` does, so there is one path that mints a crew.

use cm_db::repo::missions as missions_repo;
use sqlx::{PgPool, Row};
use time::OffsetDateTime;
use uuid::Uuid;

/// Most missions one tick will launch.
///
/// Lower than the scheduler's 25: a mission firing is a container, a repo
/// checkout and real model spend, where a routine firing may be a single turn.
/// The remainder stays due and is taken by the next tick.
const MAX_LAUNCHES_PER_TICK: usize = 5;

/// A mission whose occurrence has come due and been claimed.
#[derive(Debug)]
pub struct DueMission {
    pub id: Uuid,
    pub workspace_id: Uuid,
    pub title: String,
    pub cron: Option<String>,
    /// The occurrence that came due — the value `next_run_at` held. Identifies
    /// the slot in `mission_fires`, so it must not be re-read from the clock.
    pub slot: OffsetDateTime,
}

/// Claim every mission due at `now`, atomically.
///
/// `next_run_at` is cleared by the claim. The caller recomputes it from the
/// cron and writes it back; a mission whose cron no longer yields an occurrence
/// simply stays cleared and stops firing, which is the correct end state for
/// a one-shot or an exhausted schedule.
pub async fn claim_due(pool: &PgPool, now: OffsetDateTime) -> Result<Vec<DueMission>, String> {
    let rows = sqlx::query(
        "UPDATE missions SET next_run_at = NULL
          WHERE id IN (
              SELECT id FROM missions
               WHERE next_run_at IS NOT NULL
                 AND next_run_at <= $1
                 -- Never relaunch a mission that is mid-flight. A daily cron on
                 -- a mission that takes longer than a day must skip the
                 -- occurrence, not stack a second crew on the same workspace.
                 AND status <> 'running'
               FOR UPDATE SKIP LOCKED
          )
          RETURNING id, workspace_id, title, schedule ->> 'cron' AS cron, $1::timestamptz AS slot",
    )
    .bind(now)
    .fetch_all(pool)
    .await
    .map_err(|e| format!("claim due missions: {e}"))?;

    Ok(rows
        .into_iter()
        .map(|r| DueMission {
            id: r.get("id"),
            workspace_id: r.get("workspace_id"),
            title: r.get("title"),
            cron: r.get("cron"),
            slot: r.get("slot"),
        })
        .collect())
}

/// Record that this occurrence was taken. `false` means another replica (or an
/// earlier attempt) already has it and this one must not launch.
async fn claim_slot(pool: &PgPool, mission_id: Uuid, slot: OffsetDateTime) -> Result<bool, String> {
    let inserted = sqlx::query(
        "INSERT INTO mission_fires (mission_id, scheduled_at, status)
         VALUES ($1, $2, 'claimed')
         ON CONFLICT (mission_id, scheduled_at) DO NOTHING",
    )
    .bind(mission_id)
    .bind(slot)
    .execute(pool)
    .await
    .map_err(|e| format!("claim mission fire: {e}"))?;
    Ok(inserted.rows_affected() == 1)
}

async fn settle_slot(
    pool: &PgPool,
    mission_id: Uuid,
    slot: OffsetDateTime,
    status: &str,
    detail: Option<&str>,
) {
    if let Err(e) = sqlx::query(
        "UPDATE mission_fires SET status = $3, detail = $4, completed_at = now()
          WHERE mission_id = $1 AND scheduled_at = $2",
    )
    .bind(mission_id)
    .bind(slot)
    .bind(status)
    .bind(detail)
    .execute(pool)
    .await
    {
        eprintln!("mission_schedule: settling {mission_id} @ {slot} as {status}: {e}");
    }
}

/// Compute and persist the next occurrence.
///
/// A cron that will not parse is reported and the mission left un-scheduled
/// rather than skipped in silence — the whole point of this module is that a
/// schedule which does nothing must never look like a schedule that works.
async fn reschedule(pool: &PgPool, m: &DueMission, after: OffsetDateTime) {
    let Some(cron) = m.cron.as_deref().map(str::trim).filter(|c| !c.is_empty()) else {
        return;
    };
    match cm_runtime::scheduling::next_occurrence(cron, after) {
        Ok(next) => {
            if let Err(e) = sqlx::query("UPDATE missions SET next_run_at = $2 WHERE id = $1")
                .bind(m.id)
                .bind(next)
                .execute(pool)
                .await
            {
                eprintln!("mission_schedule: could not set next_run_at for {}: {e}", m.id);
            }
        }
        Err(e) => eprintln!(
            "mission_schedule: mission {} ({}) has an unusable cron {cron:?} — it will NOT run \
             again until the schedule is corrected: {e}",
            m.id, m.title
        ),
    }
}

/// One pass. Returns how many missions were launched.
pub async fn tick(
    pool: &PgPool,
    node_hub: Option<std::sync::Arc<crate::fleet::NodeHub>>,
    blobs: Option<std::sync::Arc<dyn cm_files::BlobStore>>,
    now: OffsetDateTime,
) -> Result<usize, String> {
    let due = claim_due(pool, now).await?;
    let mut launched = 0usize;

    for m in due.iter().take(MAX_LAUNCHES_PER_TICK) {
        // Clock first: a launch that fails must not stall the schedule.
        reschedule(pool, m, now).await;

        if !claim_slot(pool, m.id, m.slot).await? {
            continue;
        }

        // An unattended launch still needs an actor. Missions carry no creator
        // column, so the workspace owner stands in — the same identity the
        // audit trail already attributes workspace-level action to.
        let workspace = cm_domain::WorkspaceId::from(m.workspace_id);
        let owner = match cm_db::repo::users::owner_of_workspace(pool, workspace).await {
            Ok(u) => u,
            Err(e) => {
                // `fetch_one`, so "no owner" arrives as RowNotFound rather than
                // None. Either way the occurrence is settled `failed` with the
                // reason, never dropped quietly.
                let why = format!("no owner to launch as: {e}");
                eprintln!("mission_schedule: cannot launch {} — {why}", m.id);
                settle_slot(pool, m.id, m.slot, "failed", Some(&why)).await;
                continue;
            }
        };

        match crate::mission_orchestrator::on_launch(
            pool,
            workspace,
            owner,
            m.id,
            node_hub.clone(),
            blobs.clone(),
        )
        .await
        {
            Ok(_) => {
                if let Err(e) =
                    missions_repo::set_status(pool, m.id, m.workspace_id, "running").await
                {
                    let why = format!("launched but could not mark running: {e}");
                    eprintln!("mission_schedule: {} — {why}", m.id);
                    settle_slot(pool, m.id, m.slot, "failed", Some(&why)).await;
                    continue;
                }
                settle_slot(pool, m.id, m.slot, "fired", None).await;
                launched += 1;
                eprintln!(
                    "mission_schedule: launched {} ({}) for occurrence {}",
                    m.id, m.title, m.slot
                );
            }
            Err(e) => {
                eprintln!("mission_schedule: on_launch failed for {}: {e}", m.id);
                settle_slot(pool, m.id, m.slot, "failed", Some(&e)).await;
            }
        }
    }

    if due.len() > MAX_LAUNCHES_PER_TICK {
        eprintln!(
            "mission_schedule: {} due, launched {} this tick (cap {}); the rest stay due",
            due.len(),
            launched,
            MAX_LAUNCHES_PER_TICK
        );
    }
    Ok(launched)
}

/// Spawn the sweep.
pub fn spawn(
    pool: PgPool,
    node_hub: Option<std::sync::Arc<crate::fleet::NodeHub>>,
    blobs: Option<std::sync::Arc<dyn cm_files::BlobStore>>,
    interval: std::time::Duration,
) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(interval);
        // Skip the immediate first tick so a restart loop cannot become a
        // launch loop.
        ticker.tick().await;
        loop {
            ticker.tick().await;
            let now = OffsetDateTime::now_utc();
            match tick(&pool, node_hub.clone(), blobs.clone(), now).await {
                Ok(n) if n > 0 => eprintln!("mission_schedule: launched {n} due mission(s)"),
                Ok(_) => {}
                Err(e) => eprintln!("mission_schedule: sweep failed: {e}"),
            }
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The cap is what stops a backlog turning into a container stampede. A
    /// clock jump or a cron that resolves to "every minute" can leave hundreds
    /// of occurrences owed; each mission launch is a container, a checkout and
    /// real model spend, so this must stay well below the routine scheduler's
    /// 25.
    #[test]
    fn the_launch_cap_is_conservative() {
        assert!(
            MAX_LAUNCHES_PER_TICK <= 5,
            "a mission firing costs far more than a routine firing"
        );
        assert!(MAX_LAUNCHES_PER_TICK >= 1, "a cap of zero never launches");
    }

    /// The claim must never pick up a mission that is already running.
    ///
    /// A daily cron on a mission that takes longer than a day would otherwise
    /// stack a second crew on the same workspace — two containers, two vault
    /// branches, and a seen-set race. Asserted against the SQL text because the
    /// predicate is the whole safety property and it lives only in the query.
    #[test]
    fn the_claim_skips_missions_that_are_still_running() {
        // Re-read the source of the query this module issues.
        let src = include_str!("mission_schedule.rs");
        let claim = src
            .split("pub async fn claim_due")
            .nth(1)
            .expect("claim_due exists");
        let body = &claim[..claim.find("fetch_all").unwrap_or(claim.len())];
        assert!(
            body.contains("status <> 'running'"),
            "claim_due must not relaunch a mission that is mid-flight"
        );
        assert!(
            body.contains("FOR UPDATE SKIP LOCKED"),
            "the claim must be atomic or replicas double-launch"
        );
        assert!(
            body.contains("next_run_at <= $1"),
            "only occurrences that have come due may be claimed"
        );
    }

    /// The clock advances BEFORE the launch, and the slot is claimed before the
    /// launch too. Both orderings matter: reschedule-first means a failing
    /// launch cannot stall the schedule; claim-first means a crash mid-launch
    /// is retried rather than dropped.
    #[test]
    fn the_clock_advances_before_the_launch_is_attempted() {
        let src = include_str!("mission_schedule.rs");
        let tick = src.split("pub async fn tick").nth(1).expect("tick exists");
        let resched = tick.find("reschedule(pool, m, now)").expect("reschedules");
        let claim = tick.find("claim_slot(pool, m.id, m.slot)").expect("claims");
        let launch = tick.find("on_launch(").expect("launches");
        assert!(
            resched < claim && claim < launch,
            "order must be reschedule -> claim -> launch (got {resched}, {claim}, {launch})"
        );
    }
}