Omar Sobh
8a4e222aec
Hooks the topology_worker's post-terminal path into a new notify_run_completed repo helper that atomically transitions the topic processing → reviewing when the completed run has research_topic_id set AND no siblings for that topic are still queued or running. Guarded on status='processing' so a retry, a re-fire, or a topic already past processing are all no-ops. Best-effort at the worker; DB hiccups are logged and never fail the run. The manual /submit-review endpoint stays as an escape hatch for topics that end up parked in processing with nothing to complete (updated the doc comment).
376 lines
12 KiB
Rust
376 lines
12 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::{loops, research_topics, topology_runs, users, workspaces};
|
|
use cm_domain::{Role, User, UserId, Workspace, WorkspaceId};
|
|
use serde_json::json;
|
|
use uuid::Uuid;
|
|
|
|
async fn seed_user(pool: &sqlx::PgPool, ws: WorkspaceId, email: &str) -> UserId {
|
|
let user = User {
|
|
id: UserId::new(),
|
|
workspace_id: ws,
|
|
email: email.into(),
|
|
role: Role::Owner,
|
|
display_name: "Owner".into(),
|
|
created_at: time::OffsetDateTime::UNIX_EPOCH,
|
|
};
|
|
users::insert(pool, &user).await.unwrap();
|
|
user.id
|
|
}
|
|
|
|
/// Enqueue a durable topology run with `research_topic_id` set. Kept in the
|
|
/// test file to avoid a production repo helper for the topic-scoped enqueue
|
|
/// path (nothing else in the codebase writes this column yet).
|
|
async fn enqueue_run_with_topic(
|
|
pool: &sqlx::PgPool,
|
|
workspace_id: WorkspaceId,
|
|
task: &str,
|
|
topic_id: Uuid,
|
|
) -> Uuid {
|
|
let id = Uuid::now_v7();
|
|
let graph = json!({"kind": "pipeline", "nodes": [{"id":"n","role":"r"}], "edges": []});
|
|
sqlx::query!(
|
|
"INSERT INTO topology_runs
|
|
(id, workspace_id, task, kind, status, graph, tier, research_topic_id)
|
|
VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
|
|
id,
|
|
workspace_id.as_uuid(),
|
|
task,
|
|
graph,
|
|
topic_id,
|
|
)
|
|
.execute(pool)
|
|
.await
|
|
.unwrap();
|
|
id
|
|
}
|
|
|
|
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());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cancel_transitions_only_active_runs() {
|
|
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();
|
|
|
|
// A queued run cancels; the worker sees the new status (no workspace scope).
|
|
assert!(topology_runs::cancel(&pool, id, ws).await.unwrap());
|
|
assert_eq!(
|
|
topology_runs::current_status(&pool, id)
|
|
.await
|
|
.unwrap()
|
|
.as_deref(),
|
|
Some("cancelled")
|
|
);
|
|
assert_eq!(
|
|
topology_runs::status(&pool, id, ws).await.unwrap().status,
|
|
"cancelled"
|
|
);
|
|
|
|
// Already terminal → cannot cancel again; wrong workspace → no-op.
|
|
assert!(!topology_runs::cancel(&pool, id, ws).await.unwrap());
|
|
let other = seed_workspace(&pool).await;
|
|
let id2 = Uuid::now_v7();
|
|
topology_runs::enqueue_run(&pool, id2, ws, "t", &graph)
|
|
.await
|
|
.unwrap();
|
|
assert!(!topology_runs::cancel(&pool, id2, other).await.unwrap());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
|
|
let pool = cm_testkit::test_pool().await;
|
|
let ws = seed_workspace(&pool).await;
|
|
|
|
// A loop needs a valid `created_by` user in the same workspace.
|
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
|
|
|
let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
|
|
let triggers = json!({});
|
|
let repeat = json!({"kind": "infinite"});
|
|
let loop_id = loops::create(
|
|
&pool,
|
|
loops::NewLoop {
|
|
workspace_id: ws.as_uuid(),
|
|
title: "L",
|
|
description: "d",
|
|
graph: &graph,
|
|
task_template: "task",
|
|
triggers: &triggers,
|
|
repeat_policy: &repeat,
|
|
enabled: true,
|
|
next_fire_at: None,
|
|
webhook_token: None,
|
|
webhook_signing_key: None,
|
|
created_by: user_id.as_uuid(),
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
// Three iterations of the target loop, plus a naked run + another loop's
|
|
// iteration that should be filtered out.
|
|
let it1 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t1", &graph, 1, None)
|
|
.await
|
|
.unwrap();
|
|
let it2 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t2", &graph, 2, Some(it1))
|
|
.await
|
|
.unwrap();
|
|
let it3 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t3", &graph, 3, Some(it2))
|
|
.await
|
|
.unwrap();
|
|
topology_runs::enqueue_run(&pool, Uuid::now_v7(), ws, "naked", &graph)
|
|
.await
|
|
.unwrap();
|
|
let other_loop = loops::create(
|
|
&pool,
|
|
loops::NewLoop {
|
|
workspace_id: ws.as_uuid(),
|
|
title: "L2",
|
|
description: "d2",
|
|
graph: &graph,
|
|
task_template: "task2",
|
|
triggers: &triggers,
|
|
repeat_policy: &repeat,
|
|
enabled: true,
|
|
next_fire_at: None,
|
|
webhook_token: None,
|
|
webhook_signing_key: None,
|
|
created_by: user_id.as_uuid(),
|
|
},
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let _other_it =
|
|
loops::enqueue_iteration(&pool, other_loop, ws.as_uuid(), "other", &graph, 1, None)
|
|
.await
|
|
.unwrap();
|
|
|
|
let rows = topology_runs::list_by_loop(&pool, ws, loop_id, 20)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(rows.len(), 3, "only the target loop's iterations");
|
|
// Newest iteration first.
|
|
assert_eq!(rows[0].iteration, Some(3));
|
|
assert_eq!(rows[0].id, it3);
|
|
assert_eq!(rows[1].iteration, Some(2));
|
|
assert_eq!(rows[1].id, it2);
|
|
assert_eq!(rows[2].iteration, Some(1));
|
|
assert_eq!(rows[2].id, it1);
|
|
// Each iteration is still a durable run — finished_at is None until completion.
|
|
assert!(rows.iter().all(|r| r.finished_at.is_none()));
|
|
assert!(rows.iter().all(|r| r.kind == "run"));
|
|
|
|
// Wrong workspace: nothing.
|
|
let other_ws = seed_workspace(&pool).await;
|
|
let cross = topology_runs::list_by_loop(&pool, other_ws, loop_id, 20)
|
|
.await
|
|
.unwrap();
|
|
assert!(cross.is_empty(), "loops are scoped by workspace");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn notify_run_completed_transitions_topic_when_no_siblings_in_flight() {
|
|
let pool = cm_testkit::test_pool().await;
|
|
let ws = seed_workspace(&pool).await;
|
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
|
|
|
let topic = research_topics::create(
|
|
&pool,
|
|
ws.as_uuid(),
|
|
"Topic",
|
|
"desc",
|
|
"spec",
|
|
user_id.as_uuid(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// The auto-transition only fires while the topic is `processing`.
|
|
research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
|
|
.await
|
|
.unwrap();
|
|
|
|
let run_id = enqueue_run_with_topic(&pool, ws, "task", topic).await;
|
|
// Flip to a terminal state before calling — mirrors the worker order.
|
|
let result = json!({"final_output": "done"});
|
|
topology_runs::complete(&pool, run_id, &result)
|
|
.await
|
|
.unwrap();
|
|
|
|
let transitioned = topology_runs::notify_run_completed(&pool, run_id)
|
|
.await
|
|
.unwrap();
|
|
assert!(transitioned, "no siblings in flight → topic transitions");
|
|
let t = research_topics::get(&pool, topic, ws.as_uuid())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(t.status, "reviewing");
|
|
|
|
// Idempotent: a second call after the topic has already left `processing`
|
|
// is a no-op.
|
|
let again = topology_runs::notify_run_completed(&pool, run_id)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
!again,
|
|
"second call is a no-op — topic is no longer processing"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn notify_run_completed_leaves_topic_processing_when_siblings_in_flight() {
|
|
let pool = cm_testkit::test_pool().await;
|
|
let ws = seed_workspace(&pool).await;
|
|
let user_id = seed_user(&pool, ws, "[email protected]").await;
|
|
|
|
let topic = research_topics::create(
|
|
&pool,
|
|
ws.as_uuid(),
|
|
"Topic",
|
|
"desc",
|
|
"spec",
|
|
user_id.as_uuid(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
|
|
.await
|
|
.unwrap();
|
|
|
|
let done = enqueue_run_with_topic(&pool, ws, "first", topic).await;
|
|
let _still_queued = enqueue_run_with_topic(&pool, ws, "second", topic).await;
|
|
topology_runs::complete(&pool, done, &json!({"final_output": "x"}))
|
|
.await
|
|
.unwrap();
|
|
|
|
let transitioned = topology_runs::notify_run_completed(&pool, done)
|
|
.await
|
|
.unwrap();
|
|
assert!(!transitioned, "sibling still queued → hold");
|
|
let t = research_topics::get(&pool, topic, ws.as_uuid())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(t.status, "processing");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn notify_run_completed_ignores_runs_with_no_research_topic() {
|
|
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":"n","role":"r"}], "edges": []});
|
|
topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
|
|
.await
|
|
.unwrap();
|
|
topology_runs::complete(&pool, id, &json!({}))
|
|
.await
|
|
.unwrap();
|
|
|
|
let transitioned = topology_runs::notify_run_completed(&pool, id)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
!transitioned,
|
|
"no research_topic_id → nothing to transition"
|
|
);
|
|
}
|