fix(core): close event task races

Author: dcramer

packages/junior/src/chat/event-tasks/ingest.ts

@@ -5,9 +5,9 @@ import {
   type ResourceEvent,
 } from "@sentry/junior-plugin-api";
 import { dispatchTask } from "@/chat/agent-dispatch/context";
-import { getDb } from "@/chat/db";
+import { getDb, getSqlExecutor } from "@/chat/db";
 import {
-  createOrGetEventTaskMatch,
+  claimActiveEventTaskMatch,
   findMatchingEventTasks,
   markEventTaskMatchDispatched,
 } from "@/chat/event-tasks/store";
@@ -77,7 +77,8 @@ export async function ingestEventTasks(
         status: "pending",
         taskId: task.id,
       };
-      const match = await createOrGetEventTaskMatch(db, pending);
+      const match = await claimActiveEventTaskMatch(getSqlExecutor(), pending);
+      if (!match) continue;
       if (match.status === "dispatched") continue;
       const credentialSubject =
         task.credentialMode === "creator"

packages/junior/src/chat/event-tasks/store.ts

@@ -6,7 +6,7 @@ import {
 } from "@sentry/junior-plugin-api";
 import { and, asc, eq, ne, sql } from "drizzle-orm";
 import { z } from "zod";
-import type { JuniorDatabase } from "@/db/db";
+import type { JuniorDatabase, JuniorSqlDatabase } from "@/db/db";
 import {
   juniorEventTaskMatches,
   juniorEventTasks,
@@ -211,23 +211,38 @@ async function getEventTaskMatch(
   return rows[0] ? parseMatch(rows[0].match) : undefined;
 }
 
-/** Create a deduplication record or return the existing match. */
-export async function createOrGetEventTaskMatch(
-  db: JuniorDatabase,
+/** Claim a deduplication record while its event task remains active. */
+export async function claimActiveEventTaskMatch(
+  sqlDb: JuniorSqlDatabase,
   match: EventTaskMatch,
-): Promise<EventTaskMatch> {
+): Promise<EventTaskMatch | undefined> {
   const parsed = eventTaskMatchSchema.parse(match);
-  await db
-    .insert(juniorEventTaskMatches)
-    .values({
-      id: parsed.id,
-      taskId: parsed.taskId,
-      status: parsed.status,
-      createdAtMs: parsed.createdAtMs,
-      match: parsed,
-    })
-    .onConflictDoNothing();
-  return (await getEventTaskMatch(db, parsed.id)) ?? parsed;
+  return await sqlDb.transaction(async () => {
+    const db = sqlDb.db();
+    const active = await db
+      .select({ id: juniorEventTasks.id })
+      .from(juniorEventTasks)
+      .where(
+        and(
+          eq(juniorEventTasks.id, parsed.taskId),
+          eq(juniorEventTasks.status, "active"),
+        ),
+      )
+      .for("update")
+      .limit(1);
+    if (!active[0]) return undefined;
+    await db
+      .insert(juniorEventTaskMatches)
+      .values({
+        id: parsed.id,
+        taskId: parsed.taskId,
+        status: parsed.status,
+        createdAtMs: parsed.createdAtMs,
+        match: parsed,
+      })
+      .onConflictDoNothing();
+    return (await getEventTaskMatch(db, parsed.id)) ?? parsed;
+  });
 }
 
 /** Mark a pending event task match after durable agent dispatch succeeds. */

packages/junior/src/chat/tools/event-tasks.ts

@@ -312,12 +312,12 @@ export function createUpdateEventTaskTool(context: ToolRuntimeContext) {
       if (
         input.task === undefined &&
         input.trigger === undefined &&
-        input.credentialMode === undefined
+        input.credentialMode == null
       ) {
         throw new ToolInputError("Event task update requires a change.");
       }
       const changesExecution =
-        input.task !== undefined || input.trigger !== undefined;
+        input.task !== undefined && input.task !== current.task.text;
       const next: EventTask = {
         ...current,
         credentialMode:

packages/junior/tests/integration/event-tasks.test.ts

@@ -5,6 +5,7 @@ import { getDispatchRecord } from "@/chat/agent-dispatch/store";
 import { migrateSchema } from "@/chat/conversations/sql/migrations";
 import { ingestEventTasks } from "@/chat/event-tasks/ingest";
 import {
+  claimActiveEventTaskMatch,
   deleteActiveEventTask,
   getEventTask,
   updateActiveEventTask,
@@ -178,6 +179,31 @@ describe("event tasks", () => {
     };
     expect(listed.tasks).toHaveLength(2);
 
+    await execute(createUpdateEventTaskTool(context("U999")), {
+      taskId: first.task.id,
+      trigger: {
+        provider: "github",
+        resourceRef: "github:pull_request:getsentry/junior#1174",
+        resourceType: "pull_request",
+        label: "Updated GitHub PR label",
+        events: ["review.changes_requested"],
+      },
+    });
+    await execute(createUpdateEventTaskTool(context("U999")), {
+      taskId: first.task.id,
+      task: "Address the requested changes.",
+    });
+    expect(await getEventTask(fixture.sql.db(), first.task.id)).toMatchObject({
+      credentialMode: "creator",
+      trigger: { label: "Updated GitHub PR label" },
+    });
+    await expect(
+      execute(createUpdateEventTaskTool(context("U999")), {
+        taskId: first.task.id,
+        credentialMode: null,
+      }),
+    ).rejects.toThrow("Event task update requires a change.");
+
     await execute(createUpdateEventTaskTool(context("U999")), {
       taskId: first.task.id,
       task: "Only summarize the requested changes.",
@@ -248,5 +274,15 @@ describe("event tasks", () => {
       status: "deleted",
       task: { text: "Post the first summary." },
     });
+    await expect(
+      claimActiveEventTaskMatch(fixture.sql, {
+        id: "evmatch_stale_deleted_task",
+        createdAtMs: deleted.updatedAtMs + 1,
+        eventKey: "github:delivery-after-delete",
+        eventType: "review.changes_requested",
+        status: "pending",
+        taskId: created.task.id,
+      }),
+    ).resolves.toBeUndefined();
   });
 });