fix(sync-lock): key the lock per job so jobs stop starving each other

src/app/api/sync/scheduled/route.test.ts

@@ -6,7 +6,7 @@ function createPrismaMock() {
 
   const syncLockFindUnique = vi.fn();
   const syncLockDeleteMany = vi.fn().mockResolvedValue({ count: 0 });
-  const syncLockCreate = vi.fn().mockResolvedValue({ id: "global" });
+  const syncLockCreate = vi.fn().mockResolvedValue({ id: "issue-sync" });
   // Atomic claim used by acquireLock's UPDATE ... WHERE flow. Default
   // count 0 routes acquisition down the insert path, like an absent row.
   const syncLockUpdateMany = vi.fn().mockResolvedValue({ count: 0 });
@@ -266,7 +266,7 @@ describe("POST /api/sync/scheduled — locking", () => {
   it("returns 409 when a sync is already running", async () => {
     const { POST } = await import("./route");
     prismaMock.prisma.syncLock.findUnique.mockResolvedValue({
-      id: "global",
+      id: "issue-sync",
       syncRunId: "sync-run-123",
       acquiredAt: new Date(),
     });
@@ -280,7 +280,7 @@ describe("POST /api/sync/scheduled — locking", () => {
   it("allows sync when existing lock is stale (>30 min)", async () => {
     const { POST } = await import("./route");
     prismaMock.prisma.syncLock.findUnique.mockResolvedValue({
-      id: "global",
+      id: "issue-sync",
       syncRunId: "sync-run-old",
       acquiredAt: new Date(Date.now() - 31 * 60 * 1000), // 31 min ago
     });
@@ -292,7 +292,8 @@ describe("POST /api/sync/scheduled — locking", () => {
     expect(res.status).toBe(200);
     expect(prismaMock.prisma.syncLock.updateMany).toHaveBeenCalledWith(
       expect.objectContaining({
-        where: expect.objectContaining({ id: "global" }),
+        // This route is "scheduled", which maps to the issue-sync key.
+        where: expect.objectContaining({ id: "issue-sync" }),
         data: expect.objectContaining({ syncRunId: expect.any(String) }),
       }),
     );
@@ -306,10 +307,11 @@ describe("POST /api/sync/scheduled — locking", () => {
     const res = await POST(makeRequest());
     expect(res.status).toBe(200);
 
-    // deleteMany should have been called to remove the lock row
+    // releaseLock keys on the run id alone now, so callers need not know
+    // which lock key their sync type maps to.
     expect(prismaMock.prisma.syncLock.deleteMany).toHaveBeenCalledWith(
       expect.objectContaining({
-        where: expect.objectContaining({ id: "global" }),
+        where: expect.objectContaining({ syncRunId: expect.any(String) }),
       }),
     );
   });

src/app/api/sync/scheduled/route.ts

@@ -93,7 +93,20 @@ export async function POST(request: Request) {
     }
 
     // Automation sync (optional, opt-in)
+    //
+    // Held under the automation key, not this route's issue-sync one: the two
+    // write the same automation tables, and concurrent Prisma upserts on the
+    // same unique keys abort the batch transactions in automation-sync.ts.
+    // Skipping rather than failing keeps a lost race from taking the issue
+    // sync down with it — the scheduler never sets this flag, so only a manual
+    // call reaches here.
+    let automationSkipped = false;
     if (syncAutomation) {
+      const automationLock = await acquireLock("automation");
+      if (!automationLock.locked) {
+        automationSkipped = true;
+      } else {
+      try {
       const trackedRepos = await getTrackedRepos();
       const results: { repo: string; result: { success: boolean } }[] = [];
 
@@ -115,6 +128,12 @@ export async function POST(request: Request) {
         synced: results.filter((r) => r.result.success).length,
         failed: results.filter((r) => !r.result.success).length,
       };
+      } finally {
+        await releaseLock(automationLock.runId).catch((e) =>
+          console.error("Failed to release automation lock:", e),
+        );
+      }
+      }
     }
 
     // Update the sync run record
@@ -149,6 +168,13 @@ export async function POST(request: Request) {
       finishedAt,
     };
 
+    // Report a skipped automation pass explicitly rather than reporting zero
+    // synced, which reads as "ran and found nothing".
+    if (automationSkipped) {
+      response.automationSkipped = true;
+      response.automationSkippedReason = "another automation sync holds the lock";
+    }
+
     if (syncIssues && issueSync) {
       response.issues = {
         repos: issueSync.repos,

src/lib/sync-lock.integration.test.ts

@@ -37,7 +37,9 @@ suite("sync lock against a real PostgreSQL", () => {
   });
 
   it("two overlapping acquisitions on an empty table: one wins, the loser fails cleanly", async () => {
-    const results = await Promise.allSettled([acquireLock("scheduled"), acquireLock("reconcile")]);
+    // Same key: scheduled and manual both map to "issue-sync", so this is
+    // still a genuine contention on the path that produced P2039.
+    const results = await Promise.allSettled([acquireLock("scheduled"), acquireLock("manual")]);
 
     const won = results.filter((r) => r.status === "fulfilled" && r.value.locked);
     const conflicted = results.filter((r) => r.status === "fulfilled" && !r.value.locked);
@@ -54,7 +56,7 @@ suite("sync lock against a real PostgreSQL", () => {
   it("two overlapping acquisitions with an unheld row present: one wins, the loser fails cleanly", async () => {
     // Exercises the UPDATE race rather than the INSERT race: the row exists
     // with no holder, so both contenders match claimableWhere().
-    await db.syncLock.create({ data: { id: "global", syncRunId: null, acquiredAt: new Date() } });
+    await db.syncLock.create({ data: { id: "issue-sync", syncRunId: null, acquiredAt: new Date() } });
 
     const results = await Promise.allSettled([acquireLock("scheduled"), acquireLock("manual")]);
 
@@ -66,11 +68,55 @@ suite("sync lock against a real PostgreSQL", () => {
     const first = await acquireLock("scheduled");
     expect(first.locked).toBe(true);
 
-    const second = await acquireLock("reconcile");
+    // Same key as the holder, so this must still conflict.
+    const second = await acquireLock("manual");
     expect(second).toEqual({ locked: false });
 
     await releaseLock(first.locked ? first.runId : "");
-    const third = await acquireLock("reconcile");
+    const third = await acquireLock("manual");
     expect(third.locked).toBe(true);
   });
+  it("different sync types acquire independently — the starvation regression", async () => {
+    // The bug: every job shared one "global" row, so any job holding the lock
+    // 409'd all the others. With the scheduler arming all jobs on the same
+    // startup delay their ticks are phase-locked, so the same jobs lost every
+    // time: pr-followup and reconcile recorded zero successful runs over a
+    // pod's lifetime and no PR-fix work was ever queued.
+    const results = await Promise.allSettled([
+      acquireLock("scheduled"),
+      acquireLock("automation"),
+      acquireLock("pr-followup"),
+      acquireLock("reconcile"),
+      acquireLock("stale-work"),
+    ]);
+
+    const threw = results.filter((r) => r.status === "rejected");
+    expect(threw).toEqual([]);
+
+    const acquired = results
+      .filter((r): r is PromiseFulfilledResult<Awaited<ReturnType<typeof acquireLock>>> => r.status === "fulfilled")
+      .map((r) => r.value)
+      .filter((v) => v.locked);
+    // All five, concurrently. Under the shared key exactly one would win.
+    expect(acquired).toHaveLength(5);
+
+    const rows = await db.syncLock.findMany({ select: { id: true } });
+    expect(rows.map((r) => r.id).sort()).toEqual(
+      ["automation", "issue-sync", "pr-followup", "reconcile", "stale-work"],
+    );
+  });
+
+  it("releasing one key does not free another", async () => {
+    const sync = await acquireLock("scheduled");
+    const rec = await acquireLock("reconcile");
+    expect(sync.locked && rec.locked).toBe(true);
+    if (!sync.locked || !rec.locked) return;
+
+    await releaseLock(rec.runId);
+
+    // The issue-sync key is still held by a live run.
+    expect((await acquireLock("manual")).locked).toBe(false);
+    // The reconcile key was freed by its own release.
+    expect((await acquireLock("reconcile")).locked).toBe(true);
+  });
 });

src/lib/sync-lock.test.ts

@@ -56,11 +56,11 @@ describe("acquireLock", () => {
     expect(mocks.$executeRaw).toHaveBeenCalled();
 
     await releaseLock(result.locked ? result.runId : "run-1");
-    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { id: "global", syncRunId: "run-1" } });
+    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { syncRunId: "run-1" } });
   });
 
   it("conflicts when a fresh (live) lock is held", async () => {
-    mocks.syncLock.findUnique.mockResolvedValue({ id: "global", syncRunId: "other", acquiredAt: new Date() });
+    mocks.syncLock.findUnique.mockResolvedValue({ id: "issue-sync", syncRunId: "other", acquiredAt: new Date() });
 
     const result = await acquireLock("manual");
 
@@ -85,7 +85,7 @@ describe("acquireLock", () => {
   });
 
   it("two simultaneous acquires of a live lock cannot both succeed", async () => {
-    mocks.syncLock.findUnique.mockResolvedValue({ id: "global", syncRunId: "other", acquiredAt: new Date() });
+    mocks.syncLock.findUnique.mockResolvedValue({ id: "issue-sync", syncRunId: "other", acquiredAt: new Date() });
 
     const [a, b] = await Promise.all([acquireLock("scheduled"), acquireLock("reconcile")]);
 
@@ -97,7 +97,7 @@ describe("acquireLock", () => {
   it("reclaims an expired lock, and logs the takeover with holder and age", async () => {
     const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
     const stale = new Date(Date.now() - (MAX_AGE_MS + 60_000));
-    mocks.syncLock.findUnique.mockResolvedValue({ id: "global", syncRunId: "old", acquiredAt: stale });
+    mocks.syncLock.findUnique.mockResolvedValue({ id: "automation", syncRunId: "old", acquiredAt: stale });
     // Atomic claim wins in place (no delete).
     mocks.syncLock.updateMany.mockResolvedValue({ count: 1 });
 
@@ -109,7 +109,8 @@ describe("acquireLock", () => {
     expect(mocks.$executeRaw).not.toHaveBeenCalled();
     expect(mocks.syncLock.updateMany).toHaveBeenCalledWith(
       expect.objectContaining({
-        where: expect.objectContaining({ id: "global" }),
+        // "automation" maps to its own key, not the shared issue-sync one.
+        where: expect.objectContaining({ id: "automation" }),
         data: expect.objectContaining({ syncRunId: "run-1" }),
       }),
     );
@@ -126,7 +127,7 @@ describe("acquireLock", () => {
     // Fresh acquiredAt but no holder: the row #840 left stuck. A null holder
     // is claimable regardless of age because a live run always records its
     // run id.
-    mocks.syncLock.findUnique.mockResolvedValue({ id: "global", syncRunId: null, acquiredAt: new Date() });
+    mocks.syncLock.findUnique.mockResolvedValue({ id: "issue-sync", syncRunId: null, acquiredAt: new Date() });
     mocks.syncLock.updateMany.mockResolvedValue({ count: 1 });
 
     const result = await acquireLock("reconcile");
@@ -180,14 +181,14 @@ describe("acquireLock", () => {
     }).rejects.toThrow();
 
     await releaseLock(lock.locked ? lock.runId : "run-1");
-    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { id: "global", syncRunId: "run-1" } });
+    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { syncRunId: "run-1" } });
   });
 });
 
 describe("releaseLock", () => {
   it("deletes only this run's lock row", async () => {
     await releaseLock("run-1");
-    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { id: "global", syncRunId: "run-1" } });
+    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { syncRunId: "run-1" } });
   });
 
   it("is a safe no-op when no matching row exists (release-on-any-path)", async () => {
@@ -196,7 +197,7 @@ describe("releaseLock", () => {
     // nothing → no throw.
     mocks.syncLock.deleteMany.mockResolvedValue({ count: 0 });
     await expect(releaseLock("never-acquired")).resolves.toBeUndefined();
-    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { id: "global", syncRunId: "never-acquired" } });
+    expect(mocks.syncLock.deleteMany).toHaveBeenCalledWith({ where: { syncRunId: "never-acquired" } });
   });
 });
 

src/lib/sync-lock.ts

@@ -1,13 +1,22 @@
 /**
  * Shared sync-locking module.
  *
- * Provides a DB-backed single-row lock (syncLock table) that all sync
- * entry-points share to prevent overlapping concurrent runs across:
- *   - Scheduled sync (`/api/sync/scheduled`)
- *   - Manual issue sync (`/api/sync`)
- *   - Automation sync (`/api/automation/sync`)
- *   - PR follow-up sync (`/api/pr-followup/sync`)
- *   - Issue reconciliation (`/api/issues/reconcile`)
+ * Provides DB-backed locks (syncLock table), keyed per job by
+ * LOCK_KEY_BY_TYPE below. Each entry-point excludes only the runs it can
+ * actually corrupt:
+ *   - Scheduled sync (`/api/sync/scheduled`) and manual issue sync (`/api/sync`)
+ *     share the `issue-sync` key — same full-sync write path over the same rows.
+ *   - Automation sync (`/api/automation/sync`) — disjoint tables.
+ *   - PR follow-up sync (`/api/pr-followup/sync`) — PrFixQueueItem only.
+ *   - Issue reconciliation (`/api/issues/reconcile`) — column-disjoint writes.
+ *   - Stale-work sweep (`/api/agent-work/sweep`) — conditional transitions.
+ *
+ * This used to be one `"global"` row shared by all six. That made every job
+ * exclude every other, and because the scheduler arms all jobs with the same
+ * startup delay their intervals stay phase-locked, so the same jobs lost every
+ * race: `pr-followup` and `reconcile` recorded zero successful runs over a
+ * pod's lifetime while a full sync held the lock for minutes. No PR-fix work
+ * was queued at all as a result.
  *
  * Lock semantics:
  *   - First writer wins; subsequent writers get a 409 Conflict.
@@ -32,7 +41,6 @@
 
 import { prisma } from "@/lib/prisma";
 
-const LOCK_ID = "global" as const;
 const DEFAULT_MAX_AGE_MS = 30 * 60 * 1000; // 30 minutes
 
 /**
@@ -68,16 +76,41 @@ export type SyncType =
   | "reconcile"
   | "stale-work";
 
+/**
+ * Lock key per sync type. One row per key in the same `sync_lock` table that
+ * `groomer-lock.ts` already uses for its own `"groomer"` row, so the multi-row
+ * pattern is established rather than new — never reuse that id here.
+ *
+ * `scheduled` and `manual` deliberately share a key. Both run the full issue
+ * sync over the same Issue rows, and `makePrismaIssueStore().createIssue` is a
+ * plain `create`, not an upsert, so two concurrent runs that both see a new
+ * issue as absent race into P2002 (#333).
+ *
+ * Everything else only needs to exclude ITSELF — overlapping replicas of the
+ * same job (#822) — because each mutates disjoint tables or writes
+ * conditionally and idempotently. Sharing one key made them exclude each
+ * other instead: `pr-followup` and `reconcile` lost every race and never ran
+ * once over a pod's lifetime, so no PR-fix work was ever queued.
+ */
+const LOCK_KEY_BY_TYPE: Record<SyncType, string> = {
+  scheduled: "issue-sync",
+  manual: "issue-sync",
+  automation: "automation",
+  "pr-followup": "pr-followup",
+  reconcile: "reconcile",
+  "stale-work": "stale-work",
+};
+
 /**
  * Where-clause for the atomic claim: a row is claimable when it has no
  * recorded holder (an orphan — nothing can be holding the lock without one,
  * since every acquisition records its run id) or when it is older than the
  * TTL. A fresh row with a live holder matches neither, so two simultaneous
  * acquisitions of a live lock can never both succeed.
  */
-function claimableWhere(): Record<string, unknown> {
+function claimableWhere(lockId: string): Record<string, unknown> {
   return {
-    id: LOCK_ID,
+    id: lockId,
     OR: [
       { syncRunId: null },
       { acquiredAt: { lt: new Date(Date.now() - MAX_AGE_MS) } },
@@ -103,7 +136,8 @@ function claimableWhere(): Record<string, unknown> {
 export async function acquireLock(
   syncType: SyncType,
 ): Promise<AcquiredLock | LockConflict> {
-  const existing = await prisma.syncLock.findUnique({ where: { id: LOCK_ID } });
+  const lockId = LOCK_KEY_BY_TYPE[syncType];
+  const existing = await prisma.syncLock.findUnique({ where: { id: lockId } });
 
   // Fast path: a live lock conflicts without creating a run record, so
   // scheduler ticks during a running sync don't pollute the sync history.
@@ -123,7 +157,7 @@ export async function acquireLock(
       // Atomic conditional claim. count === 1 wins; count === 0 means no
       // claimable row (absent, or held live).
       const claimed = await tx.syncLock.updateMany({
-        where: claimableWhere(),
+        where: claimableWhere(lockId),
         data: { syncRunId: run.id, acquiredAt: new Date() },
       });
       if (claimed.count === 1) return run.id;
@@ -143,7 +177,7 @@ export async function acquireLock(
       // this statement returns 0 affected rows instead of throwing.
       const inserted = await tx.$executeRaw`
         INSERT INTO "sync_lock" ("id", "syncRunId", "acquiredAt")
-        VALUES (${LOCK_ID}, ${run.id}, ${new Date()})
+        VALUES (${lockId}, ${run.id}, ${new Date()})
         ON CONFLICT ("id") DO NOTHING
       `;
       if (inserted === 1) return run.id;
@@ -153,7 +187,7 @@ export async function acquireLock(
       // transaction just released or left the row claimable (stale or
       // orphaned) we take it; otherwise a live lock holds it and we lose.
       const retry = await tx.syncLock.updateMany({
-        where: claimableWhere(),
+        where: claimableWhere(lockId),
         data: { syncRunId: run.id, acquiredAt: new Date() },
       });
       if (retry.count === 1) return run.id;
@@ -184,7 +218,8 @@ export async function acquireLock(
  * Uses a conditional delete to avoid releasing another run's lock.
  */
 export async function releaseLock(runId: string): Promise<void> {
-  await prisma.syncLock.deleteMany({
-    where: { id: LOCK_ID, syncRunId: runId },
-  });
+  // Keyed on the run id alone, so callers need not know which lock key their
+  // sync type maps to. Safe because a run id is either an IssueSyncRun cuid or
+  // groomer-lock's randomUUID token — unique across every key.
+  await prisma.syncLock.deleteMany({ where: { syncRunId: runId } });
 }