fix(missions): scheduled missions never fired — nothing read missions.schedule
The wizard has collected a cron since `0047_missions.sql` ("schedule JSONB
carries the trigger config (cron | one_shot | on_event)"), the frontend posts
`{kind:"cron", cron}`, and the API persists it faithfully. Nothing has 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.
Proven before fixing, on the shipped build: a mission with `* * * * *` sat in
`draft` for 4m34s and started ZERO topology runs. After this change the same
mission launched on its next occurrence and recorded one `fired` row.
Two pieces were missing, and they are the two `routines` already had:
- `missions.next_run_at` — schedule STATE. `schedule` is user intent and stays
untouched; without somewhere to record which occurrence is owed there is
nothing to put a `<= now()` predicate on, which is why no enumerator could
be written against the JSONB alone.
- `mission_fires` — one row per (mission, occurrence). 0063_routine_fires.sql
called this exact case: "For a scheduled *mission* it costs a container, a
repo checkout, and real money — which is why this lands before mission
scheduling does."
`mission_schedule.rs` deliberately mirrors `cm-scheduler`'s shape rather than
inventing a second one: atomic `FOR UPDATE SKIP LOCKED` claim, reschedule
BEFORE dispatch so a failing launch cannot stall the clock, claim the slot
before launching so a crash mid-launch is retried rather than dropped, and a
fan-out cap. The cap is 5, not the scheduler's 25, because a mission firing is
a container and a checkout where a routine firing may be one turn.
The claim skips `status = 'running'`: 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. Launch goes through `mission_orchestrator::on_launch` +
`missions::set_status`, the same path as the draft→running transition, so one
code path mints a crew. An unattended launch acts as the workspace owner
(`users::owner_of_workspace`) since missions carry no creator column; a
workspace without one settles the occurrence `failed` with the reason rather
than dropping it silently.
Backfill blast radius was MEASURED, not assumed: prod has zero missions with a
cron, this workstation had exactly one — the control created to prove the bug.
Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
co-authored by
Claude Opus 5
Claude Opus 5
parent
3cc65c22c4
commit
f87853ecf9
@@ -0,0 +1,318 @@
|
||||
//! 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>>,
|
||||
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(),
|
||||
)
|
||||
.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>>,
|
||||
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(), 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})"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user