Omar SobhandClaude Opus 4.8
272669e1f5
POST /api/topologies/run now ENQUEUES a durable job and returns 202
{run_id, status:queued} instead of executing inside the HTTP request — the
prerequisite for long-horizon runs (no client/proxy/LB timeout, survives
restarts).
topology_worker: a spawned loop that requeues stale running jobs, claims the
next queued one (CAS via FOR UPDATE SKIP LOCKED), drives it through
execute_resumable, and checkpoints RunProgress after every step; on crash the
stale sweep requeues it and the next claim resumes from the last checkpoint.
Wired into server startup beside the scheduler + resume sweeper.
GET /api/topology-runs/{id} now reports lifecycle status/kind/error/checkpoint
+ the result blob (kept the `comparison` field name for back-compat with the
compare UI; null until completed). list_runs includes status + kind.
Tests: durable lifecycle (enqueue→claim→checkpoint→complete) + stale-requeue
resume, both green; p0 endpoints (compare path) unchanged. 13 + 2 tests pass,
clippy clean.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
88 lines
3.4 KiB
Rust
88 lines
3.4 KiB
Rust
//! The durable topology-job lifecycle at the persistence layer: enqueue → claim
|
|
//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
|
|
//! foundation that lets long-horizon topology runs survive worker restarts.
|
|
|
|
use cm_db::repo::{topology_runs, workspaces};
|
|
use cm_domain::{Workspace, WorkspaceId};
|
|
use serde_json::json;
|
|
use uuid::Uuid;
|
|
|
|
async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
|
|
let ws = Workspace {
|
|
id: WorkspaceId::new(),
|
|
name: "Acme".into(),
|
|
plan: "team".into(),
|
|
};
|
|
workspaces::insert(pool, &ws).await.unwrap();
|
|
ws.id
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn durable_run_lifecycle_enqueue_claim_checkpoint_complete() {
|
|
let pool = cm_testkit::test_pool().await;
|
|
let ws = seed_workspace(&pool).await;
|
|
|
|
let id = Uuid::now_v7();
|
|
let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
|
|
topology_runs::enqueue_run(&pool, id, ws, "write a haiku", &graph)
|
|
.await
|
|
.unwrap();
|
|
|
|
// Status starts queued, no result yet.
|
|
let st = topology_runs::status(&pool, id, ws).await.unwrap();
|
|
assert_eq!(st.status, "queued");
|
|
assert_eq!(st.kind, "run");
|
|
assert!(st.result.is_none());
|
|
|
|
// Claim flips it to running and returns the job + its graph.
|
|
let claimed = topology_runs::claim_next_queued(&pool)
|
|
.await
|
|
.unwrap()
|
|
.expect("a queued job to claim");
|
|
assert_eq!(claimed.id, id);
|
|
assert!(claimed.graph.is_some());
|
|
assert!(claimed.checkpoint.is_none());
|
|
|
|
// A second claim finds nothing (the job is no longer queued).
|
|
assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_none());
|
|
|
|
// Checkpoint mid-run progress.
|
|
let progress = json!({"completed": 1, "outputs": ["draft"], "records": [], "totals": {}});
|
|
topology_runs::checkpoint(&pool, id, &progress, 1).await.unwrap();
|
|
let st = topology_runs::status(&pool, id, ws).await.unwrap();
|
|
assert_eq!(st.status, "running");
|
|
assert_eq!(st.last_event_id, 1);
|
|
assert!(st.checkpoint.is_some());
|
|
|
|
// Complete with a result blob.
|
|
let result = json!({"final_output": "a haiku", "totals": {"turns": 1}});
|
|
topology_runs::complete(&pool, id, &result).await.unwrap();
|
|
let st = topology_runs::status(&pool, id, ws).await.unwrap();
|
|
assert_eq!(st.status, "completed");
|
|
assert_eq!(st.result.unwrap()["final_output"], "a haiku");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn stale_running_jobs_are_requeued_for_resume() {
|
|
let pool = cm_testkit::test_pool().await;
|
|
let ws = seed_workspace(&pool).await;
|
|
|
|
let id = Uuid::now_v7();
|
|
let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
|
|
topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
|
|
.await
|
|
.unwrap();
|
|
// Claim it → running.
|
|
topology_runs::claim_next_queued(&pool).await.unwrap().unwrap();
|
|
|
|
// Not stale yet (just claimed) → sweep is a no-op.
|
|
assert_eq!(topology_runs::requeue_stale(&pool, 60.0).await.unwrap(), 0);
|
|
|
|
// With a zero threshold the running job counts as stale and is requeued,
|
|
// so the next claim picks it up again (resume from checkpoint).
|
|
assert_eq!(topology_runs::requeue_stale(&pool, 0.0).await.unwrap(), 1);
|
|
let st = topology_runs::status(&pool, id, ws).await.unwrap();
|
|
assert_eq!(st.status, "queued");
|
|
assert!(topology_runs::claim_next_queued(&pool).await.unwrap().is_some());
|
|
}
|