From 2873748e1ae19fc34e44edc5d3a16aed948bae5f Mon Sep 17 00:00:00 2001
From: Sawyer Cutler <sawyer@dirtroad.dev>
Date: Sat, 26 Sep 2026 17:37:32 -0700
Subject: [PATCH] fix(capture): serialize parallel adds to one externalRef

Capture read the active version and inserted the next one with no lock, so parallel adds to the same externalRef failed on the version unique index. A transaction-scoped advisory lock on (tenant, adapter, externalRef) now orders them into sequential versions.
---
 e2e/add-search.test.ts  | 19 +++++++++++++++++++
 src/services/capture.ts | 11 ++++++++++-
 2 files changed, 29 insertions(+), 1 deletion(-)

diff --git a/e2e/add-search.test.ts b/e2e/add-search.test.ts
index 4bb9aa3..202ca6d 100644
--- a/e2e/add-search.test.ts
+++ b/e2e/add-search.test.ts
@@ -103,4 +103,23 @@ describe.skipIf(testDatabaseUrl() === undefined)("add and search", () => {
     expect(await titles("alice", "acme")).toContain("Quarterly roadmap");
     expect(await titles("bob", "globex")).toEqual([]);
   });
+
+  test("parallel adds to one externalRef become sequential versions", async () => {
+    const results = await Promise.all(
+      [1, 2, 3, 4, 5, 6].map((i) =>
+        memory!.add({
+          tenantId: "acme",
+          principalId: "alice",
+          externalRef: "parallel-ref",
+          content: { title: "Parallel", text: `parallel edit ${i}` },
+        }),
+      ),
+    );
+    const documentIds = new Set(results.map((r) => r.documentId));
+    expect(documentIds.size).toBe(1);
+    const versions = await db.sql<{ version: number }[]>`
+      SELECT version FROM memory.version
+        WHERE document_id = ${[...documentIds][0]!} ORDER BY version`;
+    expect(versions.map((v) => v.version)).toEqual([1, 2, 3, 4, 5, 6]);
+  });
 });
diff --git a/src/services/capture.ts b/src/services/capture.ts
index 4a6dc5a..53880f2 100644
--- a/src/services/capture.ts
+++ b/src/services/capture.ts
@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { and, desc, eq } from "drizzle-orm";
+import { and, desc, eq, sql } from "drizzle-orm";
 import type { Db, RawSql } from "../db/client.js";
 import type { EngineConfig } from "../config.js";
 import { newId } from "../core/id.js";
@@ -374,6 +374,9 @@ async function insertChunksAndGraph(
   return insertedChunks;
 }
 
+// Namespaces the two-int4 advisory lock; deliberately arbitrary.
+const CAPTURE_LOCK_NAMESPACE = 0x3e30_7a12;
+
 // The replay pipeline's single derivation core: adaptAndPlan's output → doc
 // upsert → version(+authority) → chunks → edges, scoped to a target
 // `generation`. The live /capture path calls this with generation='live'
@@ -393,6 +396,12 @@ async function deriveVersionInTransaction(
   generation: string,
 ): Promise<CaptureTxResult> {
   const doc = plan.document;
+  // Serializes captures of one (tenant, adapter, externalRef): without it,
+  // parallel captures read the same active version and collide on the next
+  // version number. The document row may not exist yet, so FOR UPDATE can't.
+  await tx.execute(
+    sql`SELECT pg_advisory_xact_lock(${CAPTURE_LOCK_NAMESPACE}, hashtext(${JSON.stringify([input.tenantId, input.adapter, doc.externalRef])}))`,
+  );
   const existingRows = await tx
     .select()
     .from(memoryDocument)