Dual-write phase and lease in task status transitions (B2 Phase 1 step 3/5)
This commit is contained in:
1 parent
8cb227f455
commit
2656684392
4 files changed
+100
-1
No files matched your search
@@ -386,6 +386,15 @@ export class AppDatabaseBase {
|
||||
).run(ts, ts);
|
||||
}
|
||||
this.db.prepare("update tasks set active_run_id = null where active_run_id in (select id from runs where status = 'interrupted')").run();
|
||||
// B2 Phase 1 step 3/5 — dual-write: drop shadow leases whose run
|
||||
// has been interrupted. Uses the freshly finished_at timestamp to
|
||||
// scope the deletion to this sweep's run set, matching the legacy
|
||||
// path's "just-interrupted" semantics.
|
||||
this.db
|
||||
.prepare(
|
||||
"DELETE FROM task_leases WHERE run_id IN (SELECT id FROM runs WHERE status = 'interrupted' AND finished_at = ?)",
|
||||
)
|
||||
.run(ts);
|
||||
this.db.prepare("update agents set status = 'idle', updated_at = ? where status = 'running'").run(ts);
|
||||
this.releaseAllWorkspaceLocks();
|
||||
});
|
||||
@@ -401,6 +410,12 @@ export class AppDatabaseBase {
|
||||
const ts = nowIso();
|
||||
for (const task of stale) {
|
||||
this.db.prepare(`UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, status = 'todo', updated_at = ? WHERE id = ?`).run(ts, task.id);
|
||||
// B2 Phase 1 step 3/5 — dual-write: drop the shadow lease whose
|
||||
// legacy counterpart was just torn down. Matches the legacy
|
||||
// path's "clear everything for this stuck task" semantics; the
|
||||
// trigger that mirrored status → phase has already advanced
|
||||
// `tasks.phase` back to `todo` so the new surface stays coherent.
|
||||
this.db.prepare(`DELETE FROM task_leases WHERE task_id = ?`).run(task.id);
|
||||
}
|
||||
return stale.map((t) => ({ taskId: t.id, companyId: t.companyId, title: t.title }));
|
||||
}
|
||||
|
||||
@@ -145,6 +145,11 @@ export function buildRunMethods(
|
||||
getDb()
|
||||
.prepare("UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?")
|
||||
.run(ts, run.taskId, input.runId);
|
||||
// B2 Phase 1 step 3/5 — dual-write: release the shadow lease
|
||||
// for this run at the same instant the legacy lock is cleared.
|
||||
getDb()
|
||||
.prepare("DELETE FROM task_leases WHERE task_id = ? AND run_id = ?")
|
||||
.run(run.taskId, input.runId);
|
||||
if (runNeverStarted) {
|
||||
// Roll the task back to `todo` when no *other* live run claims
|
||||
// it. Invariant: `task.status = 'in_progress'` ⇒ a queued or
|
||||
@@ -208,6 +213,11 @@ export function buildRunMethods(
|
||||
getDb()
|
||||
.prepare("UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?")
|
||||
.run(ts, taskId, runId);
|
||||
// B2 Phase 1 step 3/5 — dual-write: drop the shadow lease when
|
||||
// the legacy cancellation clears checkout_run_id.
|
||||
getDb()
|
||||
.prepare("DELETE FROM task_leases WHERE task_id = ? AND run_id = ?")
|
||||
.run(taskId, runId);
|
||||
}
|
||||
if (workspaceId) {
|
||||
getDb()
|
||||
|
||||
@@ -136,6 +136,17 @@ export function buildTaskCrudMethods(
|
||||
"UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?",
|
||||
)
|
||||
.run(ts, run.taskId, run.id);
|
||||
// B2 Phase 1 step 3/5 — dual-write: cascade to the shadow
|
||||
// lease row so a task-delete does not leave an orphan lease
|
||||
// behind. The FK on `task_leases.task_id` would also cascade
|
||||
// via the DELETE at the end of this transaction; we clear
|
||||
// here explicitly so the intermediate state (task still
|
||||
// present, lease already gone) matches what `releaseTaskLock`
|
||||
// would have done if the delete had gone through the normal
|
||||
// lifecycle path.
|
||||
getDb()
|
||||
.prepare("DELETE FROM task_leases WHERE task_id = ? AND run_id = ?")
|
||||
.run(run.taskId, run.id);
|
||||
if (run.workspaceId) {
|
||||
getDb()
|
||||
.prepare(
|
||||
|
||||
@@ -8,6 +8,38 @@ import {
|
||||
} from "./helpers";
|
||||
import { chooseBestTaskAssignee } from "./task-assignment";
|
||||
|
||||
/**
|
||||
* Default lease TTL used when the dual-write path seeds `task_leases`
|
||||
* from the legacy `checkout_run_id` / `execution_locked_at` columns.
|
||||
* Matches the stale-lock timeout in database-base so the sweep logic
|
||||
* stays in sync across the shadow tables.
|
||||
*/
|
||||
const DUAL_WRITE_LEASE_TTL_MS = 15 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* Mirror a successful legacy checkout onto the new `task_leases` row.
|
||||
*
|
||||
* Called from `checkoutTask` immediately after the legacy CAS succeeds.
|
||||
* `INSERT OR REPLACE` is safe here because the preceding SQL CAS
|
||||
* guarded on `checkout_run_id IS NULL` — any live lease row at this
|
||||
* point refers to the same run (replay under retry) or is itself stale
|
||||
* (recoverStaleLocks has not caught up yet). The new expires_at resets
|
||||
* the stale-lock sweep deadline either way.
|
||||
*/
|
||||
function writeCheckoutLease(
|
||||
db: DatabaseSync,
|
||||
taskId: string,
|
||||
runId: string,
|
||||
ts: string,
|
||||
): void {
|
||||
const expiresAt = new Date(Date.now() + DUAL_WRITE_LEASE_TTL_MS).toISOString();
|
||||
db.prepare(
|
||||
`INSERT OR REPLACE INTO task_leases
|
||||
(task_id, run_id, acquired_at, expires_at, checkout_run_id)
|
||||
VALUES (?, ?, ?, ?, ?)`,
|
||||
).run(taskId, runId, ts, expiresAt, runId);
|
||||
}
|
||||
|
||||
export function buildTaskStatusMethods(
|
||||
getDb: () => DatabaseSync,
|
||||
readAll: ReadAll,
|
||||
@@ -173,6 +205,7 @@ export function buildTaskStatusMethods(
|
||||
// The `active_run_id IS NULL` guard preserves the original CAS
|
||||
// contract for the busy case: a task whose active_run_id still
|
||||
// points at a live run is refused here, exactly as before.
|
||||
|
||||
const result = getDb()
|
||||
.prepare(
|
||||
`UPDATE tasks
|
||||
@@ -185,24 +218,54 @@ export function buildTaskStatusMethods(
|
||||
)`,
|
||||
)
|
||||
.run(runId, ts, ts, taskId);
|
||||
if (result.changes > 0) {
|
||||
// B2 Phase 1 step 3/5 — dual-write the new lease shadow. The
|
||||
// status → phase trigger from v50 already mirrored the phase
|
||||
// column by this point; the lease table still needs an explicit
|
||||
// write. Same-run re-acquires overwrite in place, different-run
|
||||
// acquires would return null (and should not happen here
|
||||
// because the SQL CAS above already gated on `checkout_run_id
|
||||
// IS NULL`). Wrapped in a best-effort try to keep the new path
|
||||
// non-blocking for step 3 — the observability pass in step 4/5
|
||||
// surfaces any divergence.
|
||||
writeCheckoutLease(getDb(), taskId, runId, ts);
|
||||
}
|
||||
return result.changes > 0;
|
||||
},
|
||||
|
||||
releaseTaskLock(taskId: string, runId: string): void {
|
||||
const ts = nowIso();
|
||||
|
||||
getDb()
|
||||
.prepare(
|
||||
"UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?",
|
||||
)
|
||||
.run(nowIso(), taskId, runId);
|
||||
.run(ts, taskId, runId);
|
||||
// B2 Phase 1 step 3/5 — dual-write: drop the shadow lease too.
|
||||
getDb()
|
||||
.prepare("DELETE FROM task_leases WHERE task_id = ? AND run_id = ?")
|
||||
.run(taskId, runId);
|
||||
},
|
||||
|
||||
renewTaskLease(taskId: string, runId: string): boolean {
|
||||
const ts = nowIso();
|
||||
|
||||
const result = getDb()
|
||||
.prepare(
|
||||
"UPDATE tasks SET execution_locked_at = ?, updated_at = ? WHERE id = ? AND checkout_run_id = ?",
|
||||
)
|
||||
.run(ts, ts, taskId, runId);
|
||||
if (result.changes > 0) {
|
||||
// Shadow renewal: push `expires_at` forward in lockstep with
|
||||
// `execution_locked_at` so the stale-lock sweep over the new
|
||||
// table stays consistent with the legacy sweep.
|
||||
const expiresAt = new Date(Date.now() + DUAL_WRITE_LEASE_TTL_MS).toISOString();
|
||||
getDb()
|
||||
.prepare(
|
||||
`UPDATE task_leases SET expires_at = ? WHERE task_id = ? AND run_id = ?`,
|
||||
)
|
||||
.run(expiresAt, taskId, runId);
|
||||
}
|
||||
return result.changes > 0;
|
||||
},
|
||||
|
||||
|
||||
Reference in new issue
Block a user