diff --git a/server/internal/handler/workspace_delete_fence_test.go b/server/internal/handler/workspace_delete_fence_test.go new file mode 100644 index 000000000..b3b7018fc --- /dev/null +++ b/server/internal/handler/workspace_delete_fence_test.go @@ -0,0 +1,406 @@ +package handler + +import ( + "context" + "errors" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + db "github.com/multica-ai/multica/server/pkg/db/generated" +) + +// enqueueViaRealQuery creates a task through the generated CreateAgentTask query — +// the actual production enqueue path, fence included. Raw INSERTs in tests would +// bypass the fence, which lives in the application SQL rather than in a trigger. +func enqueueViaRealQuery(ctx context.Context, q *db.Queries, agentID, runtimeID, issueID string) error { + params := db.CreateAgentTaskParams{ + AgentID: parseUUID(agentID), + RuntimeID: parseUUID(runtimeID), + Priority: 0, + } + if issueID != "" { + params.IssueID = parseUUID(issueID) + } + _, err := q.CreateAgentTask(ctx, params) + return err +} + +// TestTaskOwnershipWritesCallTheFence is the guard that keeps the fence from being +// a convention. Every statement that inserts an agent_task_queue row, or assigns +// one of its ownership columns, has to call lock_task_owner_workspaces — a new +// write path that forgets it would silently reopen the insert-after-sweep window +// that MUL-5999's teardown depends on being closed. +// +// This is the trade for not using a row trigger: the write path names its own +// fence, and CI checks that it did. +func TestTaskOwnershipWritesCallTheFence(t *testing.T) { + queryDir := filepath.Join("..", "..", "pkg", "db", "queries") + entries, err := os.ReadDir(queryDir) + if err != nil { + t.Fatalf("read query dir: %v", err) + } + + nameRe := regexp.MustCompile(`(?m)^-- name: (\S+)`) + // Go's regexp has no lookahead, so the SET clause is isolated first and its + // ownership assignments are inspected individually. Clearing runtime_id cannot + // move a task into another workspace, so `= NULL` is exempt; teardown's own + // detach only touches parent_task_id. + setClauseRe := regexp.MustCompile(`(?is)\bSET\b(.*?)(?:\bWHERE\b|\bFROM\b|\bRETURNING\b|;)`) + ownerAssignRe := regexp.MustCompile(`(?i)\b(agent_id|issue_id|runtime_id)\s*=\s*(\S+)`) + + assignsOwner := func(code string) bool { + for _, clause := range setClauseRe.FindAllStringSubmatch(code, -1) { + for _, assign := range ownerAssignRe.FindAllStringSubmatch(clause[1], -1) { + if !strings.EqualFold(strings.TrimSuffix(assign[2], ","), "NULL") { + return true + } + } + } + return false + } + + checked := 0 + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") { + continue + } + body, err := os.ReadFile(filepath.Join(queryDir, entry.Name())) + if err != nil { + t.Fatalf("read %s: %v", entry.Name(), err) + } + text := string(body) + matches := nameRe.FindAllStringSubmatchIndex(text, -1) + for i, m := range matches { + stop := len(text) + if i+1 < len(matches) { + stop = matches[i+1][0] + } + stmt := text[m[0]:stop] + name := text[m[2]:m[3]] + // Strip comment lines so prose cannot satisfy or trip the check. + var sqlOnly strings.Builder + for _, line := range strings.Split(stmt, "\n") { + if strings.HasPrefix(strings.TrimSpace(line), "--") { + continue + } + sqlOnly.WriteString(line) + sqlOnly.WriteString("\n") + } + code := sqlOnly.String() + + writesOwnership := strings.Contains(code, "INSERT INTO agent_task_queue") || + (strings.Contains(code, "UPDATE agent_task_queue") && assignsOwner(code)) + if !writesOwnership { + continue + } + checked++ + if !strings.Contains(code, "lock_task_owner_workspaces(") { + t.Errorf("%s: %s writes agent_task_queue ownership without calling "+ + "lock_task_owner_workspaces; that reopens the workspace-delete race "+ + "(see migration 284)", entry.Name(), name) + } + } + } + if checked < 9 { + t.Errorf("only %d ownership-writing statements found; the detector probably stopped matching", checked) + } +} + +// TestTaskWriteFence_BlocksOnWorkspaceLockAlone shows the fence, and only the +// fence, standing between a concurrent enqueue and a workspace being torn down. +// +// The holder locks ONLY the workspace row — no agent, issue or runtime row — so no +// foreign key on agent_task_queue has anything to block on: none of them +// references workspace. Anything that blocks here is the fence taking FOR KEY +// SHARE on the owners' workspace row from inside the writer's own statement. +func TestTaskWriteFence_BlocksOnWorkspaceLockAlone(t *testing.T) { + if testPool == nil { + t.Skip("database not available") + } + ctx := context.Background() + f := newWorkspaceDeletePathFixture(t, "fenceonly") + + holder, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("begin holder: %v", err) + } + defer holder.Rollback(ctx) + // Exactly what LockWorkspaceForDelete does, and nothing else. + if _, err := holder.Exec(ctx, `SELECT id FROM workspace WHERE id = $1 FOR UPDATE`, f.victimID); err != nil { + t.Fatalf("lock workspace row: %v", err) + } + + blockedEnqueue := func(name, agentID, runtimeID, issueID string) { + t.Helper() + writer, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("%s: begin: %v", name, err) + } + defer writer.Rollback(ctx) + if _, err := writer.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { + t.Fatalf("%s: set lock_timeout: %v", name, err) + } + err = enqueueViaRealQuery(ctx, testHandler.Queries.WithTx(writer), agentID, runtimeID, issueID) + if err == nil { + t.Errorf("%s: enqueue committed while the workspace row was locked; the fence is not in the write path", name) + return + } + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "55P03" { + t.Errorf("%s: got %v, want lock_not_available (55P03)", name, err) + } + } + + blockedEnqueue("enqueue via the victim's agent", f.victimAgent, f.neighbourRuntime, f.neighbourIssue) + blockedEnqueue("enqueue via the victim's runtime", f.neighbourAgent, f.victimRuntime, f.neighbourIssue) + blockedEnqueue("enqueue via the victim's issue", f.neighbourAgent, f.neighbourRuntime, f.victimIssue) + + // Reassignment onto the victim's runtime is fenced too. + reassign, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("begin reassign: %v", err) + } + defer reassign.Rollback(ctx) + if _, err := reassign.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { + t.Fatalf("reassign: set lock_timeout: %v", err) + } + _, err = testHandler.Queries.WithTx(reassign).ReassignTasksToRuntime(ctx, db.ReassignTasksToRuntimeParams{ + NewRuntimeID: parseUUID(f.victimRuntime), + OldRuntimeID: parseUUID(f.neighbourRuntime), + }) + var pgErr *pgconn.PgError + if err == nil || !errors.As(err, &pgErr) || pgErr.Code != "55P03" { + t.Errorf("reassignment onto the victim's runtime: got %v, want lock_not_available (55P03)", err) + } + + // A workspace nobody is deleting is unaffected. + unrelated, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("begin unrelated: %v", err) + } + defer unrelated.Rollback(ctx) + if _, err := unrelated.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { + t.Fatalf("unrelated: set lock_timeout: %v", err) + } + if err := enqueueViaRealQuery(ctx, testHandler.Queries.WithTx(unrelated), + f.neighbourAgent, f.neighbourRuntime, f.neighbourIssue); err != nil { + t.Errorf("neighbour-only enqueue was blocked by the victim's workspace lock: %v", err) + } + + // Status-only updates are the hot path and must not touch the fence at all. + statusUpdate, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("begin status update: %v", err) + } + defer statusUpdate.Rollback(ctx) + if _, err := statusUpdate.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { + t.Fatalf("status update: set lock_timeout: %v", err) + } + if _, err := statusUpdate.Exec(ctx, + `UPDATE agent_task_queue SET status = 'failed' WHERE id = $1`, f.taskViaAgent); err != nil { + t.Errorf("status-only update was blocked: %v", err) + } +} + +// TestTaskWriteFence_QueuedWriterCannotLandAfterTeardownCommits is the race the +// review asked for, run as an actual race: a writer queues behind the workspace +// lock while teardown is in flight, teardown commits, and the writer must then be +// unable to put a row in the table. +// +// It also discriminates the fence from the foreign keys. The writer's failure is +// "no row written" — CreateAgentTask is a :one, so it comes back as +// pgx.ErrNoRows because the fence's predicate was false. An FK doing the work +// instead would surface as SQLSTATE 23503, which the assertion rejects. +func TestTaskWriteFence_QueuedWriterCannotLandAfterTeardownCommits(t *testing.T) { + if testPool == nil { + t.Skip("database not available") + } + ctx := context.Background() + f := newWorkspaceDeletePathFixture(t, "queuedwriter") + + teardown, err := testPool.Begin(ctx) + if err != nil { + t.Fatalf("begin teardown: %v", err) + } + committed := false + defer func() { + if !committed { + _ = teardown.Rollback(ctx) + } + }() + + // The teardown transaction, in the order the handler does it. + if _, err := teardown.Exec(ctx, `SELECT id FROM workspace WHERE id = $1 FOR UPDATE`, f.victimID); err != nil { + t.Fatalf("lock workspace: %v", err) + } + + var wg sync.WaitGroup + var writerErr error + wg.Add(1) + go func() { + defer wg.Done() + // No lock_timeout: this writer is supposed to wait for teardown, which is + // what a real enqueue arriving mid-teardown does. + writerErr = enqueueViaRealQuery(context.Background(), testHandler.Queries, + f.victimAgent, f.victimRuntime, f.victimIssue) + }() + + // Give the writer time to reach the lock and block on it. + time.Sleep(300 * time.Millisecond) + + if _, err := teardown.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, f.victimID); err != nil { + t.Fatalf("delete workspace: %v", err) + } + if err := teardown.Commit(ctx); err != nil { + t.Fatalf("commit teardown: %v", err) + } + committed = true + + wg.Wait() + + if writerErr == nil { + t.Fatal("the queued enqueue landed a task after teardown committed; the fence did not hold") + } + if !errors.Is(writerErr, pgx.ErrNoRows) { + var pgErr *pgconn.PgError + if errors.As(writerErr, &pgErr) && pgErr.Code == "23503" { + t.Fatalf("the write was rejected by a foreign key (%s), not by the fence; "+ + "the fence must hold on its own once the legacy FKs go away", pgErr.Code) + } + t.Fatalf("writer error = %v, want pgx.ErrNoRows from the fence returning false", writerErr) + } + + var leftover int + if err := testPool.QueryRow(ctx, ` +SELECT count(*) FROM agent_task_queue WHERE agent_id = $1 OR runtime_id = $2 OR issue_id = $3 +`, f.victimAgent, f.victimRuntime, f.victimIssue).Scan(&leftover); err != nil { + t.Fatalf("count leftovers: %v", err) + } + if leftover != 0 { + t.Errorf("%d task rows are left pointing at the deleted workspace's owners", leftover) + } +} + +// TestTaskWriteFence_LocksWorkspacesInIDOrder covers the deadlock surface the +// review flagged: a task may reference owners in different workspaces, so the +// fence can lock more than one workspace row, and two writers touching the same +// pair from opposite directions must not be able to take them in opposite orders. +// +// The fence sorts by workspace id, so both writers queue on the lower id first. +// The test runs the two mirrored writes concurrently and requires that both +// succeed — a deadlock would surface as 40P01 on one of them. +func TestTaskWriteFence_LocksWorkspacesInIDOrder(t *testing.T) { + if testPool == nil { + t.Skip("database not available") + } + ctx := context.Background() + f := newWorkspaceDeletePathFixture(t, "lockorder") + + const rounds = 25 + + // One pending task per (issue, agent) is enforced by a partial unique index, so + // each round needs its own issue pair; otherwise the rounds collide on that + // index instead of exercising the lock order. + victimIssues := make([]string, rounds) + neighbourIssues := make([]string, rounds) + for i := 0; i < rounds; i++ { + if err := testPool.QueryRow(ctx, ` +INSERT INTO issue (workspace_id, title, creator_type, creator_id, number) +VALUES ($1, $2, 'member', $3, $4) RETURNING id +`, f.victimID, "lock-order victim "+strconv.Itoa(i), testUserID, i+100).Scan(&victimIssues[i]); err != nil { + t.Fatalf("seed victim issue %d: %v", i, err) + } + if err := testPool.QueryRow(ctx, ` +INSERT INTO issue (workspace_id, title, creator_type, creator_id, number) +VALUES ($1, $2, 'member', $3, $4) RETURNING id +`, f.neighbourID, "lock-order neighbour "+strconv.Itoa(i), testUserID, i+100).Scan(&neighbourIssues[i]); err != nil { + t.Fatalf("seed neighbour issue %d: %v", i, err) + } + } + + var wg sync.WaitGroup + errs := make([]error, 2*rounds) + for i := 0; i < rounds; i++ { + wg.Add(2) + // Owners drawn from both workspaces, in mirrored roles. + go func(i int) { + defer wg.Done() + errs[2*i] = enqueueViaRealQuery(ctx, testHandler.Queries, + f.victimAgent, f.neighbourRuntime, neighbourIssues[i]) + }(i) + go func(i int) { + defer wg.Done() + errs[2*i+1] = enqueueViaRealQuery(ctx, testHandler.Queries, + f.neighbourAgent, f.victimRuntime, victimIssues[i]) + }(i) + } + wg.Wait() + + for i, err := range errs { + if err == nil { + continue + } + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "40P01" { + t.Fatalf("write %d deadlocked (40P01): the fence is not locking workspaces in a fixed order", i) + } + t.Errorf("write %d failed: %v", i, err) + } + + if _, err := testPool.Exec(ctx, ` +DELETE FROM agent_task_queue +WHERE (agent_id = $1 AND runtime_id = $2) OR (agent_id = $3 AND runtime_id = $4) +`, f.victimAgent, f.neighbourRuntime, f.neighbourAgent, f.victimRuntime); err != nil { + t.Fatalf("clean up: %v", err) + } +} + +// TestLockTaskOwnerWorkspaces_RejectsUnresolvableOwner pins the hole the review +// found in the first version of this fence: counting only the owners that still +// resolve made a deleted owner invisible, and the statement fell through to the +// foreign key. A reference that does not resolve must make the predicate false on +// its own. +func TestLockTaskOwnerWorkspaces_RejectsUnresolvableOwner(t *testing.T) { + if testPool == nil { + t.Skip("database not available") + } + ctx := context.Background() + f := newWorkspaceDeletePathFixture(t, "unresolvable") + missing := "00000000-0000-0000-0000-0000000000ff" + + for name, args := range map[string][3]any{ + "missing agent": {missing, nil, nil}, + "missing issue": {nil, missing, nil}, + "missing runtime": {nil, nil, missing}, + "one of three owners missing": {f.victimAgent, missing, f.victimRuntime}, + } { + var ok bool + if err := testPool.QueryRow(ctx, + `SELECT lock_task_owner_workspaces($1, $2, $3)`, args[0], args[1], args[2]).Scan(&ok); err != nil { + t.Fatalf("%s: %v", name, err) + } + if ok { + t.Errorf("%s: fence returned true; the write would fall through to the foreign key", name) + } + } + + var ok bool + if err := testPool.QueryRow(ctx, `SELECT lock_task_owner_workspaces($1, $2, $3)`, + f.victimAgent, f.victimIssue, f.victimRuntime).Scan(&ok); err != nil { + t.Fatalf("all owners present: %v", err) + } + if !ok { + t.Error("fence returned false for a fully resolvable row") + } +} + +var _ = pgtype.UUID{} diff --git a/server/internal/handler/workspace_delete_task_discovery_test.go b/server/internal/handler/workspace_delete_task_discovery_test.go index e4681f12c..dee38fe87 100644 --- a/server/internal/handler/workspace_delete_task_discovery_test.go +++ b/server/internal/handler/workspace_delete_task_discovery_test.go @@ -532,134 +532,6 @@ SELECT EXISTS ( } } -// TestTaskWriteFence_BlocksOnWorkspaceLockAlone is the regression test for the -// window that a re-scan cannot close: a task inserted after teardown's final empty -// scan but before COMMIT. -// -// It also isolates the fence from the foreign keys. The holder locks ONLY the -// workspace row — no agent, issue or runtime row — so the FKs on -// agent_task_queue have nothing to block on: none of them references workspace. -// Anything that blocks here is migration 284's BEFORE trigger taking FOR KEY SHARE -// on the owners' workspace row, which is exactly the guarantee that has to survive -// the legacy FKs being removed. -func TestTaskWriteFence_BlocksOnWorkspaceLockAlone(t *testing.T) { - if testPool == nil { - t.Skip("database not available") - } - ctx := context.Background() - f := newWorkspaceDeletePathFixture(t, "fenceonly") - - holder, err := testPool.Begin(ctx) - if err != nil { - t.Fatalf("begin holder: %v", err) - } - defer holder.Rollback(ctx) - // Exactly what LockWorkspaceForDelete does, and nothing else. - if _, err := holder.Exec(ctx, `SELECT id FROM workspace WHERE id = $1 FOR UPDATE`, f.victimID); err != nil { - t.Fatalf("lock workspace row: %v", err) - } - - expectBlocked := func(name, sql string, args ...any) { - t.Helper() - writer, err := testPool.Begin(ctx) - if err != nil { - t.Fatalf("%s: begin: %v", name, err) - } - defer writer.Rollback(ctx) - if _, err := writer.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { - t.Fatalf("%s: set lock_timeout: %v", name, err) - } - _, err = writer.Exec(ctx, sql, args...) - if err == nil { - t.Errorf("%s: committed while only the workspace row was locked; the fence is not in the write path", name) - return - } - var pgErr *pgconn.PgError - if !errors.As(err, &pgErr) || pgErr.Code != "55P03" { - t.Errorf("%s: got %v, want lock_not_available (55P03)", name, err) - } - } - - expectBlocked("insert via the victim's agent", ` -INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, completed_at) -VALUES ($1, $2, $3, 'completed', now()) -`, f.victimAgent, f.neighbourIssue, f.neighbourRuntime) - - expectBlocked("insert via the victim's issue", ` -INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, completed_at) -VALUES ($1, $2, $3, 'completed', now()) -`, f.neighbourAgent, f.victimIssue, f.neighbourRuntime) - - expectBlocked("insert via the victim's runtime", ` -INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, completed_at) -VALUES ($1, $2, $3, 'completed', now()) -`, f.neighbourAgent, f.neighbourIssue, f.victimRuntime) - - expectBlocked("reassignment onto the victim's runtime", ` -UPDATE agent_task_queue SET runtime_id = $1 WHERE id = $2 -`, f.victimRuntime, f.neighbourTask) - - // A write that touches none of the locked workspace's owners must still pass, - // so the fence is not a global stop-the-world. - unrelated, err := testPool.Begin(ctx) - if err != nil { - t.Fatalf("begin unrelated: %v", err) - } - defer unrelated.Rollback(ctx) - if _, err := unrelated.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { - t.Fatalf("unrelated: set lock_timeout: %v", err) - } - if _, err := unrelated.Exec(ctx, ` -INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, completed_at) -VALUES ($1, $2, $3, 'completed', now()) -`, f.neighbourAgent, f.neighbourIssue, f.neighbourRuntime); err != nil { - t.Errorf("neighbour-only insert was blocked by the victim's workspace lock: %v", err) - } - - // Status-only updates are the hot path and must not pay the fence at all: - // they still work while the workspace row is locked. - statusUpdate, err := testPool.Begin(ctx) - if err != nil { - t.Fatalf("begin status update: %v", err) - } - defer statusUpdate.Rollback(ctx) - if _, err := statusUpdate.Exec(ctx, "SET LOCAL lock_timeout = 750"); err != nil { - t.Fatalf("status update: set lock_timeout: %v", err) - } - if _, err := statusUpdate.Exec(ctx, - `UPDATE agent_task_queue SET status = 'failed' WHERE id = $1`, f.taskViaAgent); err != nil { - t.Errorf("status-only update was blocked by the fence: %v", err) - } -} - -// TestTaskWriteFence_FailsClosedAfterWorkspaceIsGone covers the other side of the -// fence: once teardown has committed, a task write that was queued up behind it -// must fail rather than resurrect rows into a deleted tenant. -func TestTaskWriteFence_FailsClosedAfterWorkspaceIsGone(t *testing.T) { - if testPool == nil { - t.Skip("database not available") - } - ctx := context.Background() - f := newWorkspaceDeletePathFixture(t, "failclosed") - - // Keep an agent row alive whose workspace is gone, which is the state a - // late writer would see if it only had the FKs to go on. Deleting the - // workspace row directly (not via the handler) is the point: the FK from - // agent.workspace_id is ON DELETE CASCADE, so this also removes the agent — - // so the fence is exercised through the issue path, whose row we keep. - if _, err := testPool.Exec(ctx, `DELETE FROM workspace WHERE id = $1`, f.victimID); err != nil { - t.Fatalf("delete workspace: %v", err) - } - - _, err := testPool.Exec(ctx, ` -INSERT INTO agent_task_queue (agent_id, issue_id, runtime_id, status, completed_at) -VALUES ($1, $2, $3, 'completed', now()) -`, f.neighbourAgent, f.neighbourIssue, f.neighbourRuntime) - if err != nil { - t.Fatalf("neighbour insert should still work after an unrelated workspace is deleted: %v", err) - } -} - // TestDeleteWorkspaceTasks_FencesConcurrentEnqueueAndReassignment covers the two // races a per-batch sweep would otherwise lose: // diff --git a/server/migrations/284_task_owner_workspace_fence.down.sql b/server/migrations/284_task_owner_workspace_fence.down.sql new file mode 100644 index 000000000..f6db204a6 --- /dev/null +++ b/server/migrations/284_task_owner_workspace_fence.down.sql @@ -0,0 +1 @@ +DROP FUNCTION IF EXISTS lock_task_owner_workspaces(uuid, uuid, uuid); diff --git a/server/migrations/284_task_owner_workspace_fence.up.sql b/server/migrations/284_task_owner_workspace_fence.up.sql new file mode 100644 index 000000000..6fd9b3626 --- /dev/null +++ b/server/migrations/284_task_owner_workspace_fence.up.sql @@ -0,0 +1,113 @@ +-- The write half of workspace teardown's concurrency contract (MUL-5999). +-- +-- THE WINDOW +-- +-- Teardown sweeps a workspace's tasks and then commits. Anything that inserts a +-- task — or moves an existing one onto this workspace's agent, issue or runtime — +-- between the last sweep and COMMIT would leave a task behind whose owners no +-- longer exist, together with its own leaf rows. Re-scanning before commit cannot +-- close that: the check and the commit are not atomic with respect to a +-- concurrent writer, so a row can always land after the final empty scan. +-- +-- WHAT THIS IS +-- +-- A predicate that every statement writing agent_task_queue ownership calls in +-- its own WHERE clause: +-- +-- INSERT INTO agent_task_queue (...) +-- SELECT ... WHERE lock_task_owner_workspaces(@agent_id, @issue_id, @runtime_id) +-- +-- It resolves the workspaces behind the row's agent, issue and runtime, locks +-- those workspace rows FOR KEY SHARE, and returns false unless every owner +-- reference resolved and every one of their workspaces is still there. FOR KEY +-- SHARE conflicts with the FOR UPDATE that LockWorkspaceForDelete holds for the +-- whole teardown transaction, so a task write against a workspace being torn down +-- blocks until teardown finishes and then either proceeds (teardown rolled back) +-- or writes no row (teardown committed). +-- +-- WHY A CALLED PREDICATE RATHER THAN A TRIGGER OR THE FOREIGN KEYS +-- +-- * Not the foreign keys. An INSERT does take FOR KEY SHARE on the rows it +-- references, which is why this window has been closed so far — but that is a +-- side effect of FK enforcement, and workspace_delete.sql documents the legacy +-- FKs as an expand-phase net a later contract removes. Cleanup correctness +-- must not be a hostage of that migration. +-- * Not a trigger. A row trigger is a database-layer relational constraint that +-- coordinates application behaviour behind the caller's back. This repo keeps +-- relationships in the application layer, so the write path names its own +-- fence, and `TestTaskOwnershipWritesCallTheFence` fails the build if a +-- statement that writes ownership forgets to. +-- * It must be in the writer's transaction. Several enqueue statements run as +-- single autocommit statements, so a separate "take the fence" statement +-- issued by the caller would be its own transaction and release the lock +-- before the insert landed. Calling the predicate from the writing statement +-- is in-transaction by construction. +-- +-- DETERMINISTIC LOCK ORDER +-- +-- A task's agent, runtime and issue are not guaranteed to share a workspace, so +-- this can lock up to three workspace rows. They are locked in id order: LockRows +-- sits above Sort in the plan, so `ORDER BY w.id ... FOR KEY SHARE` acquires the +-- locks in that order. Two concurrent cross-workspace writes touching the same +-- pair therefore cannot take them in opposite orders and deadlock. +-- +-- VOLATILE (the default) is load-bearing: it stops the planner from inlining the +-- body or optimising the call away, which would drop the lock. + +CREATE OR REPLACE FUNCTION lock_task_owner_workspaces( + p_agent_id uuid, + p_issue_id uuid, + p_runtime_id uuid +) +RETURNS boolean +LANGUAGE plpgsql +AS $$ +DECLARE + required int := (CASE WHEN p_agent_id IS NULL THEN 0 ELSE 1 END) + + (CASE WHEN p_issue_id IS NULL THEN 0 ELSE 1 END) + + (CASE WHEN p_runtime_id IS NULL THEN 0 ELSE 1 END); + resolved int; + distinct_workspaces int; + locked int; +BEGIN + -- A row with no owner reference at all cannot belong to any workspace. + IF required = 0 THEN + RETURN TRUE; + END IF; + + WITH owners AS ( + SELECT a.workspace_id FROM agent a WHERE a.id = p_agent_id + UNION ALL + SELECT i.workspace_id FROM issue i WHERE i.id = p_issue_id + UNION ALL + SELECT r.workspace_id FROM agent_runtime r WHERE r.id = p_runtime_id + ) + SELECT count(*), count(DISTINCT workspace_id) + INTO resolved, distinct_workspaces + FROM owners; + + -- An owner reference that no longer resolves means its row has been deleted — + -- by a teardown that already committed, most likely. Refuse without leaning + -- on the foreign key to report it. + IF resolved <> required THEN + RETURN FALSE; + END IF; + + WITH locked_workspaces AS ( + SELECT w.id + FROM workspace w + WHERE w.id IN ( + SELECT a.workspace_id FROM agent a WHERE a.id = p_agent_id + UNION + SELECT i.workspace_id FROM issue i WHERE i.id = p_issue_id + UNION + SELECT r.workspace_id FROM agent_runtime r WHERE r.id = p_runtime_id + ) + ORDER BY w.id + FOR KEY SHARE + ) + SELECT count(*) INTO locked FROM locked_workspaces; + + RETURN locked = distinct_workspaces; +END; +$$; diff --git a/server/migrations/284_task_write_workspace_delete_fence.down.sql b/server/migrations/284_task_write_workspace_delete_fence.down.sql deleted file mode 100644 index ad83aa77b..000000000 --- a/server/migrations/284_task_write_workspace_delete_fence.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -DROP TRIGGER IF EXISTS trg_atq_fence_workspace_delete ON agent_task_queue; -DROP FUNCTION IF EXISTS fence_task_write_against_workspace_delete(); diff --git a/server/migrations/284_task_write_workspace_delete_fence.up.sql b/server/migrations/284_task_write_workspace_delete_fence.up.sql deleted file mode 100644 index ed1895b58..000000000 --- a/server/migrations/284_task_write_workspace_delete_fence.up.sql +++ /dev/null @@ -1,105 +0,0 @@ --- Workspace teardown needs a fence that no task write can bypass, and that does --- not borrow its atomicity from a foreign key (MUL-5999 review). --- --- THE WINDOW THIS CLOSES --- --- Teardown sweeps a workspace's tasks and then commits. Anything that inserts a --- task — or moves an existing one onto this workspace's agent, issue or runtime — --- between the last sweep and COMMIT would leave a task behind whose workspace no --- longer exists, together with its own leaf rows. Re-scanning before commit cannot --- close that window: the check and the commit are not atomic with respect to a --- concurrent inserter, so the row can always land after the final empty scan. --- --- WHY NOT THE FOREIGN KEYS --- --- Inserting an agent_task_queue row already takes FOR KEY SHARE on the agent, --- issue and runtime it references, which does block against teardown's FOR UPDATE --- on those rows. That is a side effect of FK enforcement, and the legacy FKs are --- documented in workspace_delete.sql as an expand-phase compatibility net that a --- later contract removes — so relying on them makes tenant-cleanup correctness a --- hostage of a future migration. --- --- WHY A TRIGGER RATHER THAN A CHECK IN EVERY WRITE PATH --- --- The fence has to be in the SAME transaction as the write, otherwise the lock is --- released before the insert lands. There are seven INSERT statements and several --- ownership-changing UPDATEs across agent.sql, autopilot.sql, chat.sql and --- runtime.sql, and several of them run as single autocommit statements, so a --- separate "take the fence" statement issued by the caller would be its own --- transaction and protect nothing. A BEFORE trigger is in-transaction by --- construction, covers every existing path, and — unlike a convention — cannot be --- forgotten by a path added later. --- --- WHAT IT DOES --- --- Takes FOR KEY SHARE on the workspace row behind every owner the new row --- references. FOR KEY SHARE conflicts with the FOR UPDATE that --- LockWorkspaceForDelete already holds for the whole teardown transaction, so a --- task write against a workspace being torn down blocks until teardown finishes, --- and then either proceeds (teardown rolled back) or fails closed with the error --- below (teardown committed and the workspace is gone). --- --- COST --- --- One indexed lookup per owner column that is set, plus a shared row lock, per --- inserted row. It fires on INSERT and only on UPDATEs that actually change an --- ownership column, so the hot claim / dispatch / complete path — which updates --- status, timestamps and results — does not pay for it. Teardown's own --- parent_task_id detach does not touch an ownership column either. - -CREATE OR REPLACE FUNCTION fence_task_write_against_workspace_delete() -RETURNS trigger -LANGUAGE plpgsql -AS $$ -DECLARE - expected int; - locked int; -BEGIN - -- The distinct workspaces this row belongs to. A task's agent, runtime and - -- issue are not guaranteed to share a workspace, so all of them count. - WITH owners AS ( - SELECT workspace_id FROM agent WHERE id = NEW.agent_id - UNION - SELECT workspace_id FROM issue WHERE id = NEW.issue_id - UNION - SELECT workspace_id FROM agent_runtime WHERE id = NEW.runtime_id - ) - SELECT count(*) INTO expected FROM owners; - - -- No resolvable owner at all means the referencing rows are gone; let the - -- foreign keys report that in their own words. - IF expected = 0 THEN - RETURN NEW; - END IF; - - -- The lock and the count have to be at different query levels: PostgreSQL - -- rejects a locking clause in the same SELECT as an aggregate. - WITH locked_workspaces AS ( - SELECT w.id - FROM workspace w - WHERE w.id IN ( - SELECT workspace_id FROM agent WHERE id = NEW.agent_id - UNION - SELECT workspace_id FROM issue WHERE id = NEW.issue_id - UNION - SELECT workspace_id FROM agent_runtime WHERE id = NEW.runtime_id - ) - FOR KEY SHARE - ) - SELECT count(*) INTO locked FROM locked_workspaces; - - IF locked < expected THEN - RAISE EXCEPTION - 'cannot write agent_task_queue row: its workspace is being deleted or no longer exists' - USING ERRCODE = 'foreign_key_violation'; - END IF; - - RETURN NEW; -END; -$$; - -DROP TRIGGER IF EXISTS trg_atq_fence_workspace_delete ON agent_task_queue; -CREATE TRIGGER trg_atq_fence_workspace_delete -BEFORE INSERT OR UPDATE OF agent_id, issue_id, runtime_id ON agent_task_queue -FOR EACH ROW -EXECUTE FUNCTION fence_task_write_against_workspace_delete(); diff --git a/server/pkg/db/generated/agent.sql.go b/server/pkg/db/generated/agent.sql.go index c100954e0..d6c9828cc 100644 --- a/server/pkg/db/generated/agent.sql.go +++ b/server/pkg/db/generated/agent.sql.go @@ -1734,7 +1734,7 @@ INSERT INTO agent_task_queue ( squad_id, context, originator_user_id, accountable_user_id, runtime_mcp_overlay, runtime_connected_apps, originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, $3, 'queued', $4, $5, COALESCE($6::uuid[], '{}'), $7, @@ -1757,7 +1757,7 @@ VALUES ( $20, $21, $22 -) +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -1786,6 +1786,10 @@ type CreateAgentTaskParams struct { TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // head_sha stamps the commit under review into the task's context JSONB so the // reviewer-loop dedup (HasPendingTaskForIssueAndAgent) can tell a pending run // against an OLD head apart from a fresh request against a NEW head (TEN-356). @@ -1882,25 +1886,28 @@ INSERT INTO agent_task_queue ( originator_user_id, accountable_user_id, originator_source, delegated_from_task_id, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( - $1, $2, $3, 'deferred', $4, - $5, - $6, - COALESCE($7::boolean, FALSE), +SELECT + $4, $5, $6, 'deferred', $7, $8, $9, - $10, + COALESCE($10::boolean, FALSE), $11, $12, $13, $14, $15, - $16 -) + $16, + $17, + $18, + $19 +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` type CreateDeferredAgentTaskParams struct { + PAgentID pgtype.UUID `json:"p_agent_id"` + PRuntimeID pgtype.UUID `json:"p_runtime_id"` + PIssueID pgtype.UUID `json:"p_issue_id"` AgentID pgtype.UUID `json:"agent_id"` RuntimeID pgtype.UUID `json:"runtime_id"` IssueID pgtype.UUID `json:"issue_id"` @@ -1919,6 +1926,10 @@ type CreateDeferredAgentTaskParams struct { TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Deferred tasks are inert until PromoteDueDeferredTasksForRuntime flips them // to queued. Used for comment-routing escalation: a thread-owner primary task // gets a delayed assignee fallback without waking both agents at t=0. @@ -1928,6 +1939,9 @@ type CreateDeferredAgentTaskParams struct { // (MUL-4302 §2). func (q *Queries) CreateDeferredAgentTask(ctx context.Context, arg CreateDeferredAgentTaskParams) (AgentTaskQueue, error) { row := q.db.QueryRow(ctx, createDeferredAgentTask, + arg.PAgentID, + arg.PRuntimeID, + arg.PIssueID, arg.AgentID, arg.RuntimeID, arg.IssueID, @@ -2010,7 +2024,7 @@ INSERT INTO agent_task_queue ( originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id, fire_at ) -VALUES ( +SELECT $1, $2, $3, 'deferred', $4, $5, COALESCE($6::uuid[], '{}'), $7, @@ -2033,7 +2047,7 @@ VALUES ( $21, $22, $23 -) +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -2063,6 +2077,10 @@ type CreateDeferredChannelIssueTaskParams struct { FireAt pgtype.Timestamptz `json:"fire_at"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Channel /issue media resolves after issue creation. Persist the assigned // issue task up front for crash safety, but keep it inert until attachment // binding settles or the fire_at fallback is promoted by the normal sweeper. @@ -2155,7 +2173,7 @@ INSERT INTO agent_task_queue ( accountable_user_id, runtime_mcp_overlay, runtime_connected_apps, originator_source, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, NULL, 'queued', $3, $4, $5, $6, @@ -2164,7 +2182,7 @@ VALUES ( $9, $10, $11 -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -2182,6 +2200,10 @@ type CreateQuickCreateTaskParams struct { TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Quick-create tasks have no issue / chat / autopilot link; the entire job // description (prompt, requester, workspace) lives in context JSONB. The // daemon detects this variant via context.type == "quick_create". @@ -2290,6 +2312,7 @@ SELECT p.chat_input_task_id, $2 FROM agent_task_queue p WHERE p.id = $1 + AND lock_task_owner_workspaces(p.agent_id, p.issue_id, p.runtime_id) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -2301,6 +2324,10 @@ type CreateRetryTaskParams struct { RuntimeConnectedApps []byte `json:"runtime_connected_apps"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Clones a parent task into a fresh queued attempt. Carries forward the // agent's resume context (session_id/work_dir) so the child can continue // the conversation when the backend supports it. Resume-unsafe failures are @@ -3853,6 +3880,7 @@ const linkTaskToIssue = `-- name: LinkTaskToIssue :exec UPDATE agent_task_queue SET issue_id = $2 WHERE id = $1 AND issue_id IS NULL + AND lock_task_owner_workspaces(NULL, $2, NULL) ` type LinkTaskToIssueParams struct { @@ -3860,6 +3888,10 @@ type LinkTaskToIssueParams struct { IssueID pgtype.UUID `json:"issue_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Attaches the issue a quick-create task produced back to the task row, once // the agent has finished and the issue exists. Guarded by `issue_id IS NULL` // so this never overwrites an issue id that was set at task creation (only diff --git a/server/pkg/db/generated/autopilot.sql.go b/server/pkg/db/generated/autopilot.sql.go index f5d359325..09566353e 100644 --- a/server/pkg/db/generated/autopilot.sql.go +++ b/server/pkg/db/generated/autopilot.sql.go @@ -280,7 +280,7 @@ INSERT INTO agent_task_queue ( originator_user_id, accountable_user_id, rule_version_id, originator_source, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, NULL, 'queued', $3, $4, $5, $6, $7, @@ -288,7 +288,7 @@ VALUES ( $9, $10, $11 -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -309,6 +309,10 @@ type CreateAutopilotTaskParams struct { // ===================== // Task Queue (run_only mode) // ===================== +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // run_only autopilot dispatch. Attribution depends on the trigger: // - schedule / webhook / api: no human authorized the run, so originator_user_id // stays NULL and accountable_user_id is the rule_owner (the publisher of the diff --git a/server/pkg/db/generated/chat.sql.go b/server/pkg/db/generated/chat.sql.go index 884235f1e..96814a6da 100644 --- a/server/pkg/db/generated/chat.sql.go +++ b/server/pkg/db/generated/chat.sql.go @@ -340,7 +340,7 @@ INSERT INTO agent_task_queue ( runtime_connected_apps, originator_source, trigger_evidence_kind, trigger_evidence_ref_id, fire_at ) -VALUES ( +SELECT $1, $2, NULL, CASE WHEN $6::timestamptz IS NULL THEN 'queued' ELSE 'deferred' END, $3, $4, $5, @@ -353,7 +353,7 @@ VALUES ( $13, $14, $6::timestamptz -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING id, agent_id, issue_id, status, priority, dispatched_at, started_at, completed_at, result, error, created_at, context, runtime_id, session_id, work_dir, trigger_comment_id, chat_session_id, autopilot_run_id, attempt, max_attempts, parent_task_id, failure_reason, trigger_summary, force_fresh_session, is_leader_task, wait_reason, initiator_user_id, handoff_note, prepare_lease_expires_at, squad_id, runtime_mcp_overlay, escalation_for_task_id, fire_at, originator_user_id, runtime_connected_apps, coalesced_comment_ids, delivered_comment_ids, chat_input_task_id, chat_finalize_deferred_at, originator_source, delegated_from_task_id, retry_of_task_id, rerun_of_task_id, rule_version_id, trigger_evidence_kind, trigger_evidence_ref_id, accountable_user_id, session_rollout_missing, retired_session_id, quick_actions_disabled, regenerate_quick_actions_for ` @@ -374,6 +374,10 @@ type CreateChatTaskParams struct { TriggerEvidenceRefID pgtype.UUID `json:"trigger_evidence_ref_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // The chat sender (initiator) is a direct_human originator and accountable; // attribution provenance is stamped so this path is not a NULL-source enqueue // bypass (MUL-4302 §2). diff --git a/server/pkg/db/generated/runtime.sql.go b/server/pkg/db/generated/runtime.sql.go index ef720f14a..b6199e63e 100644 --- a/server/pkg/db/generated/runtime.sql.go +++ b/server/pkg/db/generated/runtime.sql.go @@ -821,9 +821,16 @@ func (q *Queries) ReassignAgentsToRuntime(ctx context.Context, arg ReassignAgent } const reassignTasksToRuntime = `-- name: ReassignTasksToRuntime :execrows +WITH fence AS MATERIALIZED ( + -- Once per statement rather than once per row: the predicate is VOLATILE, so + -- calling it from the WHERE clause of a bulk UPDATE would re-run it for every + -- candidate row. + SELECT lock_task_owner_workspaces(NULL, NULL, $1) AS ok +) UPDATE agent_task_queue SET runtime_id = $1 WHERE runtime_id = $2 + AND (SELECT ok FROM fence) ` type ReassignTasksToRuntimeParams struct { @@ -831,6 +838,10 @@ type ReassignTasksToRuntimeParams struct { OldRuntimeID pgtype.UUID `json:"old_runtime_id"` } +// Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +// locks the owners' workspace rows in the writer's own transaction and returns +// false once they are gone, so this statement writes no row instead of stranding +// a task in a workspace that has just been deleted (MUL-5999). // Re-points every queued/running/completed task referencing old_runtime_id. // Required before deleting the old runtime row because agent_task_queue has // an ON DELETE CASCADE FK that would otherwise drop historical tasks. diff --git a/server/pkg/db/queries/agent.sql b/server/pkg/db/queries/agent.sql index 649dcfce4..de87060bb 100644 --- a/server/pkg/db/queries/agent.sql +++ b/server/pkg/db/queries/agent.sql @@ -278,6 +278,10 @@ WHERE agent_id = $1 ORDER BY created_at DESC; -- name: CreateAgentTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- head_sha stamps the commit under review into the task's context JSONB so the -- reviewer-loop dedup (HasPendingTaskForIssueAndAgent) can tell a pending run -- against an OLD head apart from a fresh request against a NEW head (TEN-356). @@ -291,7 +295,7 @@ INSERT INTO agent_task_queue ( squad_id, context, originator_user_id, accountable_user_id, runtime_mcp_overlay, runtime_connected_apps, originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, $3, 'queued', $4, sqlc.narg(trigger_comment_id), COALESCE(sqlc.narg(coalesced_comment_ids)::uuid[], '{}'), sqlc.narg(trigger_summary), @@ -314,10 +318,14 @@ VALUES ( sqlc.narg(rerun_of_task_id), sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id) -) +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING *; -- name: CreateDeferredChannelIssueTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Channel /issue media resolves after issue creation. Persist the assigned -- issue task up front for crash safety, but keep it inert until attachment -- binding settles or the fire_at fallback is promoted by the normal sweeper. @@ -328,7 +336,7 @@ INSERT INTO agent_task_queue ( originator_source, delegated_from_task_id, rule_version_id, rerun_of_task_id, trigger_evidence_kind, trigger_evidence_ref_id, fire_at ) -VALUES ( +SELECT $1, $2, $3, 'deferred', $4, sqlc.narg(trigger_comment_id), COALESCE(sqlc.narg(coalesced_comment_ids)::uuid[], '{}'), sqlc.narg(trigger_summary), @@ -351,7 +359,7 @@ VALUES ( sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id), @fire_at -) +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING *; -- name: PromoteDeferredChannelIssueTask :one @@ -376,6 +384,10 @@ WHERE id = @id AND originator_user_id IS NOT DISTINCT FROM sqlc.narg(expected_originator_user_id)::uuid; -- name: CreateQuickCreateTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Quick-create tasks have no issue / chat / autopilot link; the entire job -- description (prompt, requester, workspace) lives in context JSONB. The -- daemon detects this variant via context.type == "quick_create". @@ -387,7 +399,7 @@ INSERT INTO agent_task_queue ( accountable_user_id, runtime_mcp_overlay, runtime_connected_apps, originator_source, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, NULL, 'queued', $3, $4, sqlc.narg(originator_user_id), sqlc.narg(accountable_user_id), @@ -396,10 +408,14 @@ VALUES ( sqlc.narg(originator_source), sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id) -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING *; -- name: CreateDeferredAgentTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Deferred tasks are inert until PromoteDueDeferredTasksForRuntime flips them -- to queued. Used for comment-routing escalation: a thread-owner primary task -- gets a delayed assignee fallback without waking both agents at t=0. @@ -413,7 +429,7 @@ INSERT INTO agent_task_queue ( originator_user_id, accountable_user_id, originator_source, delegated_from_task_id, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT @agent_id, @runtime_id, @issue_id, 'deferred', @priority, sqlc.narg(trigger_comment_id), sqlc.narg(trigger_summary), @@ -427,10 +443,14 @@ VALUES ( sqlc.narg(delegated_from_task_id), sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id) -) +WHERE lock_task_owner_workspaces($1, $3, $2) RETURNING *; -- name: LinkTaskToIssue :exec +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Attaches the issue a quick-create task produced back to the task row, once -- the agent has finished and the issue exists. Guarded by `issue_id IS NULL` -- so this never overwrites an issue id that was set at task creation (only @@ -438,9 +458,14 @@ RETURNING *; -- "Creating issue" forever after completion. UPDATE agent_task_queue SET issue_id = $2 -WHERE id = $1 AND issue_id IS NULL; +WHERE id = $1 AND issue_id IS NULL + AND lock_task_owner_workspaces(NULL, $2, NULL); -- name: CreateRetryTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Clones a parent task into a fresh queued attempt. Carries forward the -- agent's resume context (session_id/work_dir) so the child can continue -- the conversation when the backend supports it. Resume-unsafe failures are @@ -523,6 +548,7 @@ SELECT p.chat_input_task_id, sqlc.narg(fire_at) FROM agent_task_queue p WHERE p.id = $1 + AND lock_task_owner_workspaces(p.agent_id, p.issue_id, p.runtime_id) RETURNING *; -- name: CancelAgentTasksByIssue :many diff --git a/server/pkg/db/queries/autopilot.sql b/server/pkg/db/queries/autopilot.sql index 57c5bcab1..90e37af22 100644 --- a/server/pkg/db/queries/autopilot.sql +++ b/server/pkg/db/queries/autopilot.sql @@ -436,6 +436,10 @@ ORDER BY t.id; -- ===================== -- name: CreateAutopilotTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- run_only autopilot dispatch. Attribution depends on the trigger: -- * schedule / webhook / api: no human authorized the run, so originator_user_id -- stays NULL and accountable_user_id is the rule_owner (the publisher of the @@ -452,7 +456,7 @@ INSERT INTO agent_task_queue ( originator_user_id, accountable_user_id, rule_version_id, originator_source, trigger_evidence_kind, trigger_evidence_ref_id ) -VALUES ( +SELECT $1, $2, NULL, 'queued', $3, $4, sqlc.narg(trigger_summary), sqlc.narg(originator_user_id), sqlc.narg(accountable_user_id), @@ -460,7 +464,7 @@ VALUES ( sqlc.narg(originator_source), sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id) -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING *; -- name: GetAutopilotTaskByRun :one diff --git a/server/pkg/db/queries/chat.sql b/server/pkg/db/queries/chat.sql index edef4a6eb..f830dd092 100644 --- a/server/pkg/db/queries/chat.sql +++ b/server/pkg/db/queries/chat.sql @@ -905,6 +905,10 @@ SELECT * FROM chat_message WHERE id = $1; -- name: CreateChatTask :one +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- The chat sender (initiator) is a direct_human originator and accountable; -- attribution provenance is stamped so this path is not a NULL-source enqueue -- bypass (MUL-4302 §2). @@ -914,7 +918,7 @@ INSERT INTO agent_task_queue ( runtime_connected_apps, originator_source, trigger_evidence_kind, trigger_evidence_ref_id, fire_at ) -VALUES ( +SELECT $1, $2, NULL, CASE WHEN sqlc.narg('fire_at')::timestamptz IS NULL THEN 'queued' ELSE 'deferred' END, $3, $4, $5, @@ -927,7 +931,7 @@ VALUES ( sqlc.narg(trigger_evidence_kind), sqlc.narg(trigger_evidence_ref_id), sqlc.narg('fire_at')::timestamptz -) +WHERE lock_task_owner_workspaces($1, NULL, $2) RETURNING *; -- name: PromoteChannelChatTasksIfMediaReady :many diff --git a/server/pkg/db/queries/runtime.sql b/server/pkg/db/queries/runtime.sql index 007639ed1..c2cfa5e0c 100644 --- a/server/pkg/db/queries/runtime.sql +++ b/server/pkg/db/queries/runtime.sql @@ -367,12 +367,23 @@ SET runtime_id = @new_runtime_id WHERE runtime_id = @old_runtime_id; -- name: ReassignTasksToRuntime :execrows +-- Fenced against workspace teardown: lock_task_owner_workspaces (migration 284) +-- locks the owners' workspace rows in the writer's own transaction and returns +-- false once they are gone, so this statement writes no row instead of stranding +-- a task in a workspace that has just been deleted (MUL-5999). -- Re-points every queued/running/completed task referencing old_runtime_id. -- Required before deleting the old runtime row because agent_task_queue has -- an ON DELETE CASCADE FK that would otherwise drop historical tasks. +WITH fence AS MATERIALIZED ( + -- Once per statement rather than once per row: the predicate is VOLATILE, so + -- calling it from the WHERE clause of a bulk UPDATE would re-run it for every + -- candidate row. + SELECT lock_task_owner_workspaces(NULL, NULL, @new_runtime_id) AS ok +) UPDATE agent_task_queue SET runtime_id = @new_runtime_id -WHERE runtime_id = @old_runtime_id; +WHERE runtime_id = @old_runtime_id + AND (SELECT ok FROM fence); -- name: RecordRuntimeLegacyDaemonID :exec -- Remembers the most recent hostname-derived daemon_id that was merged into