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 { 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)