fix(worker): persist analysis reports with neon http

Author: outof-place

apps/worker/src/run-analysis/store.test.ts

@@ -28,6 +28,30 @@ describe("run analysis report store", () => {
     expect((await getRunAnalysisReport(db, value.runId))?.jira.research.state).toBe("pending");
   });
 
+  it("persists through a driver without interactive transaction support", async () => {
+    const noTransactionDb = new Proxy(db, {
+      get(target, property, receiver) {
+        if (property === "transaction") {
+          return async () => {
+            throw new Error("No transactions support in neon-http driver");
+          };
+        }
+        return Reflect.get(target, property, receiver);
+      },
+    });
+    const value = report();
+
+    await db.insert(workflowRuns).values({
+      runId: value.runId,
+      workflowId: "wf_agent",
+      workflowName: "Agent",
+      analysisReport: null,
+    });
+    await expect(recordRunAnalysisReport(noTransactionDb, value)).resolves.toBeUndefined();
+    await expect(recordRunAnalysisReport(noTransactionDb, value)).resolves.toBeUndefined();
+    expect(await getRunAnalysisReport(db, value.runId)).toEqual(value);
+  });
+
   it("finalizes only the final usage slot", async () => {
     const value = report();
     await recordRunAnalysisReport(db, value);

apps/worker/src/run-analysis/store.ts

@@ -1,4 +1,4 @@
-import { eq, sql } from "drizzle-orm";
+import { and, eq, isNull, sql } from "drizzle-orm";
 import type { RunAnalysisReport, RunAnalysisUsageSnapshot } from "@shared/contracts";
 import type { Db } from "../db/client.js";
 import { workflowRuns } from "../db/schema.js";
@@ -17,6 +17,12 @@ const deliveryRank = {
   posted: 3,
 } as const;
 
+const MAX_REPORT_WRITE_ATTEMPTS = 8;
+
+function jsonbValue(value: unknown) {
+  return sql`${JSON.stringify(value ?? null)}::jsonb`;
+}
+
 function newerDelivery(
   current: RunAnalysisReport["jira"]["research"],
   incoming: RunAnalysisReport["jira"]["research"],
@@ -70,11 +76,11 @@ export async function recordRunAnalysisReport(
   db: Db,
   report: RunAnalysisReport,
 ): Promise<void> {
-  await db.transaction(async (tx) => {
-    // The advisory transaction lock also serializes the first insert, where no
-    // workflow_runs row exists yet for SELECT ... FOR UPDATE to lock.
-    await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${report.runId}, 0))`);
-    const [row] = await tx
+  // neon-http does not support interactive transactions. Compare-and-swap the
+  // exact JSONB snapshot that was merged so a concurrent writer cannot be
+  // overwritten; a conflicting insert follows the same retry path.
+  for (let attempt = 0; attempt < MAX_REPORT_WRITE_ATTEMPTS; attempt += 1) {
+    const [row] = await db
       .select({ analysisReport: workflowRuns.analysisReport })
       .from(workflowRuns)
       .where(eq(workflowRuns.runId, report.runId))
@@ -83,22 +89,40 @@ export async function recordRunAnalysisReport(
       parseStoredRunAnalysisReport(row?.analysisReport),
       report,
     );
-    await tx
+    if (row) {
+      const snapshotCondition = row.analysisReport === null
+        ? isNull(workflowRuns.analysisReport)
+        : sql`${workflowRuns.analysisReport} is not distinct from ${jsonbValue(row.analysisReport)}`;
+      const [updated] = await db
+        .update(workflowRuns)
+        .set({ analysisReport: merged, updatedAt: sql`now()` })
+        .where(
+          and(
+            eq(workflowRuns.runId, report.runId),
+            snapshotCondition,
+          ),
+        )
+        .returning({ runId: workflowRuns.runId });
+      if (updated) return;
+      continue;
+    }
+
+    const [inserted] = await db
       .insert(workflowRuns)
       .values({
         runId: report.runId,
         workflowId: "wf_agent",
         workflowName: "Agent",
         analysisReport: merged,
       })
-      .onConflictDoUpdate({
-        target: workflowRuns.runId,
-        set: {
-          analysisReport: merged,
-          updatedAt: sql`now()`,
-        },
-      });
-  });
+      .onConflictDoNothing({ target: workflowRuns.runId })
+      .returning({ runId: workflowRuns.runId });
+    if (inserted) return;
+  }
+
+  throw new Error(
+    `Could not persist run analysis report for ${report.runId} after ${MAX_REPORT_WRITE_ATTEMPTS} attempts`,
+  );
 }
 
 /** Read JSONB through the structural parser; callers never trust a cast from