fix(missions): a failed phase stranded its mission at running forever
Found by counting containers during a cleanup, not by a test. gw-04 was holding
a per-mission runtime container for a mission whose only topology run had failed
three days earlier — phases `pending,failed`, mission still `running`.
The interaction, which lived entirely between two queries' predicates:
`start_pending_phases` launches a phase only when EVERY lower-order phase is
`completed`, so once one fails the phases after it can never run. They stayed
`pending`. `close_finished_missions` closes a mission only when NO phase is
outside ('completed','failed','skipped') — so a `pending` phase that would never
run kept the mission `running` indefinitely. And `mission_runtime`'s sweeper
fires N minutes after a TERMINAL state, so the container was never reaped.
One leaked container per failed multi-phase mission, accumulating silently, with
nothing in any log saying so. Neither query is wrong alone; the bug is that
nothing marked the phases the failure had made unreachable.
`skip_unreachable_phases` says it: a `pending` phase with a `failed` phase at a
LOWER order_idx becomes `skipped` — strictly earlier, because order is what makes
a phase unreachable, and a failure later in the list says nothing about one still
queued ahead of it. `skipped` is not a new concept: `close_finished_missions`
already treats it as terminal, and it is the honest word for a phase that was
never run, as distinct from one that failed.
RETRY HAD TO MOVE WITH IT, or this trades one bug for another. `retry_phase`
required the mission to be `running`, so closing failed missions would have made
the one outcome you would actually want to retry the one you could not. It now
accepts `failed` too, and in one transaction: resets the phase, REOPENS the
phases its failure had skipped (without that, a retry runs the phase and stops,
because everything after it is terminal-by-skip), and puts the mission back to
`running` — every launcher and closer keys off that status. `completed` and
`cancelled` stay refused; reopening those is a different decision.
557 tests pass, clippy clean. Three DB tests against real SQL, including that a
phase queued BEFORE the failure is untouched and that a draft's phases are never
swept.
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
bf2055e725
commit
d24823b6f3
@@ -0,0 +1,156 @@
|
||||
//! A failed phase must not strand its mission at `running` forever.
|
||||
//!
|
||||
//! `start_pending_phases` launches a phase only when every lower-order phase is
|
||||
//! `completed`, so once one fails the rest can never run. They sat `pending`,
|
||||
//! and `close_finished_missions` requires no phase to be non-terminal — so the
|
||||
//! mission never finished, and `mission_runtime`'s sweeper (which fires after a
|
||||
//! terminal state) never reaped its container.
|
||||
//!
|
||||
//! Found by counting containers on gw-04, not by a test: one leaked runtime
|
||||
//! container per failed multi-phase mission, accumulating for days. This is the
|
||||
//! SQL that ends it, tested against a real database because the bug lived
|
||||
//! entirely in the interaction between two queries' predicates.
|
||||
|
||||
use cm_domain::WorkspaceId;
|
||||
use uuid::Uuid;
|
||||
|
||||
async fn workspace(pool: &sqlx::PgPool) -> WorkspaceId {
|
||||
let ws = cm_domain::Workspace {
|
||||
id: WorkspaceId::new(),
|
||||
name: "Unreachable".into(),
|
||||
plan: "team".into(),
|
||||
};
|
||||
cm_db::repo::workspaces::insert(pool, &ws).await.expect("workspace");
|
||||
ws.id
|
||||
}
|
||||
|
||||
/// A running mission whose phase 0 failed and whose phases 1..n never started.
|
||||
async fn stuck_mission(pool: &sqlx::PgPool, ws: WorkspaceId) -> Uuid {
|
||||
let id = Uuid::now_v7();
|
||||
sqlx::query(
|
||||
"INSERT INTO missions (id, workspace_id, title, template_kind, status, schedule, config)
|
||||
VALUES ($1, $2, 'stuck', 'research_and_code', 'running', '{}'::jsonb, '{}'::jsonb)",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(ws.as_uuid())
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("mission");
|
||||
for (idx, status) in [(0, "failed"), (1, "pending"), (2, "pending")] {
|
||||
sqlx::query(
|
||||
"INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
|
||||
VALUES ($1, $2, 'coding', $3, $4, '{}'::jsonb)",
|
||||
)
|
||||
.bind(Uuid::now_v7())
|
||||
.bind(id)
|
||||
.bind(idx)
|
||||
.bind(status)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("phase");
|
||||
}
|
||||
id
|
||||
}
|
||||
|
||||
/// The sweep, as `phase_runner::skip_unreachable_phases` runs it.
|
||||
async fn skip_unreachable(pool: &sqlx::PgPool) -> u64 {
|
||||
sqlx::query(
|
||||
"UPDATE mission_phases mp
|
||||
SET status = 'skipped', completed_at = now()
|
||||
WHERE mp.status = 'pending'
|
||||
AND EXISTS (SELECT 1 FROM missions m WHERE m.id = mp.mission_id AND m.status = 'running')
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM mission_phases prior
|
||||
WHERE prior.mission_id = mp.mission_id
|
||||
AND prior.order_idx < mp.order_idx
|
||||
AND prior.status = 'failed'
|
||||
)",
|
||||
)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("skip")
|
||||
.rows_affected()
|
||||
}
|
||||
|
||||
async fn statuses(pool: &sqlx::PgPool, mission: Uuid) -> Vec<String> {
|
||||
sqlx::query_scalar("SELECT status FROM mission_phases WHERE mission_id = $1 ORDER BY order_idx")
|
||||
.bind(mission)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.expect("statuses")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_phase_makes_the_later_ones_unreachable_not_pending_forever() {
|
||||
let pool = cm_testkit::test_pool().await;
|
||||
let ws = workspace(&pool).await;
|
||||
let m = stuck_mission(&pool, ws).await;
|
||||
|
||||
assert_eq!(skip_unreachable(&pool).await, 2, "both later phases are unreachable");
|
||||
assert_eq!(
|
||||
statuses(&pool, m).await,
|
||||
vec!["failed", "skipped", "skipped"],
|
||||
"a phase that can never run must say so, or the mission never closes"
|
||||
);
|
||||
// Every phase is now terminal, which is what `close_finished_missions`
|
||||
// waits for — the container sweeper keys off the mission reaching that.
|
||||
let non_terminal: i64 = sqlx::query_scalar(
|
||||
"SELECT count(*) FROM mission_phases
|
||||
WHERE mission_id = $1 AND status NOT IN ('completed','failed','skipped')",
|
||||
)
|
||||
.bind(m)
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(non_terminal, 0);
|
||||
}
|
||||
|
||||
/// A phase waiting AHEAD of the failure is untouched: order is what makes a
|
||||
/// phase unreachable, and a failure later in the list says nothing about one
|
||||
/// still queued before it.
|
||||
#[tokio::test]
|
||||
async fn a_phase_before_the_failure_is_left_alone() {
|
||||
let pool = cm_testkit::test_pool().await;
|
||||
let ws = workspace(&pool).await;
|
||||
let m = Uuid::now_v7();
|
||||
sqlx::query(
|
||||
"INSERT INTO missions (id, workspace_id, title, template_kind, status, schedule, config)
|
||||
VALUES ($1, $2, 'ordered', 'research_and_code', 'running', '{}'::jsonb, '{}'::jsonb)",
|
||||
)
|
||||
.bind(m)
|
||||
.bind(ws.as_uuid())
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
for (idx, status) in [(0, "pending"), (1, "failed"), (2, "pending")] {
|
||||
sqlx::query(
|
||||
"INSERT INTO mission_phases (id, mission_id, kind, order_idx, status, config)
|
||||
VALUES ($1, $2, 'coding', $3, $4, '{}'::jsonb)",
|
||||
)
|
||||
.bind(Uuid::now_v7())
|
||||
.bind(m)
|
||||
.bind(idx)
|
||||
.bind(status)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
skip_unreachable(&pool).await;
|
||||
assert_eq!(statuses(&pool, m).await, vec!["pending", "failed", "skipped"]);
|
||||
}
|
||||
|
||||
/// A mission that is not running is not swept: a draft's phases are pending by
|
||||
/// definition and must not be skipped out from under it.
|
||||
#[tokio::test]
|
||||
async fn only_a_running_missions_phases_are_skipped() {
|
||||
let pool = cm_testkit::test_pool().await;
|
||||
let ws = workspace(&pool).await;
|
||||
let m = stuck_mission(&pool, ws).await;
|
||||
sqlx::query("UPDATE missions SET status = 'draft' WHERE id = $1")
|
||||
.bind(m)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(skip_unreachable(&pool).await, 0);
|
||||
assert_eq!(statuses(&pool, m).await, vec!["failed", "pending", "pending"]);
|
||||
}
|
||||
Reference in New Issue
Block a user