diff --git a/doc/plans/2026-10-30-routine-status-sleep.md b/doc/plans/2026-10-30-routine-status-sleep.md index 271d0b544a7..e852bf94861 100644 --- a/doc/plans/2026-10-30-routine-status-sleep.md +++ b/doc/plans/2026-10-30-routine-status-sleep.md @@ -18,7 +18,9 @@ and belong to a routine in that company. This is the existing routine status owner contract. Origin labels can change when a plugin declaration changes or is removed; retained reciprocal links preserve historical execution custody. Repair locks the routine and task and rechecks the linked invocation before writing. -It preserves skipped/coalesced history and unrelated dispatch failures. A task +Direct route hooks also lock the execution task and linked invocation in their +own transaction, nested under caller transactions when present. An older hook +cannot write after a newer task status has committed. It preserves skipped/coalesced history and unrelated dispatch failures. A task that resumes clears an execution-task transient failure through the same owner used by native completion and issue routes. diff --git a/server/src/__tests__/routine-execution-status-work.test.ts b/server/src/__tests__/routine-execution-status-work.test.ts index 53d7e5c0225..e6bd818852c 100644 --- a/server/src/__tests__/routine-execution-status-work.test.ts +++ b/server/src/__tests__/routine-execution-status-work.test.ts @@ -174,6 +174,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, otherRun.linkedIssueId!); }); await projected(otherRun.id, "completed"); await quiet(); expect(errors).toEqual([]); }); + it("serializes a direct status hook with later task changes before projecting the routine", async () => { + const f = await fixture(), run = await dispatch(f); await start(); await quiet(); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!)); + let entered!: () => void, release!: () => void; + const enteredGate = new Promise(resolve => { entered = resolve; }); + const releaseGate = new Promise(resolve => { release = resolve; }); + const holding = db.transaction(async tx => { + await tx.select().from(routineRuns).where(eq(routineRuns.id, run.id)).for("update"); + entered(); await releaseGate; + }); + let direct: Promise | undefined, changing: Promise | undefined; + let directPid: number | undefined; + try { + await enteredGate; direct = f.service.syncRunStatusForIssue(run.linkedIssueId!); + await vi.waitFor(async () => { + const rows = await db.execute(sql`select pid from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and (query like 'update "routine_runs"%' or query like 'select %"routine_runs"%')`); + expect(rows).toHaveLength(1); directPid = Number(rows[0]!.pid); + }, { timeout: 5000 }); + await expect(db.transaction(tx => tx.select({ id: issues.id }).from(issues) + .where(eq(issues.id, run.linkedIssueId!)).for("update", { noWait: true }))) + .rejects.toMatchObject({ cause: expect.objectContaining({ code: "55P03" }) }); + changing = issueService(db).update(run.linkedIssueId!, { status: "blocked" }); + await vi.waitFor(async () => { + const rows = await db.execute(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and ${directPid}::int = any(pg_blocking_pids(pid))`); + expect(Number(rows[0]?.count)).toBeGreaterThan(0); + }, { timeout: 5000 }); + release(); await holding; await direct; await changing; + await projected(run.id, "failed"); await quiet(); expect(errors).toEqual([]); + } finally { release(); await holding; await direct; await changing; } + }); it("waits for the real outer transaction and forgets a rolled-back nested status write", async () => { const f = await fixture(), run = await dispatch(f); await start(); await quiet(); let release!: () => void, entered!: () => void; diff --git a/server/src/services/routine-run-status.ts b/server/src/services/routine-run-status.ts index 2a182def8f9..46a2bae4cf5 100644 --- a/server/src/services/routine-run-status.ts +++ b/server/src/services/routine-run-status.ts @@ -51,6 +51,10 @@ export async function finalizeRoutineRun(executor: Db, runId: string, patch: Par } export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string) { + return executor.transaction(tx => syncLockedRoutineRunStatusForIssue(tx as unknown as Db, issueId)); +} + +async function syncLockedRoutineRunStatusForIssue(executor: Db, issueId: string) { const issue = await executor .select({ id: issues.id, @@ -60,7 +64,8 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string originRunId: issues.originRunId, }) .from(issues) - .where(eq(issues.id, issueId)) + .where(and(eq(issues.id, issueId), eq(issues.originKind, "routine_execution"))) + .for("update") .then((rows) => rows[0] ?? null); if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; const run = await executor @@ -75,6 +80,7 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId))) .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId), eq(routineRuns.linkedIssueId, issue.id))) + .for("update", { of: routineRuns }) .then((rows) => rows[0] ?? null); if (!run || ["received", "skipped", "coalesced"].includes(run.status)) return null; if (issue.status === "done") {