fix(missions): #54 — the worker was killing live microVM runs at 180 seconds
My hypothesis in #54 was WRONG, and it was wrong because I built it on a bad measurement: `grep -c 'microvm phase'` returned 0, so I concluded the completion log never printed and blamed the 15-minute reaper. The line was there all along, at 14:17:45. The real cause is worse. `requeue_stale` has NO TIER FILTER. A microvm run's `updated_at` is written once at insert and never again — it is driven by a `tokio::spawn` that owns it start to finish, and nothing in `microvm_executor` writes `topology_runs`. So at 180s the sweeper declared a perfectly healthy run stale and flipped it to `queued`; `claim_next_queued` (no tier filter either) handed it to the worker; `run_job` tried to parse the microvm graph placeholder, which `TopologyGraph` cannot deserialize; and it failed the run with "missing or invalid graph". Mission 019fd43e: run created 14:11:16, mission failed ~14:14:46. 210 seconds — the 180s window plus a tick. The agent went on working and finished at 14:17:45 with three modules written, by which time the phase was already dead and the VM was orphaned. A firecracker process was still alive 1h37m later. THE UNCOMFORTABLE PART: every microVM mission that appeared to work this session did so only by finishing inside three minutes. The 90-second ones dodged this. The harness scenario dodges it. Nothing about that was visible. `WORKER_DRIVEN_TIERS` (team, company, org, swarm, compare) is now the allowlist for all three sweep paths — claim, requeue, reap. An allowlist rather than a denylist so the next self-driven tier is safe by default instead of exposed until someone remembers the file. `tier='session'` had exactly the same exposure and is covered too. A unit test asserts microvm and session are NOT in it, next to the code that inserts them. Two more fixes from the same wreckage: - `destroy` reported `killed: pgid.is_some()` — true whenever there was a pgid to signal, whether or not anything died. It now sends the signal, polls /proc for the group leader, retries, and reports what it OBSERVED; `signalled` keeps the old meaning so "nothing to kill" is distinguishable from "it would not die". - the run-status update is now guarded with `AND status <> 'cancelled'`. An operator cancelling is a decision; this task reporting an outcome minutes later is an observation, and it must not overwrite one with the other. And the root cause of the collect timeout itself: `mission_fs::pack_dir` shipped `target/` in both directions. `mission_delivery` has excluded build output from the DIFF since day one; the TRANSPORT never knew. The host checkout was 9.4 MB of which 8.9 MB was `target/`, tarred and base64'd over vsock each way. `EXCLUDED_PATHS` is now one list shared by both layers, matched on directory name at any depth so a workspace's per-crate `target/` dirs are all covered. 483 tests pass, clippy clean.
This commit is contained in:
@@ -475,17 +475,46 @@ pub async fn destroy(vms: &Vms, vm_id: &str) -> Result<Value, String> {
|
|||||||
(None, wd.clone(), uds, eg)
|
(None, wd.clone(), uds, eg)
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
// Killed means OBSERVED GONE, not asked-to-die.
|
||||||
|
//
|
||||||
|
// This used to report `killed: pgid.is_some()` — true whenever there was a
|
||||||
|
// pgid to signal, whether or not anything died. A firecracker process was
|
||||||
|
// found alive 1h37m after its VM was destroyed, holding a config file in a
|
||||||
|
// workdir that no longer existed, while destroy had reported success. So the
|
||||||
|
// signal is sent, the process is polled, and the answer is what was seen.
|
||||||
|
let mut killed = false;
|
||||||
if let Some(pgid) = pgid {
|
if let Some(pgid) = pgid {
|
||||||
kill_group(pgid).await;
|
for attempt in 0..20 {
|
||||||
|
kill_group(pgid).await;
|
||||||
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||||
|
// The pgid is the group leader's pid, so /proc/<pgid> is the check.
|
||||||
|
if !std::path::Path::new(&format!("/proc/{pgid}")).exists() {
|
||||||
|
killed = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if attempt == 19 {
|
||||||
|
eprintln!(
|
||||||
|
"microvm {vm_id}: process group {pgid} SURVIVED {} kill attempts — \
|
||||||
|
a firecracker process is still holding this VM's resources",
|
||||||
|
attempt + 1
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// Give the group a moment to die before removing the files it has open.
|
|
||||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
|
||||||
let _ = tokio::fs::remove_file(&uds).await;
|
let _ = tokio::fs::remove_file(&uds).await;
|
||||||
// Same trap as firecracker's own socket: nothing unlinks these for us, and a
|
// Same trap as firecracker's own socket: nothing unlinks these for us, and a
|
||||||
// stale file makes the next bind fail with EADDRINUSE.
|
// stale file makes the next bind fail with EADDRINUSE.
|
||||||
let _ = tokio::fs::remove_file(&egress_uds).await;
|
let _ = tokio::fs::remove_file(&egress_uds).await;
|
||||||
let removed = tokio::fs::remove_dir_all(&workdir).await.is_ok();
|
let removed = tokio::fs::remove_dir_all(&workdir).await.is_ok();
|
||||||
Ok(json!({ "vm_id": vm_id, "killed": pgid.is_some(), "workdir_removed": removed }))
|
// `killed` is now observed rather than assumed; `signalled` keeps the old
|
||||||
|
// meaning so a caller can tell "there was nothing to kill" from
|
||||||
|
// "we tried and it would not die".
|
||||||
|
Ok(json!({
|
||||||
|
"vm_id": vm_id,
|
||||||
|
"killed": killed,
|
||||||
|
"signalled": pgid.is_some(),
|
||||||
|
"workdir_removed": removed,
|
||||||
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// VMs this node currently holds, so the server can reap orphans.
|
/// VMs this node currently holds, so the server can reap orphans.
|
||||||
@@ -869,6 +898,9 @@ pub async fn selftest() -> bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let r = destroy(&vms, id).await;
|
let r = destroy(&vms, id).await;
|
||||||
|
// `killed` is now an OBSERVATION — the process was polled and found gone —
|
||||||
|
// rather than "we sent a signal". A firecracker was found alive 1h37m after a
|
||||||
|
// destroy that had reported success.
|
||||||
check(
|
check(
|
||||||
r.as_ref()
|
r.as_ref()
|
||||||
.map(|v| v["killed"] == json!(true) && v["workdir_removed"] == json!(true))
|
.map(|v| v["killed"] == json!(true) && v["workdir_removed"] == json!(true))
|
||||||
|
|||||||
@@ -666,6 +666,25 @@ mod tests {
|
|||||||
assert!(agent_command(&vm_prompt("task")).contains(&format!("--agents {quoted}")));
|
assert!(agent_command(&vm_prompt("task")).contains(&format!("--agents {quoted}")));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The tier this executor inserts must NOT be one the topology worker drives.
|
||||||
|
/// A microvm run is owned by its own spawned task for its whole life; if the
|
||||||
|
/// worker also considers it fair game, it requeues it at 180s, fails it on a
|
||||||
|
/// graph it was never meant to parse, and orphans a live VM. Mission 019fd43e
|
||||||
|
/// died at 210 seconds that way.
|
||||||
|
#[test]
|
||||||
|
fn a_microvm_run_is_not_worker_driven() {
|
||||||
|
assert!(
|
||||||
|
!cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&"microvm"),
|
||||||
|
"the worker must not claim, requeue or reap a self-driven run"
|
||||||
|
);
|
||||||
|
// Same exposure, same reason.
|
||||||
|
assert!(!cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&"session"));
|
||||||
|
// And the tiers it does drive are still there, or nothing runs at all.
|
||||||
|
for t in ["team", "swarm"] {
|
||||||
|
assert!(cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS.contains(&t), "{t}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Solo is the default, and asking for a team must be explicit. Multi-agent
|
/// Solo is the default, and asking for a team must be explicit. Multi-agent
|
||||||
/// costs 3-10x the tokens and is often slower, so a mission that said nothing
|
/// costs 3-10x the tokens and is often slower, so a mission that said nothing
|
||||||
/// must not get one.
|
/// must not get one.
|
||||||
|
|||||||
@@ -40,7 +40,14 @@ use crate::mission_workspace;
|
|||||||
/// coding phase that ran `cargo build` leaves a `target/` directory larger
|
/// coding phase that ran `cargo build` leaves a `target/` directory larger
|
||||||
/// than most repositories, and a patch containing it is unreadable as well as
|
/// than most repositories, and a patch containing it is unreadable as well as
|
||||||
/// enormous.
|
/// enormous.
|
||||||
const EXCLUDED_PATHS: &[&str] = &[
|
///
|
||||||
|
/// `mission_fs` uses this same list for the TRANSPORT, and that is not a
|
||||||
|
/// convenience — it is the fix for a real failure. The diff excluded `target/`
|
||||||
|
/// while the tar that carried the tree in and out did not, so a phase that ran
|
||||||
|
/// `cargo test` shipped its whole build directory over vsock twice. `vm_collect`
|
||||||
|
/// timed out at 300s on mission 019fd43e with the agent's work finished and
|
||||||
|
/// stranded inside a VM. Two layers, one list.
|
||||||
|
pub(crate) const EXCLUDED_PATHS: &[&str] = &[
|
||||||
"target",
|
"target",
|
||||||
"node_modules",
|
"node_modules",
|
||||||
".venv",
|
".venv",
|
||||||
|
|||||||
@@ -48,14 +48,73 @@ pub fn pack_dir(root: &Path, name_in_archive: &str) -> Result<Vec<u8>, String> {
|
|||||||
// Follow no symlinks: a checkout can contain a link pointing outside the
|
// Follow no symlinks: a checkout can contain a link pointing outside the
|
||||||
// tree, and dereferencing it would pull host files into the container.
|
// tree, and dereferencing it would pull host files into the container.
|
||||||
builder.follow_symlinks(false);
|
builder.follow_symlinks(false);
|
||||||
builder
|
append_filtered(&mut builder, root, Path::new(name_in_archive))
|
||||||
.append_dir_all(name_in_archive, root)
|
|
||||||
.map_err(|e| format!("pack {}: {e}", root.display()))?;
|
.map_err(|e| format!("pack {}: {e}", root.display()))?;
|
||||||
builder
|
builder
|
||||||
.into_inner()
|
.into_inner()
|
||||||
.map_err(|e| format!("finish archive for {}: {e}", root.display()))
|
.map_err(|e| format!("finish archive for {}: {e}", root.display()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Directory names never carried across the boundary.
|
||||||
|
///
|
||||||
|
/// The same list the delivery diff uses, deliberately: see
|
||||||
|
/// [`crate::mission_delivery::EXCLUDED_PATHS`]. A build directory is not work —
|
||||||
|
/// it is regenerable output that dwarfs the source, and shipping it cost a
|
||||||
|
/// mission its results when `vm_collect` timed out with the agent's finished work
|
||||||
|
/// still inside the VM.
|
||||||
|
pub fn transport_excludes() -> &'static [&'static str] {
|
||||||
|
crate::mission_delivery::EXCLUDED_PATHS
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Should this directory entry be left out of the archive?
|
||||||
|
///
|
||||||
|
/// Matched on the entry NAME at any depth, not on a path prefix: a workspace has
|
||||||
|
/// a `target/` per crate, and excluding only the root one would still ship the
|
||||||
|
/// rest.
|
||||||
|
pub fn is_excluded(name: &str) -> bool {
|
||||||
|
transport_excludes().contains(&name)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Recursive `append_dir_all` that skips [`transport_excludes`].
|
||||||
|
///
|
||||||
|
/// Hand-rolled because `tar::Builder::append_dir_all` takes no filter. Symlinks
|
||||||
|
/// are added as links rather than followed, matching `follow_symlinks(false)`.
|
||||||
|
fn append_filtered<W: std::io::Write>(
|
||||||
|
builder: &mut tar::Builder<W>,
|
||||||
|
dir: &Path,
|
||||||
|
prefix: &Path,
|
||||||
|
) -> std::io::Result<()> {
|
||||||
|
builder.append_dir(prefix, dir)?;
|
||||||
|
let mut entries: Vec<_> = std::fs::read_dir(dir)?.collect::<Result<Vec<_>, _>>()?;
|
||||||
|
// Stable order so an archive of the same tree is byte-identical, which makes
|
||||||
|
// a size or content difference between two runs mean something.
|
||||||
|
entries.sort_by_key(|e| e.file_name());
|
||||||
|
for entry in entries {
|
||||||
|
let name = entry.file_name();
|
||||||
|
let name_str = name.to_string_lossy();
|
||||||
|
let path = entry.path();
|
||||||
|
let dest = prefix.join(&name);
|
||||||
|
let meta = std::fs::symlink_metadata(&path)?;
|
||||||
|
if meta.is_dir() {
|
||||||
|
if is_excluded(&name_str) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
append_filtered(builder, &path, &dest)?;
|
||||||
|
} else if meta.is_symlink() {
|
||||||
|
let mut header = tar::Header::new_gnu();
|
||||||
|
header.set_metadata(&meta);
|
||||||
|
header.set_entry_type(tar::EntryType::Symlink);
|
||||||
|
header.set_size(0);
|
||||||
|
let target = std::fs::read_link(&path)?;
|
||||||
|
builder.append_link(&mut header, &dest, &target)?;
|
||||||
|
} else {
|
||||||
|
let mut f = std::fs::File::open(&path)?;
|
||||||
|
builder.append_file(&dest, &mut f)?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Unpack a tar into a host directory.
|
/// Unpack a tar into a host directory.
|
||||||
///
|
///
|
||||||
/// `tar` refuses entries whose paths escape the destination, which is the
|
/// `tar` refuses entries whose paths escape the destination, which is the
|
||||||
|
|||||||
@@ -743,8 +743,13 @@ async fn launch_microvm_phase(
|
|||||||
(subagents: {subagents}, teammates: {teammates}) — {}",
|
(subagents: {subagents}, teammates: {teammates}) — {}",
|
||||||
note.chars().take(300).collect::<String>()
|
note.chars().take(300).collect::<String>()
|
||||||
);
|
);
|
||||||
|
// Never overwrite a cancellation. The operator asking to stop is a decision;
|
||||||
|
// this task reporting how the VM turned out is an observation, and it may
|
||||||
|
// land minutes later. Without the guard a cancelled run silently reappears
|
||||||
|
// as completed or failed.
|
||||||
if let Err(e) = sqlx::query(
|
if let Err(e) = sqlx::query(
|
||||||
"UPDATE topology_runs SET status = $2, updated_at = now() WHERE id = $1",
|
"UPDATE topology_runs SET status = $2, updated_at = now()
|
||||||
|
WHERE id = $1 AND status <> 'cancelled'",
|
||||||
)
|
)
|
||||||
.bind(run_id)
|
.bind(run_id)
|
||||||
.bind(status)
|
.bind(status)
|
||||||
@@ -827,8 +832,13 @@ async fn launch_direct_session(
|
|||||||
summary.chars().take(200).collect::<String>()
|
summary.chars().take(200).collect::<String>()
|
||||||
);
|
);
|
||||||
let status = if ok { "completed" } else { "failed" };
|
let status = if ok { "completed" } else { "failed" };
|
||||||
|
// Never overwrite a cancellation. The operator asking to stop is a decision;
|
||||||
|
// this task reporting how the VM turned out is an observation, and it may
|
||||||
|
// land minutes later. Without the guard a cancelled run silently reappears
|
||||||
|
// as completed or failed.
|
||||||
if let Err(e) = sqlx::query(
|
if let Err(e) = sqlx::query(
|
||||||
"UPDATE topology_runs SET status = $2, updated_at = now() WHERE id = $1",
|
"UPDATE topology_runs SET status = $2, updated_at = now()
|
||||||
|
WHERE id = $1 AND status <> 'cancelled'",
|
||||||
)
|
)
|
||||||
.bind(run_id)
|
.bind(run_id)
|
||||||
.bind(status)
|
.bind(status)
|
||||||
|
|||||||
@@ -80,10 +80,21 @@ async fn reap_stuck_runs(pool: &PgPool) -> Result<(), sqlx::Error> {
|
|||||||
FROM topology_runs
|
FROM topology_runs
|
||||||
WHERE status = 'running'
|
WHERE status = 'running'
|
||||||
AND mission_id IS NOT NULL
|
AND mission_id IS NOT NULL
|
||||||
|
-- Only jobs this worker drives. mission_id IS NOT NULL used to mean
|
||||||
|
-- the same thing as orchestrator-driven, and the microvm and session
|
||||||
|
-- tiers broke that: their checkpoint is NULL for life BY DESIGN, so the
|
||||||
|
-- zero-step-records test below is true of a perfectly healthy run.
|
||||||
|
AND tier = ANY($2)
|
||||||
AND created_at < now() - make_interval(secs => $1::float)
|
AND created_at < now() - make_interval(secs => $1::float)
|
||||||
AND coalesce(jsonb_array_length(coalesce(checkpoint->'records', '[]'::jsonb)), 0) = 0",
|
AND coalesce(jsonb_array_length(coalesce(checkpoint->'records', '[]'::jsonb)), 0) = 0",
|
||||||
)
|
)
|
||||||
.bind(REAP_STUCK_AFTER_SECS as f64)
|
.bind(REAP_STUCK_AFTER_SECS as f64)
|
||||||
|
.bind(
|
||||||
|
cm_db::repo::topology_runs::WORKER_DRIVEN_TIERS
|
||||||
|
.iter()
|
||||||
|
.map(|s| (*s).to_string())
|
||||||
|
.collect::<Vec<_>>(),
|
||||||
|
)
|
||||||
.fetch_all(pool)
|
.fetch_all(pool)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
|||||||
@@ -203,32 +203,67 @@ pub async fn check_ephemeral_teardown(
|
|||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Tiers the topology worker drives, and therefore the only ones it may claim,
|
||||||
|
/// requeue or reap.
|
||||||
|
///
|
||||||
|
/// **A load-bearing allowlist, not tidiness.** Both sweepers were written when
|
||||||
|
/// every `running` row was a `cm_orchestrator` job that checkpointed after each
|
||||||
|
/// step. `tier='microvm'` and `tier='session'` broke that assumption: they are
|
||||||
|
/// inserted directly as `running` by `phase_runner`, driven by a `tokio::spawn`
|
||||||
|
/// that owns them start to finish, and they never write `updated_at` or
|
||||||
|
/// `checkpoint` while in flight.
|
||||||
|
///
|
||||||
|
/// Measured cost of the omission: `requeue_stale` flipped an in-flight microVM run
|
||||||
|
/// to `queued` at 180s, `claim_next_queued` handed it to the worker, and the worker
|
||||||
|
/// failed it with "missing or invalid graph" — a microvm run's graph is a
|
||||||
|
/// placeholder `TopologyGraph` cannot parse. Mission 019fd43e died at 210 seconds
|
||||||
|
/// with its agent still working and its VM orphaned. Every microVM mission that
|
||||||
|
/// appeared to work did so only by finishing inside three minutes.
|
||||||
|
///
|
||||||
|
/// An allowlist rather than a denylist on purpose: the next self-driven tier is
|
||||||
|
/// then safe by default, instead of exposed until someone remembers this file.
|
||||||
|
pub const WORKER_DRIVEN_TIERS: &[&str] = &["team", "company", "org", "swarm", "compare"];
|
||||||
|
|
||||||
|
/// The allowlist as owned strings, for binding as `text[]`.
|
||||||
|
fn worker_driven() -> Vec<String> {
|
||||||
|
WORKER_DRIVEN_TIERS.iter().map(|s| (*s).to_string()).collect()
|
||||||
|
}
|
||||||
|
|
||||||
/// Atomically claim the oldest queued job, flipping it to `running`. Uses
|
/// Atomically claim the oldest queued job, flipping it to `running`. Uses
|
||||||
/// `FOR UPDATE SKIP LOCKED` so multiple workers never claim the same job.
|
/// `FOR UPDATE SKIP LOCKED` so multiple workers never claim the same job.
|
||||||
/// Returns `None` when the queue is empty.
|
/// Returns `None` when the queue is empty.
|
||||||
pub async fn claim_next_queued(pool: &PgPool) -> Result<Option<ClaimedTopologyRun>, DbError> {
|
pub async fn claim_next_queued(pool: &PgPool) -> Result<Option<ClaimedTopologyRun>, DbError> {
|
||||||
let row = sqlx::query!(
|
use sqlx::Row as _;
|
||||||
|
// A runtime query rather than `query!` so the tier allowlist can be bound
|
||||||
|
// without regenerating the offline metadata on a machine with no database.
|
||||||
|
let row = sqlx::query(
|
||||||
"UPDATE topology_runs
|
"UPDATE topology_runs
|
||||||
SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now()
|
SET status = 'running', started_at = COALESCE(started_at, now()), updated_at = now()
|
||||||
WHERE id = (
|
WHERE id = (
|
||||||
SELECT id FROM topology_runs
|
SELECT id FROM topology_runs
|
||||||
WHERE status = 'queued'
|
WHERE status = 'queued'
|
||||||
|
-- Defence in depth. Even if a self-driven row somehow reaches
|
||||||
|
-- `queued`, the worker must not adopt a job it cannot execute:
|
||||||
|
-- doing so is what turned a live microVM run into a
|
||||||
|
-- missing-or-invalid-graph failure.
|
||||||
|
AND tier = ANY($1)
|
||||||
ORDER BY created_at
|
ORDER BY created_at
|
||||||
FOR UPDATE SKIP LOCKED
|
FOR UPDATE SKIP LOCKED
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
)
|
)
|
||||||
RETURNING id, workspace_id, task, graph, checkpoint, last_event_id, tier",
|
RETURNING id, workspace_id, task, graph, checkpoint, last_event_id, tier",
|
||||||
)
|
)
|
||||||
|
.bind(worker_driven())
|
||||||
.fetch_optional(pool)
|
.fetch_optional(pool)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(row.map(|r| ClaimedTopologyRun {
|
Ok(row.map(|r| ClaimedTopologyRun {
|
||||||
id: r.id,
|
id: r.get("id"),
|
||||||
workspace_id: r.workspace_id,
|
workspace_id: r.get("workspace_id"),
|
||||||
task: r.task,
|
task: r.get("task"),
|
||||||
graph: r.graph,
|
graph: r.get("graph"),
|
||||||
checkpoint: r.checkpoint,
|
checkpoint: r.get("checkpoint"),
|
||||||
last_event_id: r.last_event_id,
|
last_event_id: r.get("last_event_id"),
|
||||||
tier: r.tier,
|
tier: r.get("tier"),
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -325,12 +360,19 @@ pub async fn current_status(pool: &PgPool, id: Uuid) -> Result<Option<String>, D
|
|||||||
/// touch within `older_than_secs`). The next claim resumes them from checkpoint.
|
/// touch within `older_than_secs`). The next claim resumes them from checkpoint.
|
||||||
/// Returns how many were requeued.
|
/// Returns how many were requeued.
|
||||||
pub async fn requeue_stale(pool: &PgPool, older_than_secs: f64) -> Result<u64, DbError> {
|
pub async fn requeue_stale(pool: &PgPool, older_than_secs: f64) -> Result<u64, DbError> {
|
||||||
let result = sqlx::query!(
|
let result = sqlx::query(
|
||||||
"UPDATE topology_runs
|
"UPDATE topology_runs
|
||||||
SET status = 'queued', updated_at = now()
|
SET status = 'queued', updated_at = now()
|
||||||
WHERE status = 'running' AND updated_at < now() - make_interval(secs => $1)",
|
WHERE status = 'running'
|
||||||
older_than_secs,
|
AND updated_at < now() - make_interval(secs => $1)
|
||||||
|
-- Only jobs the WORKER drives. A self-driven run (microvm, session) is
|
||||||
|
-- owned by its own task for its whole life and never touches
|
||||||
|
-- `updated_at`, so without this every one of them looked stale after
|
||||||
|
-- three minutes and was requeued out from under a live VM.
|
||||||
|
AND tier = ANY($2)",
|
||||||
)
|
)
|
||||||
|
.bind(older_than_secs)
|
||||||
|
.bind(worker_driven())
|
||||||
.execute(pool)
|
.execute(pool)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(result.rows_affected())
|
Ok(result.rows_affected())
|
||||||
|
|||||||
Reference in New Issue
Block a user