diff --git a/README.md b/README.md index 2067184..a0cdf1b 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,8 @@ Returns a `Hono` sub-app. Mount it at `/api/tenants/:tenantId/memory` A missing principal answers `401` and a missing grant answers `403`. `callerResolver` is optional: it resolves a non-session caller to a `{ tenantId, principalId }` before the same grant checks run, and a malformed result answers `500`. +Sharing on `add` needs grants the host may have to mint. `share.principals` writes each peer a `search` grant on `memory.doc:` only when `grantStore` has a `putGrant` method; Interchange's stock grant store has none, so on a stock hub the host mints those grants. `share.tenant` never writes grants: the host grants `search` on `memory.tenant:` to whoever should see tenant-wide documents. + ### `runMemoryMigrations(config, { schema, ftsLanguage })` Takes the same `DBConfig` and `schema` as Interchange's `runMigrations`. `schema` holds the host's `tenant` and `principal` tables; this package's tables always go in `memory`. `ftsLanguage` is fixed into the full-text index and must match `FTS_LANGUAGE` at runtime. diff --git a/docs/FEED.md b/docs/FEED.md index 4393650..2970d57 100644 --- a/docs/FEED.md +++ b/docs/FEED.md @@ -26,6 +26,7 @@ memory.feed({ tenantId, principalId, after?, limit?, excludeGenerator? }) | `excludeGenerator` | Skip versions with this `generator_agent_id` (loop-safe). | - Ordered by `feed_seq` ascending (Postgres `bigserial` on `memory.version`). +- Live versions of a tenant commit in `feed_seq` order (a per-tenant transaction lock), so a cursor never passes a version still being written. - **Live generation only** — same rule as default search. - Capability: `memory` / `search` (same as list/retrieve). - Document access: grant-tag post-filter identical to search. diff --git a/e2e/feed.test.ts b/e2e/feed.test.ts new file mode 100644 index 0000000..e6ff63d --- /dev/null +++ b/e2e/feed.test.ts @@ -0,0 +1,86 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { createInMemoryGrantStore } from "@intx/authz"; + +import type { Memory } from "../src/memory.js"; +import { + allow, + createTestDb, + createTestMemory, + seedPrincipal, + testDatabaseUrl, + type TestDb, +} from "./helpers.js"; + +const alice = { tenantId: "acme", principalId: "alice" }; +const settle = () => new Promise((resolve) => setTimeout(resolve, 500)); + +describe.skipIf(testDatabaseUrl() === undefined)("feed", () => { + let db: TestDb; + let memory: Memory; + + beforeAll(async () => { + db = await createTestDb(); + await seedPrincipal(db, "acme", "alice"); + memory = createTestMemory( + db, + createInMemoryGrantStore([ + allow("alice", "add"), + allow("alice", "search"), + ]), + ); + }); + + afterAll(async () => { + await memory?.close(); + await db?.close(); + }); + + // A capture that took a lower feed_seq but commits after a later one must + // not fall behind a cursor that already moved past it. + test("a version that commits out of feed_seq order is not skipped", async () => { + const { documentId } = await memory.add({ + ...alice, + externalRef: "slow", + content: { title: "Slow", text: "first draft" }, + }); + const start = await memory.feed!({ ...alice }); + + // Holding the document row stalls the next capture of it after it has + // inserted its version. + let release = () => {}; + const held = new Promise((resolve) => { + release = resolve; + }); + const blocker = db.sql.begin(async (tx) => { + await tx`SELECT 1 FROM memory.document WHERE id = ${documentId} FOR UPDATE`; + await held; + }); + await settle(); + const slow = memory.add({ + ...alice, + externalRef: "slow", + content: { title: "Slow", text: "second draft" }, + }); + await settle(); + const fast = memory.add({ + ...alice, + externalRef: "fast", + content: { title: "Fast", text: "quick note" }, + }); + await settle(); + + const seen = await memory.feed!({ ...alice, after: start.nextCursor ?? 0 }); + release(); + await blocker; + const [slowResult] = await Promise.all([slow, fast]); + const rest = await memory.feed!({ + ...alice, + after: seen.nextCursor ?? start.nextCursor ?? 0, + }); + + const versionIds = [...seen.entries, ...rest.entries].map( + (e) => e.versionId, + ); + expect(versionIds).toContain(slowResult.versionId); + }, 30_000); +}); diff --git a/src/services/capture.ts b/src/services/capture.ts index 53880f2..baafe64 100644 --- a/src/services/capture.ts +++ b/src/services/capture.ts @@ -120,6 +120,9 @@ function parseOptionalDate(value: string | undefined): Date | null { return new Date(value); } +// Namespaces the two-int4 advisory lock; deliberately arbitrary. +const FEED_LOCK_NAMESPACE = 0x3e30_7a13; + async function insertVersion( tx: Tx, input: CaptureInput, @@ -135,6 +138,14 @@ async function insertVersion( ): Promise { const versionId = newId("kver"); const authoritySignals = deriveAuthoritySignals(plan); + if (opts.generation === LIVE_GENERATION) { + // feed_seq is taken at insert but becomes visible at commit. Holding a + // per-tenant lock from here to commit makes live versions commit in + // feed_seq order, so a feed cursor never passes a row still in flight. + await tx.execute( + sql`SELECT pg_advisory_xact_lock(${FEED_LOCK_NAMESPACE}, hashtext(${input.tenantId}))`, + ); + } await tx.insert(memoryVersion).values({ id: versionId, tenantId: input.tenantId,