From e21062cd7dc59f9149a4a11a74f45b8fc0c327cd Mon Sep 17 00:00:00 2001 From: Paul Dlug Date: Fri, 5 Jun 2026 09:12:20 -0700 Subject: [PATCH] fix(vector): provision per-field embedding storage via durable markers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Vector ops no longer issue DDL on the runtime hot path, so a least-privilege (DML-only) Postgres role can read and write embeddings. Previously every vector op lazily ran CREATE TABLE IF NOT EXISTS for its per-(kind,field) table on the request connection, failing with "permission denied for schema public" (42501) under a USAGE-only role — even when the table existed, since Postgres runs the schema aclcheck before the IF NOT EXISTS short-circuit. Vectors now ride the same #135 durable-contribution machinery as fulltext: the privileged migrator (createStoreWithSchema, and evolve() for runtime-added fields) provisions each table + durable marker at boot, and the runtime asserts the marker with a cached SELECT — never DDL. - contribution-materializations: generalized to any contribution set; ensureVectorSlot / assertVectorSlot / dropVectorSlot (+ force, deleteMarker); per-contribution-identity cache - postgres/sqlite: marker-gated vector methods, shared materializer across tx-scoped backends; removed the in-process vectorSlotLatch - store/manager: boot + evolve provision, createVerifiedStore asserts, reembedVectorField re-stamps the marker, reclaim clears it - shared resolveGraphVectorSlots enumerator (one source of truth for boot materialization and runtime assertion) - StoreNotInitializedError now names the actual storage (e.g. vector storage "Doc.embedding") Adds a Postgres least-privilege regression test that reproduces the prod failure and proves a USAGE-only role runs vector DML after the owner provisions. BREAKING CHANGE: vector ops now require a prior privileged createStoreWithSchema (as fulltext already does); a plain createStore + embedding write without provisioning throws StoreNotInitializedError. Migration: run createStoreWithSchema(graph, adminBackend) once under the schema-owner role after upgrading — no GRANT CREATE needed. --- .changeset/marker-gated-vector-storage.md | 37 +++ apps/docs/src/content/docs/architecture.md | 9 +- apps/docs/src/content/docs/backend-setup.md | 43 +-- .../docs/src/content/docs/graph-extensions.md | 8 +- .../docs/src/content/docs/schema-evolution.md | 10 +- .../src/content/docs/schema-management.md | 8 +- apps/docs/src/content/docs/semantic-search.md | 21 +- apps/docs/src/content/docs/troubleshooting.md | 8 +- .../typegraph/examples/11-semantic-search.ts | 7 +- .../examples/12-knowledge-graph-rag.ts | 7 +- .../drizzle/contribution-materializations.ts | 232 ++++++++++++--- .../typegraph/src/backend/drizzle/postgres.ts | 258 +++++++++------- .../typegraph/src/backend/drizzle/sqlite.ts | 277 ++++++++++-------- .../src/backend/drizzle/vector-runtime.ts | 95 ++---- .../typegraph/src/backend/migrate-vectors.ts | 50 +++- packages/typegraph/src/backend/types.ts | 44 ++- packages/typegraph/src/core/embedding.ts | 32 ++ packages/typegraph/src/errors/index.ts | 28 +- packages/typegraph/src/schema/manager.ts | 28 ++ .../src/store/materialize-removals.ts | 16 +- packages/typegraph/src/store/store.ts | 69 ++++- .../backends/postgres/migrate-vectors.test.ts | 9 + .../backends/postgres/pglite-backend.test.ts | 16 +- .../postgres/postgres-backend.test.ts | 19 +- .../postgres/postgres-js-backend.test.ts | 14 +- .../postgres-vector-least-privilege.test.ts | 237 +++++++++++++++ .../reclaim-removed-vector-fields.test.ts | 6 + .../sqlite/libsql-vector-strategy.test.ts | 21 +- .../backends/sqlite/sqlite-backend.test.ts | 49 +++- .../tests/contribution-materializer.test.ts | 166 ++++++++++- .../tests/reembed-vector-field.test.ts | 37 ++- 31 files changed, 1384 insertions(+), 477 deletions(-) create mode 100644 .changeset/marker-gated-vector-storage.md create mode 100644 packages/typegraph/tests/backends/postgres/postgres-vector-least-privilege.test.ts diff --git a/.changeset/marker-gated-vector-storage.md b/.changeset/marker-gated-vector-storage.md new file mode 100644 index 00000000..82f733e0 --- /dev/null +++ b/.changeset/marker-gated-vector-storage.md @@ -0,0 +1,37 @@ +--- +"@nicia-ai/typegraph": minor +--- + +Vector storage now rides the #135 durable-contribution machinery, so the +runtime never issues DDL on the embedding hot path. + +Previously every vector op (`upsertEmbedding` / `deleteEmbedding` / +`vectorSearch` / `createVectorIndex`) lazily ran `CREATE TABLE IF NOT EXISTS` +for its per-`(kind, field)` table on whatever connection it executed on. On a +least-privilege Postgres role (USAGE on `public`, full DML, but no `CREATE`) +this failed with `permission denied for schema public` (SQLSTATE 42501) — even +when the table already existed, because Postgres runs the schema aclcheck before +the `IF NOT EXISTS` short-circuit. The fulltext path already avoided this via +durable markers; vectors now do too. + +What changed: + +- **Boot (privileged):** `createStoreWithSchema` provisions every embedding + `(kind, field)` table + a durable contribution marker, enumerated from the + graph. `evolve()` provisions any embedding fields it introduces. +- **Runtime (DML-only):** the hot path asserts the durable marker with a cached + SELECT and runs DML — never DDL. `createVerifiedStore` verifies vector markers + at attach, alongside fulltext. +- `reembedVectorField` re-stamps the marker after recreating storage at a new + dimension; vector-field reclaim (`materializeRemovals`) clears the marker when + it drops a table. + +**Breaking:** vector ops now require a prior privileged `createStoreWithSchema` +(exactly as fulltext already does). A plain `createStore` + embedding write with +no provisioning step throws `StoreNotInitializedError` instead of lazily +creating the table. + +**Migration:** after upgrading, run `createStoreWithSchema(graph, adminBackend)` +once under the schema-owner role. It creates the per-field vector tables + +markers; least-privilege runtimes then assert markers (SELECT) and run vector +DML with zero DDL — no `GRANT CREATE` required. diff --git a/apps/docs/src/content/docs/architecture.md b/apps/docs/src/content/docs/architecture.md index 6117cf78..b6bcdfb1 100644 --- a/apps/docs/src/content/docs/architecture.md +++ b/apps/docs/src/content/docs/architecture.md @@ -442,9 +442,10 @@ backend.capabilities.vector; // { supported, metrics, indexTypes, maxDimensions, ### Storage Embeddings are stored in **per-field typed tables**, one per `(graphId, nodeKind, fieldPath)`, each carrying -that field's fixed dimension. Tables are created lazily on first write and named -`tg_vec___`. Graph-scoping the table name lets multiple graphs in one database declare the -same kind+field at different dimensions without collision. +that field's fixed dimension. Tables are provisioned by the privileged migrator (`createStoreWithSchema`, and +`evolve()` for runtime-added fields), with a durable contribution marker; the runtime hot path asserts the +marker and never issues DDL. They are named `tg_vec___`. Graph-scoping the table name lets +multiple graphs in one database declare the same kind+field at different dimensions without collision. ```sql -- PostgreSQL with pgvector: one table per (graphId, kind, field). The kind @@ -463,7 +464,7 @@ CREATE INDEX ON tg_vec_my_graph_document_embedding ``` `generatePostgresMigrationSQL()` runs `CREATE EXTENSION IF NOT EXISTS vector` but creates no embedding table — -the per-field tables are created at runtime on first write. +the per-field tables are provisioned by `createStoreWithSchema` at boot (under the privileged role). ### Query Flow diff --git a/apps/docs/src/content/docs/backend-setup.md b/apps/docs/src/content/docs/backend-setup.md index b591d72a..8e88822b 100644 --- a/apps/docs/src/content/docs/backend-setup.md +++ b/apps/docs/src/content/docs/backend-setup.md @@ -32,11 +32,12 @@ const { backend, db } = createLocalSqliteBackend({ path: "./app.db" }); const store = createStore(graph, backend); ``` -:::caution[Fulltext requires `createStoreWithSchema`] -`createLocalSqliteBackend` creates the tables but does not durably -materialize fulltext storage. If your graph has `searchable()` fields, -boot with `const [store] = await createStoreWithSchema(graph, backend);` -instead of bare `createStore()` — otherwise the first fulltext operation +:::caution[Fulltext and embeddings require `createStoreWithSchema`] +`createLocalSqliteBackend` creates the base tables but does not durably +materialize strategy-owned storage. If your graph has `searchable()` or +`embedding()` fields, boot with +`const [store] = await createStoreWithSchema(graph, backend);` instead of +bare `createStore()` — otherwise the first fulltext or embedding operation throws `StoreNotInitializedError`. ::: @@ -99,7 +100,9 @@ const backend = createSqliteBackend(db, { vector: sqliteVecStrategy }); ``` sqlite-vec stores embeddings in `vec0` virtual tables and supports the `cosine` and `l2` metrics. Per-field -vector tables are created lazily on first write — no embedding-table DDL is part of the migration. +vector tables are provisioned by `createStoreWithSchema` at boot (not by the generated migration SQL), and the +runtime asserts a durable marker rather than issuing DDL on first write — see +[Database roles & least privilege](#database-roles--least-privilege). See [Semantic Search](/semantic-search) for query examples. @@ -435,9 +438,9 @@ const db = drizzle(pool); const backend = createPostgresBackend(db); ``` -pgvector stores embeddings in per-field typed `vector(N)` tables (created lazily on first write — the migration -creates no embedding table) with HNSW or IVFFlat indexes, and supports the `cosine`, `l2`, and `inner_product` -metrics. +pgvector stores embeddings in per-field typed `vector(N)` tables (provisioned by `createStoreWithSchema` at boot +— the generated migration SQL creates no embedding table) with HNSW or IVFFlat indexes, and supports the +`cosine`, `l2`, and `inner_product` metrics. See [Semantic Search](/semantic-search) for query examples. @@ -807,19 +810,25 @@ least-privilege, DML-only database role. - **`createStoreWithSchema(graph, backend)` runs DDL.** It bootstraps the base tables on a fresh database, applies safe auto-migrations, and - durably materializes strategy-owned runtime storage (e.g. fulltext). It - re-issues idempotent DDL on every cold boot — at minimum a - `CREATE TABLE IF NOT EXISTS` for the contribution-marker table — so the - role it runs under **must hold `CREATE` / DDL privileges**. Run it once - at startup, outside request handlers and transactions. + durably materializes strategy-owned runtime storage — both fulltext and + each `embedding()` field's per-`(kind, field)` vector table, plus a + durable marker for each. It re-issues idempotent DDL on every cold boot + — at minimum a `CREATE TABLE IF NOT EXISTS` for the contribution-marker + table — so the role it runs under **must hold `CREATE` / DDL + privileges**. Run it once at startup, outside request handlers and + transactions. (`store.evolve()` likewise provisions any embedding field + it introduces, so it too needs DDL privileges.) - **`createStore(graph, backend)` is a synchronous, zero-I/O attach.** It does not create tables, repair DDL, or record that runtime storage is materialized — it issues **no DDL ever**. Use it only to attach to a database a prior `createStoreWithSchema` boot already initialized. A - fulltext read or write against a database that was never initialized - throws `StoreNotInitializedError` rather than silently emitting DDL on - the hot path. Graphs with no `searchable()` fields are unaffected. + fulltext or **embedding** read/write against a database that was never + initialized — a `create({ embedding })` write or a `store.search.vector` + / `.similarTo()` query — throws `StoreNotInitializedError` rather than + silently emitting `CREATE TABLE` on the hot path. This is what lets a + least-privilege role run vector ops: the table already exists. Graphs + with no `searchable()` or `embedding()` fields are unaffected. - **`createVerifiedStore(graph, backend)` is the same zero-DDL attach with a verification gate.** It reads the active schema row, folds the diff --git a/apps/docs/src/content/docs/graph-extensions.md b/apps/docs/src/content/docs/graph-extensions.md index f116e262..8440c88a 100644 --- a/apps/docs/src/content/docs/graph-extensions.md +++ b/apps/docs/src/content/docs/graph-extensions.md @@ -587,8 +587,12 @@ const Manual = defineNode("Manual", { Embeddings live in per-`(graphId, nodeKind, fieldPath)` typed tables named `tg_vec___` (each carrying the field's -fixed dimension), created lazily on first write — there is no single -shared embeddings table. +fixed dimension) — there is no single shared embeddings table. The +privileged migrator provisions each table plus a durable marker: at boot +via `createStoreWithSchema`, and for a field a runtime `evolve()` +introduces, by that `evolve()` call. The runtime hot path then asserts +the marker (never DDL), so a least-privilege role can read/write +embeddings. On `materializeIndexes()`: diff --git a/apps/docs/src/content/docs/schema-evolution.md b/apps/docs/src/content/docs/schema-evolution.md index 77d081c7..8af1588e 100644 --- a/apps/docs/src/content/docs/schema-evolution.md +++ b/apps/docs/src/content/docs/schema-evolution.md @@ -528,11 +528,13 @@ if (result.status === "migrated" && result.toVersion === 3) { ## Reclaiming Removed Embedding Storage -Embeddings live in per-`(graphId, kind, field)` tables (`tg_vec_*`), created -lazily on first write. When you remove an `embedding()` field from a **surviving** +Embeddings live in per-`(graphId, kind, field)` tables (`tg_vec_*`), provisioned +by the privileged migrator (`createStoreWithSchema`, or `evolve()` for a +runtime-added field). When you remove an `embedding()` field from a **surviving** kind, the schema change commits fast but the field's now-orphaned vector table -remains until you reconcile it. `store.materializeRemovals()` drops it (this is -the same pass that cleans up storage for fully removed kinds): +remains until you reconcile it. `store.materializeRemovals()` drops it — and +clears its durable contribution marker so a later re-add re-provisions cleanly +(this is the same pass that cleans up storage for fully removed kinds): ```typescript const result = await store.materializeRemovals(); diff --git a/apps/docs/src/content/docs/schema-management.md b/apps/docs/src/content/docs/schema-management.md index cf002519..e1f5687e 100644 --- a/apps/docs/src/content/docs/schema-management.md +++ b/apps/docs/src/content/docs/schema-management.md @@ -272,9 +272,11 @@ const result = await ensureSchema(backend, graph, { ## Migrating Legacy Embedding Storage Embeddings now live in per-`(graphId, kind, field)` typed tables -(`tg_vec___`), created lazily on first write. This -replaces the single shared `typegraph_node_embeddings` table. New deployments -need no action — the per-field tables materialize as you write vectors. +(`tg_vec___`), provisioned by `createStoreWithSchema` (the +privileged migrator) at boot. This replaces the single shared +`typegraph_node_embeddings` table. New deployments need no action — the per-field +tables are materialized by `createStoreWithSchema`, which the legacy migration +below also relies on having run. Deployments that already hold rows in the legacy table run a one-time, idempotent cutover with `migrateLegacyEmbeddings()`, exported from the package root: diff --git a/apps/docs/src/content/docs/semantic-search.md b/apps/docs/src/content/docs/semantic-search.md index 13c5bbb0..03a2793c 100644 --- a/apps/docs/src/content/docs/semantic-search.md +++ b/apps/docs/src/content/docs/semantic-search.md @@ -149,8 +149,8 @@ import { generatePostgresMigrationSQL } from "@nicia-ai/typegraph/postgres"; // Generates DDL including `CREATE EXTENSION IF NOT EXISTS vector;`. // It does NOT create a single embeddings table — each embedding field gets -// its own typed `vector(N)` table, created lazily on first write (see -// Storage Layout below). +// its own typed `vector(N)` table, provisioned by `createStoreWithSchema` +// (the privileged migrator) at boot (see Storage Layout below). const migrationSQL = generatePostgresMigrationSQL(); ``` @@ -227,12 +227,17 @@ sqlite-vec feature set. Each embedding field is stored in its own typed, graph-scoped table named `tg_vec___`, carrying that field's fixed dimension -(pgvector `vector(N)`, libSQL `F32_BLOB(N)`, sqlite-vec `vec0`). Tables are -created lazily on the first write, so no migration step provisions them. -Graph-scoping means several graphs in one database can declare the same -`kind`+`field` at different dimensions without collision. This is transparent -to queries — `.similarTo()`, `store.search.vector`, and `store.search.hybrid` -read it for you. +(pgvector `vector(N)`, libSQL `F32_BLOB(N)`, sqlite-vec `vec0`). The privileged +migrator (`createStoreWithSchema`, and `evolve()` for runtime-added fields) +provisions each table plus a durable contribution marker at boot; the runtime +hot path then asserts the marker (a cached SELECT) and never issues DDL, so a +least-privilege, DML-only role can read and write embeddings. A vector op +against an un-provisioned slot throws `StoreNotInitializedError` rather than +lazily creating the table — see [Database roles & least +privilege](/backend-setup#database-roles--least-privilege). Graph-scoping means +several graphs in one database can declare the same `kind`+`field` at different +dimensions without collision. This is transparent to queries — `.similarTo()`, +`store.search.vector`, and `store.search.hybrid` read it for you. ### Changing an embedding dimension diff --git a/apps/docs/src/content/docs/troubleshooting.md b/apps/docs/src/content/docs/troubleshooting.md index 6dc32d19..75772535 100644 --- a/apps/docs/src/content/docs/troubleshooting.md +++ b/apps/docs/src/content/docs/troubleshooting.md @@ -366,8 +366,12 @@ markers. See that never materializes runtime storage) against a database that no `createStoreWithSchema()` boot has initialized — commonly the runtime started before the privileged migration step ran, or the wrong role/ -database is configured. `createVerifiedStore()` catches this case at -boot rather than at first hot-path operation. +database is configured. This covers both fulltext and **embedding** +operations: a `store.nodes.*.create({ embedding })` write or a +`store.search.vector` / `.similarTo()` query against an un-provisioned +per-`(kind, field)` table throws here rather than lazily issuing +`CREATE TABLE` on the hot path. `createVerifiedStore()` catches this +case at boot rather than at the first hot-path operation. **Solution:** Run `createStoreWithSchema(graph, adminBackend)` once under the privileged role before the runtime attaches (it writes the diff --git a/packages/typegraph/examples/11-semantic-search.ts b/packages/typegraph/examples/11-semantic-search.ts index 39945872..481f196a 100644 --- a/packages/typegraph/examples/11-semantic-search.ts +++ b/packages/typegraph/examples/11-semantic-search.ts @@ -20,7 +20,7 @@ import { z } from "zod"; import { - createStore, + createStoreWithSchema, defineGraph, defineNode, embedding, @@ -183,7 +183,10 @@ export async function main() { console.log("\n=== Part 2: Storing Documents with Embeddings ===\n"); const backend = createExampleBackend(); - const store = createStore(graph, backend); + // createStoreWithSchema provisions each embedding field's per-(kind, field) + // vector table + durable marker under the (privileged) migrator role, so + // runtime embedding writes/searches never issue DDL. Run it once at boot. + const [store] = await createStoreWithSchema(graph, backend); // Sample documents to index const documents = [ diff --git a/packages/typegraph/examples/12-knowledge-graph-rag.ts b/packages/typegraph/examples/12-knowledge-graph-rag.ts index ec6c04f6..520657df 100644 --- a/packages/typegraph/examples/12-knowledge-graph-rag.ts +++ b/packages/typegraph/examples/12-knowledge-graph-rag.ts @@ -13,7 +13,7 @@ import { z } from "zod"; import { - createStore, + createStoreWithSchema, defineEdge, defineGraph, defineNode, @@ -109,7 +109,10 @@ export async function main() { console.log("=== Knowledge Graph for RAG ===\n"); const backend = createExampleBackend(); - const store = createStore(graph, backend); + // createStoreWithSchema provisions each embedding field's per-(kind, field) + // vector table + durable marker under the (privileged) migrator role, so + // runtime embedding writes/searches never issue DDL. Run it once at boot. + const [store] = await createStoreWithSchema(graph, backend); // ---------------------------------------------------------- // Seed data: Two documents about AI companies diff --git a/packages/typegraph/src/backend/drizzle/contribution-materializations.ts b/packages/typegraph/src/backend/drizzle/contribution-materializations.ts index 593a736b..4dfb6635 100644 --- a/packages/typegraph/src/backend/drizzle/contribution-materializations.ts +++ b/packages/typegraph/src/backend/drizzle/contribution-materializations.ts @@ -22,6 +22,10 @@ import { } from "../../errors"; import { type SqlDialect } from "../../query/dialect"; import { type FulltextStrategy } from "../../query/dialect/fulltext-strategy"; +import { + type VectorSlot, + type VectorStrategy, +} from "../../query/dialect/vector-strategy"; import { sortedReplacer } from "../../schema/canonical"; import { sha256Hex } from "../../utils/hash"; import { isMissingTableError } from "../../utils/sql-errors"; @@ -345,6 +349,14 @@ export type ContributionMaterializerDeps = Readonly<{ dialect: SqlDialect; fulltextStrategy: FulltextStrategy; fulltextTableName: string; + /** + * Active vector strategy, or `undefined` when vector support is + * disabled. When present, its per-`(kind, field)` `ownedTables(slot)` + * contributions ride this same durable-marker machinery — boot + * materializes them under the privileged role, the runtime hot path + * asserts them with a SELECT (never DDL), exactly like fulltext. + */ + vectorStrategy: VectorStrategy | undefined; /** Run one raw DDL statement (dialect's `execute`/`run` of `sql.raw`). */ execDdl: (statement: string) => Promise; /** Idempotently create the `contribution_materializations` table. */ @@ -355,6 +367,15 @@ export type ContributionMaterializerDeps = Readonly<{ recordMarker: ( params: RecordContributionMaterializationParams, ) => Promise; + /** + * Delete a marker row by identity. Used when a contribution's physical + * table is torn down out-of-band (vector-field reclaim) so a later + * re-provision sees "missing" and re-creates the table rather than + * trusting an orphaned "initialized" marker. + */ + deleteMarker: ( + identity: ContributionMaterializationIdentity, + ) => Promise; }>; export type ContributionMaterializer = Readonly<{ @@ -366,18 +387,66 @@ export type ContributionMaterializer = Readonly<{ * the first missing/stale/failed contribution. Zero DDL, zero writes. */ assertInitialized: (graphId: string) => Promise; + /** + * Privileged materializer for one vector slot's `ownedTables` + * contribution(s): creates the per-`(kind, field)` table and records + * its durable marker, idempotently. Pass `{ force: true }` to bypass + * the drift-guard and overwrite the marker at the current signature — + * the sanctioned path for `reembedVectorField`'s deliberate + * dimension change. No-op when vector support is disabled. + */ + ensureVectorSlot: ( + slot: VectorSlot, + options?: Readonly<{ force?: boolean }>, + ) => Promise; + /** + * Hot-path gate for one vector slot: SELECT-only marker assert over + * the slot's `ownedTables` contribution(s), cached per backend + * instance. Throws `StoreNotInitializedError` when the slot is + * missing/stale/failed. No-op when vector support is disabled. + */ + assertVectorSlot: (slot: VectorSlot) => Promise; + /** + * Forget a vector slot: delete its `ownedTables` contribution + * marker(s) and evict the per-instance cache. Called after the slot's + * physical table is dropped (vector-field reclaim) so a future + * `ensureVectorSlot` re-creates the table instead of trusting an + * orphaned marker. No-op when vector support is disabled. + */ + dropVectorSlot: (slot: VectorSlot) => Promise; }>; +// NUL separator for the per-instance contribution cache key: collision-safe +// across arbitrary graph ids / names (a printable delimiter could appear in a +// caller-supplied graph id). +const CONTRIBUTION_KEY_SEPARATOR = String.fromCodePoint(0); + +function contributionKey( + graphId: string, + contribution: StrategyTableContribution, +): string { + return [ + graphId, + contribution.owner, + contribution.logicalName, + contribution.tableName, + ].join(CONTRIBUTION_KEY_SEPARATOR); +} + export function createContributionMaterializer( deps: ContributionMaterializerDeps, ): ContributionMaterializer { const { dialect, fulltextStrategy, fulltextTableName } = deps; - // Positive-only cache: a graph id lands here once every runtime - // contribution's durable marker has been observed materialized for - // it. Missing/stale/failed verdicts are never cached, so a concurrent - // boot that fixes the state is picked up on the next call. - const initializedGraphIds = new Set(); + // Positive-only cache keyed per contribution identity + // (`graphId | owner | logicalName | tableName`): a key lands here once + // its durable marker has been observed materialized at the current + // signature on this connection. Per-contribution (not per-graph) so + // each per-`(kind, field)` vector slot caches independently and the + // hot-path assert stays a pure `Set.has` after the first SELECT. + // Missing/stale/failed verdicts are never cached, so a concurrent boot + // that fixes the state is picked up on the next call. + const initializedKeys = new Set(); function runtimeContributions(): readonly StrategyTableContribution[] { return runtimeStrategyContributions(fulltextStrategy, fulltextTableName); @@ -386,7 +455,9 @@ export function createContributionMaterializer( async function materializeOne( graphId: string, contribution: StrategyTableContribution, + options?: Readonly<{ force?: boolean }>, ): Promise { + const force = options?.force === true; const signature = await computeContributionSignature( dialect, contribution, @@ -395,8 +466,13 @@ export function createContributionMaterializer( const identity = identityOf(graphId, contribution); const existing = await deps.getMarker(identity); - // Already materialized at this exact shape — nothing to do. - if (evaluateContributionState(existing, signature) === "initialized") { + // Already materialized at this exact shape — nothing to do. `force` + // re-runs the DDL and re-stamps the marker even on a match: the path + // `reembedVectorField` relies on after it has recreated the table. + if ( + !force && + evaluateContributionState(existing, signature) === "initialized" + ) { return; } const priorSuccess = existing?.materializedAt !== undefined; @@ -407,7 +483,9 @@ export function createContributionMaterializer( // table. Refuse loudly instead — mirrors the index materializer's // signature-drift handling. (A row with no prior success, or one // whose last attempt errored, falls through and re-runs the DDL.) - if (priorSuccess && existing.signature !== signature) { + // `force` deliberately bypasses this — the caller has already dropped + // and recreated the table at the new shape (`reembedVectorField`). + if (!force && priorSuccess && existing.signature !== signature) { const error = new Error( `Contribution "${contribution.logicalName}" (owner ` + `"${contribution.owner}", table "${contribution.tableName}") was ` + @@ -486,53 +564,69 @@ export function createContributionMaterializer( } /** - * SELECT-only verdict over every runtime contribution: `true` only when - * each one is already materialized at its current signature. Short- - * circuits on the first non-initialized read — a missing marker table - * or any missing/stale/failed contribution. Never runs DDL. + * Privileged materialize over an arbitrary contribution set. Per + * contribution: skip if cached; else (unless `force`) a read-only + * pre-check short-circuits the whole pending set when every marker is + * already initialized at its current signature, so a warm graph stays + * DDL-free — the marker `CREATE TABLE IF NOT EXISTS` itself would fail + * on a connection that can't run DDL (#149). Otherwise ensure the + * marker table and `materializeOne` each pending contribution. `force` + * re-runs the DDL and re-stamps the marker unconditionally (drift-guard + * bypassed) — the `reembedVectorField` recreate path. */ - async function allContributionsInitialized( + async function ensureContributions( graphId: string, contributions: readonly StrategyTableContribution[], - ): Promise { - for (const contribution of contributions) { - const read = await readContributionState(graphId, contribution); - if (read.kind !== "state" || read.state !== "initialized") return false; - } - return true; - } - - async function ensureRuntimeContributions(graphId: string): Promise { - // Cache guard so a redundant boot call (the warm path materializes - // via loadActiveSchemaWithBootstrap, then createStoreWithSchema - // calls again) is a true O(1) no-op rather than a re-SELECT. - if (initializedGraphIds.has(graphId)) return; - const contributions = runtimeContributions(); - - // Read-only pre-check mirroring `assertInitialized` (#149): when every - // contribution is already materialized at its current signature, this - // open needs no DDL — skip `ensureMarkerTable()` / `materializeOne` - // entirely. A consumer that builds a fresh backend per request (empty - // per-instance cache) therefore stays DDL-free on a warm graph instead - // of re-running the marker `CREATE TABLE IF NOT EXISTS` on every open, - // which fails on connections that can't run it. A missing marker table - // or any missing/stale/failed contribution falls through to the - // privileged first-materialization path below, unchanged. - if (await allContributionsInitialized(graphId, contributions)) { - initializedGraphIds.add(graphId); - return; + options?: Readonly<{ force?: boolean }>, + ): Promise { + const force = options?.force === true; + const pending = + force ? + contributions + : contributions.filter( + (contribution) => + !initializedKeys.has(contributionKey(graphId, contribution)), + ); + if (pending.length === 0) return; + + if (!force) { + let allInitialized = true; + for (const contribution of pending) { + const read = await readContributionState(graphId, contribution); + if (read.kind !== "state" || read.state !== "initialized") { + allInitialized = false; + break; + } + } + if (allInitialized) { + for (const contribution of pending) { + initializedKeys.add(contributionKey(graphId, contribution)); + } + return; + } } await deps.ensureMarkerTable(); - for (const contribution of contributions) { - await materializeOne(graphId, contribution); + for (const contribution of pending) { + await materializeOne(graphId, contribution, { force }); + initializedKeys.add(contributionKey(graphId, contribution)); } - initializedGraphIds.add(graphId); } - async function assertInitialized(graphId: string): Promise { - if (initializedGraphIds.has(graphId)) return; - for (const contribution of runtimeContributions()) { + /** + * SELECT-only gate over an arbitrary contribution set: throws + * `StoreNotInitializedError` on the first missing/stale/failed + * contribution (or a never-bootstrapped marker table). Caches each + * confirmed-initialized contribution so the steady state is a pure + * `Set.has`. Never runs DDL or writes. + */ + async function assertContributions( + graphId: string, + contributions: readonly StrategyTableContribution[], + ): Promise { + for (const contribution of contributions) { + const key = contributionKey(graphId, contribution); + if (initializedKeys.has(key)) continue; const read = await readContributionState(graphId, contribution); // A never-bootstrapped database has no marker table — that is // precisely "not initialized". `readContributionState` has already @@ -550,9 +644,51 @@ export function createContributionMaterializer( details: { logicalName: contribution.logicalName }, }); } + initializedKeys.add(key); + } + } + + async function ensureRuntimeContributions(graphId: string): Promise { + await ensureContributions(graphId, runtimeContributions()); + } + + async function assertInitialized(graphId: string): Promise { + await assertContributions(graphId, runtimeContributions()); + } + + async function ensureVectorSlot( + slot: VectorSlot, + options?: Readonly<{ force?: boolean }>, + ): Promise { + if (deps.vectorStrategy === undefined) return; + await ensureContributions( + slot.graphId, + deps.vectorStrategy.ownedTables(slot), + options, + ); + } + + async function assertVectorSlot(slot: VectorSlot): Promise { + if (deps.vectorStrategy === undefined) return; + await assertContributions( + slot.graphId, + deps.vectorStrategy.ownedTables(slot), + ); + } + + async function dropVectorSlot(slot: VectorSlot): Promise { + if (deps.vectorStrategy === undefined) return; + for (const contribution of deps.vectorStrategy.ownedTables(slot)) { + await deps.deleteMarker(identityOf(slot.graphId, contribution)); + initializedKeys.delete(contributionKey(slot.graphId, contribution)); } - initializedGraphIds.add(graphId); } - return { ensureRuntimeContributions, assertInitialized }; + return { + ensureRuntimeContributions, + assertInitialized, + ensureVectorSlot, + assertVectorSlot, + dropVectorSlot, + }; } diff --git a/packages/typegraph/src/backend/drizzle/postgres.ts b/packages/typegraph/src/backend/drizzle/postgres.ts index 86c70aed..bfffb712 100644 --- a/packages/typegraph/src/backend/drizzle/postgres.ts +++ b/packages/typegraph/src/backend/drizzle/postgres.ts @@ -92,6 +92,7 @@ import { import { buildContributionInsertValues, buildContributionOnConflictSet, + type ContributionMaterializer, createContributionMaterializer, gateFulltext, gateFulltextMethods, @@ -139,12 +140,10 @@ import { tables as defaultTables, } from "./schema/postgres"; import { - createVectorSlotLatch, mapVectorWriteError, vectorSlotFromCreateIndexParams, vectorSlotFromDropIndexParams, vectorSlotFromParams, - type VectorSlotLatch, } from "./vector-runtime"; // ============================================================ @@ -301,11 +300,6 @@ export function createPostgresBackend( // default strategy's `vector(N)` DDL would hard-fail. const vectorStrategy = options.vector === false ? undefined : (options.vector ?? pgvectorStrategy); - // One latch per backend instance, shared with every transaction-scoped - // backend so a slot's per-field table is created at most once per process. - // Absent when vector support is disabled. - const vectorSlotLatch = - vectorStrategy === undefined ? undefined : createVectorSlotLatch(); const baseCapabilities = buildPostgresCapabilities( fulltextStrategy, vectorStrategy, @@ -352,72 +346,14 @@ export function createPostgresBackend( tables, fulltextStrategy, ); - const operations = createPostgresOperationBackend({ - db, - executionAdapter, - adapterOptions, - operationStrategy, - tableNames, - capabilities, - fulltextStrategy, - vectorStrategy, - vectorSlotLatch, - }); - - /** - * Runs `fn` inside a Postgres transaction, holding an - * `pg_advisory_xact_lock` keyed on the graph id. The advisory lock - * serializes all schema commits per-graph: the read-then-write CAS in - * `commitSchemaVersion` is safe even for the initial-commit case - * where there is no row yet to `SELECT ... FOR UPDATE`. - * - * Refuses on backends that don't support transactions - * (`drizzle-orm/neon-http`). The orphan-row crash window cannot be - * eliminated without atomicity, so silent best-effort degradation is - * worse than a typed error. - */ - function runSchemaWriteTransaction( - graphId: string, - fn: (tx: CommonOperationBackend) => Promise, - ): Promise { - if (!capabilities.transactions) { - throw new ConfigurationError( - "commitSchemaVersion and setActiveVersion require atomic transactions, " + - "but this Postgres backend does not provide them. The drizzle-orm/neon-http " + - "driver communicates over HTTP and cannot hold a session across statements; " + - "use drizzle-orm/neon-serverless (websocket) for transactional writes.", - { - backend: "postgres", - capability: "transactions", - supportsTransactions: false, - }, - ); - } - return db.transaction(async (tx) => { - // Advisory lock: hashtext($graphId) is collision-tolerant for the - // size of an active graph set; collisions just serialize unrelated - // graphs which is harmless. Held until the transaction commits. - await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${graphId}))`); - // Advisory lock is held here, so the schema-write-capable - // InternalOperationBackend is used intentionally (see its type). - const txBackend = createTransactionBackend({ - db: tx, - adapterOptions, - operationStrategy, - tableNames, - capabilities, - fulltextStrategy, - vectorStrategy, - vectorSlotLatch, - }); - return fn(txBackend); - }); - } - - // Durable fulltext materialization (#135): the dialect-specific + // Durable fulltext + vector materialization (#135): the dialect-specific // marker-table primitives. Orchestration (materialize / assert / - // per-instance cache) lives once in `createContributionMaterializer`. + // per-instance cache) lives once in `createContributionMaterializer`, + // shared by the outer backend and every transaction-scoped backend so a + // slot's marker is resolved at most once per process. Built before + // `operations` so the operation backend's vector methods can assert/ + // ensure through it instead of issuing DDL on the hot path. const matTable = tables.contributionMaterializations; async function ensureContributionMaterializationsTableImpl(): Promise { @@ -471,18 +407,98 @@ export function createPostgresBackend( }); } + async function deleteContributionMaterializationRow( + identity: ContributionMaterializationIdentity, + ): Promise { + await db + .delete(matTable) + .where( + and( + eq(matTable.graphId, identity.graphId), + eq(matTable.logicalName, identity.logicalName), + eq(matTable.owner, identity.owner), + eq(matTable.tableName, identity.tableName), + ), + ); + } + const contributionMaterializer = createContributionMaterializer({ dialect: "postgres", fulltextStrategy, fulltextTableName: tables.fulltextTableName, + vectorStrategy, execDdl: async (statement) => { await db.execute(sql.raw(statement)); }, ensureMarkerTable: ensureContributionMaterializationsTableImpl, getMarker: getContributionMaterializationRow, recordMarker: recordContributionMaterializationRow, + deleteMarker: deleteContributionMaterializationRow, }); + const operations = createPostgresOperationBackend({ + db, + executionAdapter, + adapterOptions, + operationStrategy, + tableNames, + capabilities, + fulltextStrategy, + vectorStrategy, + contributionMaterializer, + }); + + /** + * Runs `fn` inside a Postgres transaction, holding an + * `pg_advisory_xact_lock` keyed on the graph id. The advisory lock + * serializes all schema commits per-graph: the read-then-write CAS in + * `commitSchemaVersion` is safe even for the initial-commit case + * where there is no row yet to `SELECT ... FOR UPDATE`. + * + * Refuses on backends that don't support transactions + * (`drizzle-orm/neon-http`). The orphan-row crash window cannot be + * eliminated without atomicity, so silent best-effort degradation is + * worse than a typed error. + */ + function runSchemaWriteTransaction( + graphId: string, + fn: (tx: CommonOperationBackend) => Promise, + ): Promise { + if (!capabilities.transactions) { + throw new ConfigurationError( + "commitSchemaVersion and setActiveVersion require atomic transactions, " + + "but this Postgres backend does not provide them. The drizzle-orm/neon-http " + + "driver communicates over HTTP and cannot hold a session across statements; " + + "use drizzle-orm/neon-serverless (websocket) for transactional writes.", + { + backend: "postgres", + capability: "transactions", + supportsTransactions: false, + }, + ); + } + + return db.transaction(async (tx) => { + // Advisory lock: hashtext($graphId) is collision-tolerant for the + // size of an active graph set; collisions just serialize unrelated + // graphs which is harmless. Held until the transaction commits. + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext(${graphId}))`); + // Advisory lock is held here, so the schema-write-capable + // InternalOperationBackend is used intentionally (see its type). + const txBackend = createTransactionBackend({ + db: tx, + adapterOptions, + operationStrategy, + tableNames, + capabilities, + fulltextStrategy, + vectorStrategy, + contributionMaterializer, + }); + return fn(txBackend); + }); + } + // Shared by `transaction()` (TypeGraph opens the tx) and // `adoptTransaction()` (#134 — the caller already opened it): bind a // tx-scoped backend to the *literal* `tx` client and gate fulltext on @@ -496,7 +512,7 @@ export function createPostgresBackend( capabilities, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); return gateFulltext(txBackend, contributionMaterializer.assertInitialized); } @@ -660,6 +676,29 @@ export function createPostgresBackend( await contributionMaterializer.ensureRuntimeContributions(graphId); }, + // Vector counterparts of the runtime-contribution methods. Present + // only when a vector strategy is wired (omitted under `vector: false`, + // mirroring the embedding/search methods), so a no-vector backend + // doesn't advertise vector materialization it can't perform. + ...(vectorStrategy === undefined ? + {} + : { + async ensureVectorSlotContribution( + slot: VectorSlot, + options_?: Readonly<{ force?: boolean }>, + ): Promise { + await contributionMaterializer.ensureVectorSlot(slot, options_); + }, + + async assertVectorSlotInitialized(slot: VectorSlot): Promise { + await contributionMaterializer.assertVectorSlot(slot); + }, + + async deleteVectorSlotContribution(slot: VectorSlot): Promise { + await contributionMaterializer.dropVectorSlot(slot); + }, + }), + async getReconciliationMarker( graphId: string, ): Promise { @@ -789,10 +828,13 @@ type CreatePostgresOperationBackendOptions = Readonly<{ */ vectorStrategy: VectorStrategy | undefined; /** - * Shared per-`(kind, field)` storage-ensure latch. Paired with - * `vectorStrategy`: both present, or both `undefined`. + * Shared durable-marker materializer. The vector methods assert a + * slot's marker (SELECT, never DDL) on the hot path and `createVectorIndex` + * ensures it (privileged) — replacing the old in-process ensure-latch. + * Shared across the outer backend and every transaction-scoped backend + * so a slot's marker is resolved at most once per process. */ - vectorSlotLatch: VectorSlotLatch | undefined; + contributionMaterializer: ContributionMaterializer; }>; type CreatePostgresTransactionBackendOptions = Readonly<{ @@ -805,8 +847,8 @@ type CreatePostgresTransactionBackendOptions = Readonly<{ fulltextStrategy: FulltextStrategy; /** Active vector strategy. See {@link CreatePostgresOperationBackendOptions}. */ vectorStrategy: VectorStrategy | undefined; - /** Shared storage-ensure latch. See {@link CreatePostgresOperationBackendOptions}. */ - vectorSlotLatch: VectorSlotLatch | undefined; + /** Shared durable-marker materializer. See {@link CreatePostgresOperationBackendOptions}. */ + contributionMaterializer: ContributionMaterializer; }>; function createPostgresOperationBackend( @@ -821,7 +863,7 @@ function createPostgresOperationBackend( capabilities, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, } = options; // Route through the execution adapter so driver-specific result shapes @@ -921,16 +963,6 @@ function createPostgresOperationBackend( }); } - // Runs the strategy's per-field DDL on this backend's connection (the tx - // client for a transaction-scoped backend) so a slot's table exists - // before the first write/search hits it. - async function ensureVectorSlotStorage(slot: VectorSlot): Promise { - if (vectorStrategy === undefined || vectorSlotLatch === undefined) return; - await vectorSlotLatch.ensure(vectorStrategy, slot, async (statement) => { - await execRun(sql.raw(statement)); - }); - } - const commonBackend = createCommonOperationBackend({ batchConfig: { checkUniqueBatchChunkSize: POSTGRES_CHECK_UNIQUE_BATCH_CHUNK_SIZE, @@ -977,7 +1009,11 @@ function createPostgresOperationBackend( : { async upsertEmbedding(params: UpsertEmbeddingParams): Promise { const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + // Assert the slot's durable marker (SELECT, cached) — never DDL. + // The per-field table is provisioned by the privileged migrator + // (`createStoreWithSchema` → `materializeVectorContributions`), so + // a least-privilege runtime role writes embeddings without CREATE. + await contributionMaterializer.assertVectorSlot(slot); const statements = vectorStrategy.buildUpsert(slot, params, nowIso()); try { for (const statement of statements) { @@ -989,15 +1025,15 @@ function createPostgresOperationBackend( }, async deleteEmbedding(params: DeleteEmbeddingParams): Promise { - // Ensure the per-field table exists before deleting. A delete can + // Assert the slot's durable marker before deleting. A delete can // run before any embedding was ever written for the field (e.g. a - // node hard-deleted having never carried one), and on Postgres a - // DELETE against a missing relation INSIDE a transaction aborts the - // whole transaction — swallowing the JS error can't un-abort it. - // The idempotent ensure makes the DELETE target an existing - // (possibly empty) table, so it's always a clean no-op. + // node hard-deleted having never carried one); the per-field table + // was provisioned at boot, so the DELETE targets an existing + // (possibly empty) table and is a clean no-op — never a DELETE + // against a missing relation, which would abort an enclosing + // Postgres transaction. SELECT-only assert, never DDL. const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + await contributionMaterializer.assertVectorSlot(slot); const statements = vectorStrategy.buildDelete(slot, params); for (const statement of statements) { await execRun(statement); @@ -1012,7 +1048,7 @@ function createPostgresOperationBackend( // before `runVectorSearch` inlines it into `SET LOCAL`. assertPgvectorEfSearch(params.efSearch); const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + await contributionMaterializer.assertVectorSlot(slot); const query = vectorStrategy.buildSearch(slot, params); let rows: readonly { node_id: string; score: number }[]; try { @@ -1044,12 +1080,12 @@ function createPostgresOperationBackend( async createVectorIndex(params: CreateVectorIndexParams): Promise { if (vectorStrategy === undefined) return; const slot = vectorSlotFromCreateIndexParams(params); - // Ensure the per-field table exists first (idempotent), then create its - // ANN index. pgvector's `ownedTables` builds the table only — the HNSW/ - // IVFFlat index is created here (and only here) so it picks up the - // declared `m`/`ef_construction`/`lists` from `slot.indexParams` rather - // than defaults. - await ensureVectorSlotStorage(slot); + // Ensure the per-field table + its durable marker first (privileged, + // idempotent), then create its ANN index. pgvector's `ownedTables` + // builds the table only — the HNSW/IVFFlat index is created here (and + // only here) so it picks up the declared `m`/`ef_construction`/`lists` + // from `slot.indexParams` rather than defaults. + await contributionMaterializer.ensureVectorSlot(slot); // Honor the `concurrent` flag materializeIndexes passes on Postgres so the // ANN build doesn't take a write-blocking lock on a live table. execRun is // autocommit, which CONCURRENTLY requires. @@ -1161,13 +1197,12 @@ function createTransactionBackend( options.executionAdapter ?? createPostgresExecutionAdapter(options.db, options.adapterOptions); - // A transaction-scoped backend gets its OWN per-field ensure-latch, never - // the outer process-global one: a `CREATE TABLE/INDEX` that runs inside the - // caller's transaction and then rolls back must not leave the shared latch - // marking the slot "ensured" (which would skip the re-CREATE and make every - // later write fail with "relation does not exist"). The fresh latch is - // discarded with the transaction, so the next write re-ensures idempotently. - // No latch when vector support is disabled (`vector: false`). + // The transaction-scoped backend shares the outer backend's + // contribution materializer: the per-field vector table is provisioned + // (DDL) only by the privileged outer backend, so a tx-scoped vector op + // only ASSERTS the durable marker (SELECT, never DDL) and can't poison + // anything on rollback. The shared per-instance cache means a slot + // confirmed once stays a pure `Set.has` inside every later transaction. return createPostgresOperationBackend({ db: options.db, executionAdapter: txExecutionAdapter, @@ -1177,8 +1212,7 @@ function createTransactionBackend( capabilities: options.capabilities, fulltextStrategy: options.fulltextStrategy, vectorStrategy: options.vectorStrategy, - vectorSlotLatch: - options.vectorStrategy === undefined ? undefined : createVectorSlotLatch(), + contributionMaterializer: options.contributionMaterializer, }); } diff --git a/packages/typegraph/src/backend/drizzle/sqlite.ts b/packages/typegraph/src/backend/drizzle/sqlite.ts index 7d0defb8..8b9b7c19 100644 --- a/packages/typegraph/src/backend/drizzle/sqlite.ts +++ b/packages/typegraph/src/backend/drizzle/sqlite.ts @@ -81,17 +81,16 @@ import { type SqliteExecutionProfileHints, } from "./execution/sqlite-execution"; import { - createVectorSlotLatch, mapVectorWriteError, vectorSlotFromCreateIndexParams, vectorSlotFromDropIndexParams, vectorSlotFromParams, - type VectorSlotLatch, } from "./vector-runtime"; export type { SqliteTransactionMode } from "./execution/sqlite-execution"; import { buildContributionInsertValues, buildContributionOnConflictSet, + type ContributionMaterializer, createContributionMaterializer, gateFulltext, gateFulltextMethods, @@ -342,12 +341,13 @@ type CreateSqliteOperationBackendOptions = Readonly<{ */ vectorStrategy?: VectorStrategy | undefined; /** - * Per-`(kind, field)` storage-ensure latch shared across the outer - * backend and every transaction-scoped backend (see - * {@link createVectorSlotLatch}). Required whenever `vectorStrategy` is - * set so writes never hit a missing per-field table. + * Shared durable-marker materializer. The vector methods assert a + * slot's marker (SELECT, never DDL) on the hot path and `createVectorIndex` + * ensures it (privileged) — replacing the old in-process ensure-latch. + * Shared across the outer backend and every transaction-scoped backend + * so a slot's marker is resolved at most once per process. */ - vectorSlotLatch?: VectorSlotLatch | undefined; + contributionMaterializer: ContributionMaterializer; }>; type CreateSqliteTransactionBackendOptions = Readonly<{ @@ -360,8 +360,8 @@ type CreateSqliteTransactionBackendOptions = Readonly<{ fulltextStrategy: FulltextStrategy; /** Active vector strategy. See {@link CreateSqliteOperationBackendOptions}. */ vectorStrategy?: VectorStrategy | undefined; - /** Shared storage-ensure latch. See {@link CreateSqliteOperationBackendOptions}. */ - vectorSlotLatch?: VectorSlotLatch | undefined; + /** Shared durable-marker materializer. See {@link CreateSqliteOperationBackendOptions}. */ + contributionMaterializer: ContributionMaterializer; }>; function createSqliteOperationBackend( @@ -376,7 +376,7 @@ function createSqliteOperationBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, } = options; function execGet(query: SQL): Promise { @@ -443,23 +443,16 @@ function createSqliteOperationBackend( }, }; - // The latch runs per-field DDL on the same connection a write/search - // executes on, so a transaction-scoped write materializes its slot inside - // the caller's transaction. - async function ensureVectorSlotStorage(slot: VectorSlot): Promise { - if (vectorStrategy === undefined || vectorSlotLatch === undefined) return; - await vectorSlotLatch.ensure(vectorStrategy, slot, async (statement) => { - await execRun(sql.raw(statement)); - }); - } - const vectorEmbeddingMethods = vectorStrategy === undefined ? {} : { async upsertEmbedding(params: UpsertEmbeddingParams): Promise { const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + // Assert the slot's durable marker (SELECT, cached) — never DDL. + // The per-field table is provisioned by the privileged migrator + // (`createStoreWithSchema` → `materializeVectorContributions`). + await contributionMaterializer.assertVectorSlot(slot); const statements = vectorStrategy.buildUpsert(slot, params, nowIso()); try { for (const statement of statements) { @@ -470,15 +463,15 @@ function createSqliteOperationBackend( } }, async deleteEmbedding(params: DeleteEmbeddingParams): Promise { - // Ensure the per-field table exists before deleting. A delete - // can run before any embedding was ever written for the field - // (e.g. a node hard-deleted having never carried one); the - // idempotent ensure makes the DELETE a clean no-op against an - // existing (possibly empty) table, matching the Postgres path - // (where a DELETE on a missing relation would abort an - // enclosing transaction). + // Assert the slot's durable marker before deleting. A delete can + // run before any embedding was ever written for the field (e.g. a + // node hard-deleted having never carried one); the per-field table + // was provisioned at boot, so the DELETE is a clean no-op against + // an existing (possibly empty) table, matching the Postgres path + // (where a DELETE on a missing relation would abort an enclosing + // transaction). SELECT-only assert, never DDL. const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + await contributionMaterializer.assertVectorSlot(slot); const statements = vectorStrategy.buildDelete(slot, params); for (const statement of statements) { await execRun(statement); @@ -489,7 +482,7 @@ function createSqliteOperationBackend( ): Promise { assertVectorSearchLimit(params.limit); const slot = vectorSlotFromParams(params); - await ensureVectorSlotStorage(slot); + await contributionMaterializer.assertVectorSlot(slot); const query = vectorStrategy.buildSearch(slot, params); let rows: readonly { node_id: string; score: number }[]; try { @@ -519,11 +512,12 @@ function createSqliteOperationBackend( async createVectorIndex(params: CreateVectorIndexParams): Promise { if (vectorStrategy === undefined) return; const slot = vectorSlotFromCreateIndexParams(params); - // Ensure the per-field table exists first (idempotent), then create - // its ANN index. `ownedTables` already folds the index DDL in, so the - // explicit `buildCreateIndex` step is belt-and-suspenders for slots - // whose index intent changed after the table was first materialized. - await ensureVectorSlotStorage(slot); + // Ensure the per-field table + its durable marker first (privileged, + // idempotent), then create its ANN index. `ownedTables` already folds + // the index DDL in, so the explicit `buildCreateIndex` step is + // belt-and-suspenders for slots whose index intent changed after the + // table was first materialized. + await contributionMaterializer.ensureVectorSlot(slot); const indexStatement = vectorStrategy.buildCreateIndex?.(slot); if (indexStatement !== undefined) { await execRun(indexStatement); @@ -639,10 +633,6 @@ export function createSqliteBackend( // (`createLibsqlBackend` always; `createLocalSqliteBackend` when the // extension loads); absent for plain SQLite drivers with no extension. const vectorStrategy = options.vector; - // One latch per backend instance, shared with every transaction-scoped - // backend so a slot's per-field table is created at most once per process. - const vectorSlotLatch = - vectorStrategy === undefined ? undefined : createVectorSlotLatch(); const capabilities: BackendCapabilities = buildSqliteCapabilities({ fulltextStrategy, vectorStrategy, @@ -660,6 +650,96 @@ export function createSqliteBackend( fulltextStrategy, ); const serializedQueue = isSync ? createSerializedExecutionQueue() : undefined; + + // Durable fulltext + vector materialization (#135): the dialect-specific + // marker-table primitives. Orchestration (materialize / assert / + // per-instance cache) lives once in `createContributionMaterializer`, + // shared by the outer backend and every transaction-scoped backend so a + // slot's marker is resolved at most once per process. Built before + // `operations` so the operation backend's vector methods can assert/ + // ensure through it instead of issuing DDL on the hot path. + const matTable = tables.contributionMaterializations; + + async function ensureContributionMaterializationsTableImpl(): Promise { + await db.run(sql.raw(generateSqliteCreateTableSQL(matTable))); + } + + async function getContributionMaterializationRow( + identity: ContributionMaterializationIdentity, + ): Promise { + const rows = await db + .select() + .from(matTable) + .where( + and( + eq(matTable.graphId, identity.graphId), + eq(matTable.logicalName, identity.logicalName), + eq(matTable.owner, identity.owner), + eq(matTable.tableName, identity.tableName), + ), + ); + const row = rows[0]; + if (row === undefined) return undefined; + return mapContributionMaterializationRow( + row, + SQLITE_CONTRIBUTION_MAT_TIMESTAMPS.decode, + ); + } + + async function recordContributionMaterializationRow( + params: RecordContributionMaterializationParams, + ): Promise { + await db + .insert(matTable) + .values( + buildContributionInsertValues( + params, + SQLITE_CONTRIBUTION_MAT_TIMESTAMPS.encode, + ), + ) + .onConflictDoUpdate({ + target: [ + matTable.graphId, + matTable.logicalName, + matTable.owner, + matTable.tableName, + ], + set: buildContributionOnConflictSet( + matTable.materializedAt, + params.materializedAt, + ), + }); + } + + async function deleteContributionMaterializationRow( + identity: ContributionMaterializationIdentity, + ): Promise { + await db + .delete(matTable) + .where( + and( + eq(matTable.graphId, identity.graphId), + eq(matTable.logicalName, identity.logicalName), + eq(matTable.owner, identity.owner), + eq(matTable.tableName, identity.tableName), + ), + ); + } + + const contributionMaterializer = createContributionMaterializer({ + dialect: "sqlite", + fulltextStrategy, + fulltextTableName: tables.fulltextTableName, + vectorStrategy, + execDdl: async (statement) => { + await db.run(sql.raw(statement)); + }, + ensureMarkerTable: ensureContributionMaterializationsTableImpl, + getMarker: getContributionMaterializationRow, + recordMarker: recordContributionMaterializationRow, + deleteMarker: deleteContributionMaterializationRow, + }); + const operations = createSqliteOperationBackend({ capabilities, db, @@ -668,7 +748,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, ...(serializedQueue === undefined ? {} : { serializedQueue }), }); @@ -737,7 +817,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); db.run(sql`BEGIN IMMEDIATE`); try { @@ -768,7 +848,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); return fn(txBackend); }); @@ -791,7 +871,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); return fn(txBackend); }, @@ -800,74 +880,6 @@ export function createSqliteBackend( ); } - // Durable fulltext materialization (#135): the dialect-specific - // marker-table primitives. Orchestration (materialize / assert / - // per-instance cache) lives once in `createContributionMaterializer`. - const matTable = tables.contributionMaterializations; - - async function ensureContributionMaterializationsTableImpl(): Promise { - await db.run(sql.raw(generateSqliteCreateTableSQL(matTable))); - } - - async function getContributionMaterializationRow( - identity: ContributionMaterializationIdentity, - ): Promise { - const rows = await db - .select() - .from(matTable) - .where( - and( - eq(matTable.graphId, identity.graphId), - eq(matTable.logicalName, identity.logicalName), - eq(matTable.owner, identity.owner), - eq(matTable.tableName, identity.tableName), - ), - ); - const row = rows[0]; - if (row === undefined) return undefined; - return mapContributionMaterializationRow( - row, - SQLITE_CONTRIBUTION_MAT_TIMESTAMPS.decode, - ); - } - - async function recordContributionMaterializationRow( - params: RecordContributionMaterializationParams, - ): Promise { - await db - .insert(matTable) - .values( - buildContributionInsertValues( - params, - SQLITE_CONTRIBUTION_MAT_TIMESTAMPS.encode, - ), - ) - .onConflictDoUpdate({ - target: [ - matTable.graphId, - matTable.logicalName, - matTable.owner, - matTable.tableName, - ], - set: buildContributionOnConflictSet( - matTable.materializedAt, - params.materializedAt, - ), - }); - } - - const contributionMaterializer = createContributionMaterializer({ - dialect: "sqlite", - fulltextStrategy, - fulltextTableName: tables.fulltextTableName, - execDdl: async (statement) => { - await db.run(sql.raw(statement)); - }, - ensureMarkerTable: ensureContributionMaterializationsTableImpl, - getMarker: getContributionMaterializationRow, - recordMarker: recordContributionMaterializationRow, - }); - // Shared by the "drizzle" branch of `transaction()` (TypeGraph opens // the tx) and `adoptTransaction()` (#134 — the caller already opened // it): bind a tx-scoped backend to the *literal* `tx` client and gate @@ -881,7 +893,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); return gateFulltext(txBackend, contributionMaterializer.assertInitialized); } @@ -1045,6 +1057,29 @@ export function createSqliteBackend( await contributionMaterializer.ensureRuntimeContributions(graphId); }, + // Vector counterparts of the runtime-contribution methods. Present + // only when a vector strategy is wired (omitted on a plain SQLite + // connection with no vector extension), mirroring the embedding/ + // search methods. + ...(vectorStrategy === undefined ? + {} + : { + async ensureVectorSlotContribution( + slot: VectorSlot, + options_?: Readonly<{ force?: boolean }>, + ): Promise { + await contributionMaterializer.ensureVectorSlot(slot, options_); + }, + + async assertVectorSlotInitialized(slot: VectorSlot): Promise { + await contributionMaterializer.assertVectorSlot(slot); + }, + + async deleteVectorSlotContribution(slot: VectorSlot): Promise { + await contributionMaterializer.dropVectorSlot(slot); + }, + }), + async getReconciliationMarker( graphId: string, ): Promise { @@ -1123,7 +1158,7 @@ export function createSqliteBackend( tableNames, fulltextStrategy, vectorStrategy, - vectorSlotLatch, + contributionMaterializer, }); db.run(sql`BEGIN`); @@ -1205,12 +1240,12 @@ function createTransactionBackend( profileHints: options.profileHints, }); - // A transaction-scoped backend gets its OWN per-field ensure-latch, never - // the outer process-global one. A `CREATE TABLE/INDEX` that runs inside the - // caller's transaction and then rolls back must not leave the shared latch - // marking the slot "ensured" — that would skip the re-CREATE and make every - // later write fail with "no such table". The fresh latch is discarded with - // the transaction, so the next write re-ensures idempotently. + // The transaction-scoped backend shares the outer backend's + // contribution materializer: the per-field vector table is provisioned + // (DDL) only by the privileged outer backend, so a tx-scoped vector op + // only ASSERTS the durable marker (SELECT, never DDL) and can't poison + // anything on rollback. The shared per-instance cache means a slot + // confirmed once stays a pure `Set.has` inside every later transaction. return createSqliteOperationBackend({ capabilities: options.capabilities, db: options.db, @@ -1219,9 +1254,7 @@ function createTransactionBackend( tableNames: options.tableNames, fulltextStrategy: options.fulltextStrategy, vectorStrategy: options.vectorStrategy, - ...(options.vectorStrategy === undefined - ? {} - : { vectorSlotLatch: createVectorSlotLatch() }), + contributionMaterializer: options.contributionMaterializer, }); } diff --git a/packages/typegraph/src/backend/drizzle/vector-runtime.ts b/packages/typegraph/src/backend/drizzle/vector-runtime.ts index 8180fa98..9adc7615 100644 --- a/packages/typegraph/src/backend/drizzle/vector-runtime.ts +++ b/packages/typegraph/src/backend/drizzle/vector-runtime.ts @@ -1,90 +1,31 @@ /** - * Backend-runtime glue between a {@link VectorStrategy} and the Drizzle - * SQLite / PostgreSQL backends. Holds the two responsibilities that belong - * in the backend rather than the strategy: + * Backend-runtime glue between a `VectorStrategy` and the Drizzle + * SQLite / PostgreSQL backends. Holds the slot-construction and + * write-error-mapping responsibilities that belong in the backend rather + * than the strategy: * - * 1. **Slot construction.** The backend stays graph-agnostic: it never - * inspects the graph/registry. The store resolves a field's - * `dimensions` / `metric` / `indexType` from the schema and threads - * them through `UpsertEmbeddingParams` / `VectorSearchParams` / - * `CreateVectorIndexParams`; these helpers reshape those params into the - * `VectorSlot` a strategy addresses its storage with. - * 2. **The storage-ensure latch.** A per-storage-shape - * in-process latch that runs `strategy.ownedTables(slot).createDdl` - * (idempotent `CREATE ... IF NOT EXISTS`) once before the first write to - * a slot, so a write never hits a missing per-field table. Shared across - * the outer backend and every transaction-scoped backend within one - * `createBackend` call. `materializeIndexes()` creates the same storage - * idempotently, so the two never conflict. + * - **Slot construction.** The backend stays graph-agnostic: it never + * inspects the graph/registry. The store resolves a field's + * `dimensions` / `metric` / `indexType` from the schema and threads + * them through `UpsertEmbeddingParams` / `VectorSearchParams` / + * `CreateVectorIndexParams`; these helpers reshape those params into the + * `VectorSlot` a strategy addresses its storage with. + * - **Write-error mapping.** Turns a raw engine dimension-mismatch into a + * typed {@link EmbeddingDimensionChangedError} with actionable guidance. + * + * Per-field table creation is no longer done here: it rides the #135 + * durable-contribution machinery (`createContributionMaterializer`), + * materialized by the privileged migrator at boot and asserted (SELECT, + * never DDL) on the runtime hot path — see `contribution-materializations.ts`. */ import { EmbeddingDimensionChangedError } from "../../errors"; -import { - type VectorSlot, - type VectorStrategy, -} from "../../query/dialect/vector-strategy"; +import { type VectorSlot } from "../../query/dialect/vector-strategy"; import { parseDimensionMismatch } from "../../utils/sql-errors"; import { type CreateVectorIndexParams, type DropVectorIndexParams, } from "../types"; -/** - * In-process per-storage-shape storage-ensure latch. The DDL it runs is - * idempotent, so re-running it is harmless; the latch only avoids redundant - * round-trips and de-duplicates concurrent ensures. - */ -export type VectorSlotLatch = Readonly<{ - ensure: ( - strategy: VectorStrategy, - slot: VectorSlot, - execDdl: (statement: string) => Promise, - ) => Promise; -}>; - -function vectorSlotKey(slot: VectorSlot): string { - // JSON of the storage-shaping fields: collision-safe across arbitrary - // graph/kind/field strings (a raw delimiter could otherwise appear in a name). - // Metric and index type matter even when the physical table name does not: - // libSQL folds DiskANN DDL into `ownedTables` only for ANN slots, and - // sqlite-vec bakes the metric into its virtual-table declaration. - return JSON.stringify([ - slot.graphId, - slot.nodeKind, - slot.fieldPath, - slot.dimensions, - slot.metric, - slot.indexType, - ]); -} - -export function createVectorSlotLatch(): VectorSlotLatch { - const inFlight = new Map>(); - - return { - ensure(strategy, slot, execDdl): Promise { - const key = vectorSlotKey(slot); - const existing = inFlight.get(key); - if (existing !== undefined) return existing; - - const run = (async (): Promise => { - for (const contribution of strategy.ownedTables(slot)) { - for (const statement of contribution.createDdl) { - await execDdl(statement); - } - } - })(); - // Cache only the successful resolution: a failed CREATE (e.g. a - // transient lock) must not poison the slot so the next write retries. - const cached = run.catch((error: unknown) => { - inFlight.delete(key); - throw error; - }); - inFlight.set(key, cached); - return cached; - }, - }; -} - /** * Placeholder dimension for slots built from params that don't carry one * (`DropVectorIndexParams`). The dimension is never read on the drop path — diff --git a/packages/typegraph/src/backend/migrate-vectors.ts b/packages/typegraph/src/backend/migrate-vectors.ts index bc34da85..1573d7c1 100644 --- a/packages/typegraph/src/backend/migrate-vectors.ts +++ b/packages/typegraph/src/backend/migrate-vectors.ts @@ -223,10 +223,20 @@ export async function migrateLegacyEmbeddings( ); } const upsertEmbedding = backend.upsertEmbedding; + const ensureVectorSlotContribution = backend.ensureVectorSlotContribution; const perField: Record = {}; const skippedDimensionMismatch: Record = {}; const skippedDecodeError: Record = {}; + // (kind, field) slots whose per-field table + durable marker this run has + // already provisioned. `upsertEmbedding` asserts the marker (#135) and no + // longer self-creates the table, so this offline migration (privileged) + // provisions each slot once — at the first row's dimension, which fixes the + // typed table just as the original lazy write did. Differently-sized later + // rows for the same slot are skipped by the dimension-mismatch path below, + // so the ensure is deliberately NOT re-run per row (that would trip the + // marker drift-guard). + const ensuredSlots = new Set(); let migrated = 0; let cursor: LegacyRowCursor | undefined; @@ -270,20 +280,36 @@ export async function migrateLegacyEmbeddings( continue; } const config = resolveSlotConfig?.(row.node_kind, row.field_path); + // The decoded length is the authoritative fixed dimension — the + // strategy provisions its column type (e.g. `F32_BLOB(N)`) from it. + const slot = { + graphId: row.graph_id, + nodeKind: row.node_kind, + fieldPath: row.field_path, + dimensions: embedding.length, + metric: config?.metric ?? DEFAULT_SLOT_METRIC, + indexType: config?.indexType ?? DEFAULT_SLOT_INDEX_TYPE, + }; + + // Provision the per-field table + durable marker before the first + // write into this slot. Once per (graph, kind, field): the first row's + // dimension fixes the table. Keyed by graph id too — the legacy table + // is graph-scoped, so the same `kind.field` can recur across graphs. + const ensuredKey = JSON.stringify([ + row.graph_id, + row.node_kind, + row.field_path, + ]); + if ( + ensureVectorSlotContribution !== undefined && + !ensuredSlots.has(ensuredKey) + ) { + await ensureVectorSlotContribution(slot); + ensuredSlots.add(ensuredKey); + } try { - await upsertEmbedding({ - graphId: row.graph_id, - nodeKind: row.node_kind, - nodeId: row.node_id, - fieldPath: row.field_path, - embedding, - // The decoded length is the authoritative fixed dimension — the - // strategy provisions its column type (e.g. `F32_BLOB(N)`) from it. - dimensions: embedding.length, - metric: config?.metric ?? DEFAULT_SLOT_METRIC, - indexType: config?.indexType ?? DEFAULT_SLOT_INDEX_TYPE, - }); + await upsertEmbedding({ ...slot, nodeId: row.node_id, embedding }); } catch (error) { // The legacy shared column allowed mixed dimensions for one // (nodeKind, fieldPath); the per-field table is fixed at the first diff --git a/packages/typegraph/src/backend/types.ts b/packages/typegraph/src/backend/types.ts index 155e5e39..af69224b 100644 --- a/packages/typegraph/src/backend/types.ts +++ b/packages/typegraph/src/backend/types.ts @@ -13,7 +13,10 @@ import { } from "../core/types"; import { type SqlTableNames } from "../query/compiler/schema"; import { type FulltextStrategy } from "../query/dialect/fulltext-strategy"; -import { type VectorStrategy } from "../query/dialect/vector-strategy"; +import { + type VectorSlot, + type VectorStrategy, +} from "../query/dialect/vector-strategy"; import { type SerializedSchema } from "../schema/types"; import { type AnyPgDatabase, @@ -1047,6 +1050,45 @@ export type GraphBackend = Readonly<{ */ ensureRuntimeContributions?: (graphId: string) => Promise; + /** + * Privileged materializer for one embedding `(kind, field)` slot's + * `ownedTables` contribution(s): creates the per-field vector table and + * records its durable marker, idempotently. The vector counterpart of + * `ensureRuntimeContributions` — vectors are per-`(kind, field)` and + * graph-derived, so the store enumerates slots (via + * `resolveEmbeddingFields`) and calls this once per slot at boot under + * the privileged role. Pass `{ force: true }` to overwrite the marker + * at the current signature, bypassing the drift-guard — the sanctioned + * path `store.reembedVectorField()` uses after recreating storage at a + * new dimension. Present only on backends wired with a vector strategy. + */ + ensureVectorSlotContribution?: ( + slot: VectorSlot, + options?: Readonly<{ force?: boolean }>, + ) => Promise; + + /** + * SELECT-only gate for one embedding `(kind, field)` slot: asserts the + * durable marker(s) for the slot's `ownedTables` contribution(s), + * cached per backend instance. Throws `StoreNotInitializedError` when + * the slot is missing, stale (signature drift), or recorded a failed + * last attempt. The vector counterpart of + * `assertRuntimeContributionsInitialized`, consulted by the verified + * runtime attach. Performs ZERO DDL and ZERO writes. Present only on + * backends wired with a vector strategy. + */ + assertVectorSlotInitialized?: (slot: VectorSlot) => Promise; + + /** + * Forget one embedding `(kind, field)` slot's durable contribution + * marker(s). Called after the slot's per-field table is dropped + * (vector-field reclaim) so a later `ensureVectorSlotContribution` + * re-creates the table instead of trusting an orphaned "initialized" + * marker. Does NOT drop the table itself (the caller already did). + * Present only on backends wired with a vector strategy. + */ + deleteVectorSlotContribution?: (slot: VectorSlot) => Promise; + /** * Bootstraps the fulltext storage table the active `FulltextStrategy` * owns. Same focused-bootstrap rationale as the other `ensure*Table` diff --git a/packages/typegraph/src/core/embedding.ts b/packages/typegraph/src/core/embedding.ts index ab8783f2..a5561d92 100644 --- a/packages/typegraph/src/core/embedding.ts +++ b/packages/typegraph/src/core/embedding.ts @@ -6,6 +6,9 @@ */ import { z } from "zod"; +import type { VectorSlot } from "../query/dialect/vector-strategy"; +import type { GraphDef } from "./define-graph"; + // ============================================================ // Embedding Brand Symbol // ============================================================ @@ -331,6 +334,35 @@ export function resolveEmbeddingFields( return fields; } +/** + * Enumerates every embedding `(kind, field)` slot a graph declares, as a + * fully-resolved {@link VectorSlot}. The single source of truth for "what + * vector slots does this graph have?", shared by the privileged boot + * materializer (`materializeVectorContributions`) and the verified-attach + * gate (`assertVectorContributionsInitialized`) so the two can never + * provision and assert different sets. Walks `graph.nodes` and reuses + * {@link resolveEmbeddingFields} per node schema. + */ +export function resolveGraphVectorSlots( + graph: GraphDef, +): readonly VectorSlot[] { + const slots: VectorSlot[] = []; + for (const registration of Object.values(graph.nodes)) { + const node = registration.type; + for (const field of resolveEmbeddingFields(node.schema)) { + slots.push({ + graphId: graph.id, + nodeKind: node.kind, + fieldPath: field.fieldPath, + dimensions: field.dimensions, + metric: field.metric, + indexType: field.indexType, + }); + } + } + return slots; +} + function resolveEmbeddingDimensions(schema: z.ZodType): number | undefined { const direct = getEmbeddingDimensions(schema); if (direct !== undefined) return direct; diff --git a/packages/typegraph/src/errors/index.ts b/packages/typegraph/src/errors/index.ts index c8cfbda8..447cd4a7 100644 --- a/packages/typegraph/src/errors/index.ts +++ b/packages/typegraph/src/errors/index.ts @@ -811,14 +811,14 @@ const STORE_NOT_INITIALIZED_REASON_PHRASE: Readonly< }; /** - * Thrown when a fulltext-dependent operation runs against a connection - * whose strategy-owned storage has not been durably materialized. + * Thrown when a fulltext- or vector-dependent operation runs against a + * connection whose strategy-owned storage has not been durably materialized. * * `createStore()` is a synchronous, zero-I/O attach: it never creates * tables, repairs DDL, or writes materialization markers. The durable * marker is written exclusively by the async boot path - * (`createStoreWithSchema`). When a fulltext read/write — or an - * adopted/business transaction — observes no valid marker, it refuses + * (`createStoreWithSchema`). When a fulltext or embedding read/write — or + * an adopted/business transaction — observes no valid marker, it refuses * loudly here instead of lazily emitting DDL on the hot path. */ export class StoreNotInitializedError extends TypeGraphError { @@ -831,7 +831,8 @@ export class StoreNotInitializedError extends TypeGraphError { options?: { cause?: unknown; details?: Record }, ) { super( - `fulltext storage for graph "${graphId}" ${STORE_NOT_INITIALIZED_REASON_PHRASE[reason]}. ` + + `${storageLabelFromLogicalName(options?.details?.logicalName)} for ` + + `graph "${graphId}" ${STORE_NOT_INITIALIZED_REASON_PHRASE[reason]}. ` + `Run createStoreWithSchema(graph, backend) during application boot, ` + `outside request handlers and adopted transactions, before using createStore().`, "STORE_NOT_INITIALIZED", @@ -849,6 +850,23 @@ export class StoreNotInitializedError extends TypeGraphError { } } +/** + * Human label for the un-materialized storage, derived from the + * contribution `logicalName`: fulltext keeps "fulltext storage"; a vector + * slot ("vector:<kind>.<field>") reads as `vector storage + * "<kind>.<field>"` so the message names the exact embedding + * field. Falls back to a neutral phrase when no logical name is supplied. + */ +function storageLabelFromLogicalName(logicalName: unknown): string { + if (typeof logicalName !== "string") return "runtime storage"; + const VECTOR_PREFIX = "vector:"; + if (logicalName.startsWith(VECTOR_PREFIX)) { + return `vector storage "${logicalName.slice(VECTOR_PREFIX.length)}"`; + } + if (logicalName === "fulltext") return "fulltext storage"; + return `"${logicalName}" storage`; +} + // ============================================================ // Database Errors (category: "system") // ============================================================ diff --git a/packages/typegraph/src/schema/manager.ts b/packages/typegraph/src/schema/manager.ts index 361555f0..4613bc6b 100644 --- a/packages/typegraph/src/schema/manager.ts +++ b/packages/typegraph/src/schema/manager.ts @@ -9,6 +9,7 @@ */ import { type GraphBackend, type SchemaVersionRow } from "../backend/types"; import { type GraphDef } from "../core/define-graph"; +import { resolveGraphVectorSlots } from "../core/embedding"; import { ConfigurationError, DatabaseOperationError, @@ -471,6 +472,7 @@ export async function loadAndVerifyGraph( } await backend.assertRuntimeContributionsInitialized?.(merged.id); + await assertVectorContributionsInitialized(backend, merged); return { graph: merged, activeRow, @@ -478,6 +480,32 @@ export async function loadAndVerifyGraph( }; } +/** + * SELECT-only verification that every embedding `(kind, field)` slot's + * durable contribution marker is initialized — the vector counterpart of + * `backend.assertRuntimeContributionsInitialized` (which covers fulltext). + * Keeps `createVerifiedStore`'s "throws when runtime-contribution markers + * are missing/stale" guarantee honest for vectors: without it the verified + * attach would pass but the first vector op would then throw. Enumerated + * from the merged graph (same idiom as the privileged boot materializer); + * a no-op on backends without vector support or graphs with no embeddings. + */ +async function assertVectorContributionsInitialized( + backend: GraphBackend, + graph: GraphDef, +): Promise { + const assertVectorSlotInitialized = backend.assertVectorSlotInitialized; + if ( + assertVectorSlotInitialized === undefined || + backend.capabilities.vector?.supported !== true + ) { + return; + } + for (const slot of resolveGraphVectorSlots(graph)) { + await assertVectorSlotInitialized(slot); + } +} + function schemaBehindError( graphId: string, fromVersion: number, diff --git a/packages/typegraph/src/store/materialize-removals.ts b/packages/typegraph/src/store/materialize-removals.ts index 71e5056e..3bd42749 100644 --- a/packages/typegraph/src/store/materialize-removals.ts +++ b/packages/typegraph/src/store/materialize-removals.ts @@ -436,7 +436,7 @@ function buildEmbeddingTableCleanup( const slots = await resolveRemovedKindEmbeddingSlots(ctx.backend, row); await Promise.all( slots.map((slot) => - dropVectorSlotStorage(vectorStrategy, executeDdl, slot), + dropVectorSlotStorage(ctx.backend, vectorStrategy, executeDdl, slot), ), ); })(); @@ -452,6 +452,7 @@ function buildEmbeddingTableCleanup( * removed-field reclamation so both reclaim storage the same way. */ async function dropVectorSlotStorage( + backend: GraphBackend, strategy: VectorStrategy, executeDdl: (statement: string) => Promise, slot: VectorSlot, @@ -463,6 +464,12 @@ async function dropVectorSlotStorage( } catch (error) { if (!isMissingTableError(error)) throw error; } + // Forget the slot's durable contribution marker(s) in lockstep with the + // table drop (#135), so a later re-add of the same `(kind, field)` field + // re-creates the table instead of trusting an orphaned "initialized" + // marker. Runs even when the table was already absent — the marker can + // outlive the table. No-op on backends without the vector marker method. + await backend.deleteVectorSlotContribution?.(slot); } /** @@ -611,7 +618,12 @@ async function reclaimRemovedVectorFieldTables( // materialized field has nothing to reclaim), so reaching the catch means // a genuine failure. try { - await dropVectorSlotStorage(vectorStrategy, executeDdl, field.slot); + await dropVectorSlotStorage( + backend, + vectorStrategy, + executeDdl, + field.slot, + ); results.push({ kind: field.kind, fieldPath: field.fieldPath, diff --git a/packages/typegraph/src/store/store.ts b/packages/typegraph/src/store/store.ts index 28f41219..994ccd71 100644 --- a/packages/typegraph/src/store/store.ts +++ b/packages/typegraph/src/store/store.ts @@ -24,7 +24,10 @@ import { isKnownKind, type NodeKinds, } from "../core/define-graph"; -import { resolveEmbeddingFields } from "../core/embedding"; +import { + resolveEmbeddingFields, + resolveGraphVectorSlots, +} from "../core/embedding"; import type { KindEntity, NodeId } from "../core/types"; import { ConfigurationError, @@ -1294,6 +1297,15 @@ export class Store { merged, activeRow.version, ); + // Provision per-field vector tables + durable markers for any embedding + // fields this evolution introduced (idempotent for fields that already + // existed). `evolve()` is a privileged migrator path — it commits schema + // versions and runs index DDL in eager mode — so emitting the table DDL + // here mirrors how `createStoreWithSchema` provisions at first boot, and + // keeps a runtime write to a freshly-added embedding field from hitting + // an unmaterialized slot. The shared fulltext table needs no such step + // (one table for all kinds); vectors are per-`(kind, field)`. + await materializeVectorContributions(this.#backend, merged); const evolved = this.#cloneWithGraph( merged, options?.ref, @@ -1478,18 +1490,26 @@ export class Store { indexType: declaration.indexType, }; - // Recreate storage at the new dimension: drop, then re-emit the table DDL - // the strategy owns (libSQL/sqlite-vec also (re)create their index/vtable - // here; pgvector's index is built via createVectorIndex below). Run via - // executeDdl directly so it bypasses the ensure-latch, which may still - // record the pre-drop slot as ensured. + // Recreate storage at the new dimension: drop, then re-create the per- + // field table the strategy owns (libSQL/sqlite-vec also (re)create their + // index/vtable here; pgvector's index is built via createVectorIndex + // below). `ensureVectorSlotContribution({ force: true })` re-runs the + // table DDL AND re-stamps the durable contribution marker at the new + // signature, bypassing the drift-guard — this is the sanctioned shape + // change. Without the marker reset, the post-reembed runtime assert would + // see a stale marker and refuse every write. Backends without the marker + // method (custom, pre-#135) fall back to raw recreate DDL. for (const ddl of strategy.buildDropStorage(slot)) { await backend.executeDdl(ddl); } - for (const contribution of strategy.ownedTables(slot)) { - for (const ddl of contribution.createDdl) { - await backend.executeDdl(ddl); + if (backend.ensureVectorSlotContribution === undefined) { + for (const contribution of strategy.ownedTables(slot)) { + for (const ddl of contribution.createDdl) { + await backend.executeDdl(ddl); + } } + } else { + await backend.ensureVectorSlotContribution(slot, { force: true }); } // Whether this backend would actually materialize an ANN index for the @@ -2134,6 +2154,33 @@ async function materializeRuntimeContributions( await backend.ensureFulltextTable?.(graphId); } +/** + * Privileged boot step: materialize every embedding `(kind, field)` slot's + * per-field vector table + durable marker, enumerated from the graph via + * {@link resolveGraphVectorSlots}. The vector counterpart of + * {@link materializeRuntimeContributions} (fulltext) — runs once under the + * privileged role inside `createStoreWithSchema` so a least-privilege runtime + * can assert the markers (a cached SELECT) and write embeddings without + * holding `CREATE` on the schema. No-op on backends without vector support + * (`ensureVectorSlotContribution` absent, or `capabilities.vector` unsupported) + * and for graphs that declare no embedding fields. + */ +async function materializeVectorContributions( + backend: GraphBackend, + graph: GraphDef, +): Promise { + const ensureVectorSlotContribution = backend.ensureVectorSlotContribution; + if ( + ensureVectorSlotContribution === undefined || + backend.capabilities.vector?.supported !== true + ) { + return; + } + for (const slot of resolveGraphVectorSlots(graph)) { + await ensureVectorSlotContribution(slot); + } +} + // ============================================================ // Async Factory with Schema Management // ============================================================ @@ -2199,6 +2246,10 @@ export async function createStoreWithSchema( // first — otherwise contribution DDL derived from the new code graph // would hit a stale table shape and mask `MigrationError`. await materializeRuntimeContributions(backend, merged.id); + // Provision every embedding field's per-`(kind, field)` vector table + + // durable marker under the privileged role, so a least-privilege runtime + // asserts the markers (SELECT) and writes embeddings without `CREATE`. + await materializeVectorContributions(backend, merged); const store = new Store( merged, diff --git a/packages/typegraph/tests/backends/postgres/migrate-vectors.test.ts b/packages/typegraph/tests/backends/postgres/migrate-vectors.test.ts index 9c3c3966..98d51eab 100644 --- a/packages/typegraph/tests/backends/postgres/migrate-vectors.test.ts +++ b/packages/typegraph/tests/backends/postgres/migrate-vectors.test.ts @@ -64,6 +64,15 @@ beforeEach(async () => { for (const { tablename } of tables.rows) { await sharedPool.query(`DROP TABLE IF EXISTS "${tablename}" CASCADE`); } + // Drop the durable contribution markers (#135) in lockstep with the per- + // field tables above: `migrateLegacyEmbeddings` now provisions each slot via + // `ensureVectorSlotContribution`, which trusts the marker and skips the + // CREATE when one already exists. On this shared, long-lived database a + // marker left over from a prior run would outlive the dropped table and the + // migration's first write would hit a missing relation. + await sharedPool.query( + `DROP TABLE IF EXISTS "typegraph_contribution_materializations" CASCADE`, + ); }); type LegacySeed = Readonly<{ diff --git a/packages/typegraph/tests/backends/postgres/pglite-backend.test.ts b/packages/typegraph/tests/backends/postgres/pglite-backend.test.ts index 5af58782..70f8168f 100644 --- a/packages/typegraph/tests/backends/postgres/pglite-backend.test.ts +++ b/packages/typegraph/tests/backends/postgres/pglite-backend.test.ts @@ -27,7 +27,7 @@ import { defineGraph, defineNode, embedding } from "../../../src"; import { generatePostgresDDL } from "../../../src/backend/drizzle/ddl"; import { createPostgresBackend } from "../../../src/backend/postgres"; import { createLocalPgliteBackend } from "../../../src/backend/postgres/pglite"; -import { createStore } from "../../../src/store"; +import { createStore, createStoreWithSchema } from "../../../src/store"; const Person = defineNode("Person", { schema: z.object({ name: z.string(), email: z.string().optional() }), @@ -87,7 +87,7 @@ describe("PGlite backend", () => { cleanups.push(() => backend.close()); expect(backend.capabilities.vector?.supported).toBe(true); - const store = createStore(documentsGraph, backend); + const [store] = await createStoreWithSchema(documentsGraph, backend); await store.nodes.Doc.create({ title: "near", embedding: [1, 0, 0, 0] }); await store.nodes.Doc.create({ title: "far", embedding: [0, 0, 0, 1] }); @@ -139,7 +139,7 @@ describe("PGlite backend", () => { cleanups.push(() => backend.close()); expect(backend.capabilities.vector?.supported).toBe(true); - const store = createStore(documentsGraph, backend); + const [store] = await createStoreWithSchema(documentsGraph, backend); await store.nodes.Doc.create({ title: "only", embedding: [0, 1, 0, 0] }); const hits = await store.search.vector("Doc", { fieldPath: "embedding", @@ -155,11 +155,17 @@ describe("PGlite backend", () => { cleanups.push(() => rm(dataDir, { recursive: true, force: true })); // Session 1: write a node + embedding, then close to flush to disk. + // createStoreWithSchema provisions the per-field vector table + marker + // (both persist to the dataDir for session 2 to assert against). const first = await createLocalPgliteBackend({ dataDir }); - const created = await createStore( + const [firstStore] = await createStoreWithSchema( documentsGraph, first.backend, - ).nodes.Doc.create({ title: "persisted", embedding: [1, 0, 0, 0] }); + ); + const created = await firstStore.nodes.Doc.create({ + title: "persisted", + embedding: [1, 0, 0, 0], + }); await first.backend.close(); // Session 2: reopen the same dataDir — the node and its embedding diff --git a/packages/typegraph/tests/backends/postgres/postgres-backend.test.ts b/packages/typegraph/tests/backends/postgres/postgres-backend.test.ts index 8cf3e8f4..df49c187 100644 --- a/packages/typegraph/tests/backends/postgres/postgres-backend.test.ts +++ b/packages/typegraph/tests/backends/postgres/postgres-backend.test.ts @@ -29,7 +29,7 @@ import { createPostgresBackend, createPostgresTables, } from "../../../src/backend/postgres"; -import { createStore } from "../../../src/store"; +import { createStore, createStoreWithSchema } from "../../../src/store"; import { createAdapterTestSuite } from "../adapter-test-suite"; import { createIntegrationTestSuite } from "../integration-test-suite"; @@ -161,17 +161,24 @@ async function clearTestData(): Promise { // parent tables alone leaves orphaned rows that leak into subsequent // integration tests (particularly fulltext search, where orphan rows // can outrank fresh ones and cause missing hits). + // + // The durable contribution markers (#135) are cleared alongside the data so + // each test starts genuinely un-provisioned and the integration suite's + // per-test createStoreWithSchema re-materializes every per-field vector + // table + marker in lockstep. Otherwise a marker could outlive a dropped + // table on this shared database and a later boot would skip the CREATE. await sharedPool.query( `TRUNCATE typegraph_node_fulltext, typegraph_nodes, typegraph_edges, typegraph_node_uniques, + typegraph_contribution_materializations, typegraph_schema_versions CASCADE`, ); - // Per-(kind, field) vector tables are created lazily by the strategy, - // so enumerate and truncate any that exist to keep embedding rows from - // leaking across tests that reuse graph ids. + // Per-(kind, field) vector tables are materialized per field, so enumerate + // and truncate any that exist to keep embedding rows from leaking across + // tests that reuse graph ids; createStoreWithSchema re-creates any missing. await truncatePerFieldVectorTables(sharedPool); } @@ -1639,7 +1646,7 @@ describe("Vector Search End-to-End (Query Builder)", () => { const { db } = requirePostgres(ctx); const backend = createPostgresBackend(db); - const store = createStore(vectorTestGraph, backend); + const [store] = await createStoreWithSchema(vectorTestGraph, backend); // Create documents with embeddings. The store's embedding-sync path // persists each through the pgvector strategy's per-field table — @@ -1693,7 +1700,7 @@ describe("Vector Search End-to-End (Query Builder)", () => { const { db } = requirePostgres(ctx); const backend = createPostgresBackend(db); - const store = createStore(vectorTestGraph, backend); + const [store] = await createStoreWithSchema(vectorTestGraph, backend); // Create documents — embeddings persist through the strategy's // per-field table via the store's embedding-sync path. diff --git a/packages/typegraph/tests/backends/postgres/postgres-js-backend.test.ts b/packages/typegraph/tests/backends/postgres/postgres-js-backend.test.ts index a3ab92bb..c753aecd 100644 --- a/packages/typegraph/tests/backends/postgres/postgres-js-backend.test.ts +++ b/packages/typegraph/tests/backends/postgres/postgres-js-backend.test.ts @@ -92,16 +92,24 @@ async function setupTestDatabase(): Promise { async function clearTestData(): Promise { if (!sharedSql) return; + // Clear the durable contribution markers (#135) alongside the data so each + // test starts genuinely un-provisioned: `createStoreWithSchema` (run per + // test by the integration suite) then re-materializes every per-field + // vector table + marker in lockstep. Leaving markers behind on this shared, + // long-lived test database would let a marker outlive a dropped table, so a + // later boot would trust the marker, skip the CREATE, and writes would hit a + // missing relation. await sharedSql.unsafe( `TRUNCATE typegraph_node_fulltext, typegraph_nodes, typegraph_edges, typegraph_node_uniques, + typegraph_contribution_materializations, typegraph_schema_versions CASCADE`, ); - // Per-(kind, field) vector tables are created lazily by the strategy, - // so enumerate and truncate any that exist (there is no single shared - // embeddings table to TRUNCATE). + // Per-(kind, field) vector tables are materialized per field, so enumerate + // and truncate any that exist (there is no single shared embeddings table). + // createStoreWithSchema re-creates any that are missing (markers cleared). const rows = await sharedSql.unsafe<{ tablename: string }[]>( String.raw`SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename LIKE 'tg_vec\_%'`, diff --git a/packages/typegraph/tests/backends/postgres/postgres-vector-least-privilege.test.ts b/packages/typegraph/tests/backends/postgres/postgres-vector-least-privilege.test.ts new file mode 100644 index 00000000..b6d816c9 --- /dev/null +++ b/packages/typegraph/tests/backends/postgres/postgres-vector-least-privilege.test.ts @@ -0,0 +1,237 @@ +/** + * Regression test for the prod failure that drove the marker-based vector + * gate: a least-privilege Postgres role (USAGE on schema `public`, full DML, + * but NO `CREATE`) must be able to run every runtime vector op — because the + * privileged migrator (`createStoreWithSchema`) provisions each per-`(kind, + * field)` embedding table + durable marker up front, and the runtime hot path + * only ASSERTS the marker (a SELECT) instead of issuing `CREATE TABLE`. + * + * Before the fix, `upsertEmbedding`/`vectorSearch` lazily ran + * `CREATE TABLE IF NOT EXISTS` on the request connection, which a USAGE-only + * role rejects with `permission denied for schema public` (SQLSTATE 42501) — + * even when the table already exists, because Postgres runs the schema + * aclcheck before the `IF NOT EXISTS` short-circuit. + * + * Skipped unless `POSTGRES_URL` is set (or `scripts/test-postgres.sh`). + */ +import { drizzle } from "drizzle-orm/node-postgres"; +import { Pool } from "pg"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, +} from "vitest"; +import { z } from "zod"; + +import { + createStore, + createStoreWithSchema, + createVerifiedStore, + defineGraph, + defineNode, + embedding, + pgvectorStrategy, + StoreNotInitializedError, +} from "../../../src"; +import { generatePostgresMigrationSQL } from "../../../src/backend/drizzle/ddl"; +import { createPostgresBackend } from "../../../src/backend/postgres"; + +const TEST_DATABASE_URL = + process.env.POSTGRES_URL ?? + "postgresql://typegraph:typegraph@127.0.0.1:5432/typegraph_test"; + +const LEAST_PRIV_ROLE = "tg_vec_lp_runtime"; +const LEAST_PRIV_PASSWORD = "lp_runtime_pw"; + +const Document = defineNode("Doc", { + schema: z.object({ + title: z.string(), + embedding: embedding(4), + }), +}); + +const VecGraph = defineGraph({ + // Distinct graph id so this file runs alongside the other postgres suites. + id: "pg_vector_least_privilege", + nodes: { Doc: { type: Document } }, + edges: {}, +}); + +const PER_FIELD_TABLE = pgvectorStrategy.tableName( + VecGraph.id, + "Doc", + "embedding", +); +const CONTRIB_MAT_TABLE = "typegraph_contribution_materializations"; + +let ownerPool: Pool | undefined; +let leastPrivPool: Pool | undefined; +let postgresAvailable = false; + +/** A Pool authenticating as the USAGE-only role against the same database. */ +function leastPrivConnection(): Pool { + const url = new URL(TEST_DATABASE_URL); + return new Pool({ + host: url.hostname, + port: url.port ? Number(url.port) : 5432, + database: url.pathname.replace(/^\//, ""), + user: LEAST_PRIV_ROLE, + password: LEAST_PRIV_PASSWORD, + }); +} + +beforeAll(async () => { + if (!process.env.POSTGRES_URL) return; + try { + ownerPool = new Pool({ connectionString: TEST_DATABASE_URL }); + await ownerPool.query("SELECT 1"); + // Base typegraph_* tables for the shared test database. + await ownerPool.query(generatePostgresMigrationSQL()); + + // (Re)create the least-privilege runtime role: USAGE + full DML, but no + // CREATE on schema public — the exact prod `nicia_app` shape. + await ownerPool.query(`DROP OWNED BY ${LEAST_PRIV_ROLE}`).catch(() => { + // The role may not exist yet (first run) — nothing to drop. + }); + await ownerPool.query(`DROP ROLE IF EXISTS ${LEAST_PRIV_ROLE}`); + await ownerPool.query( + `CREATE ROLE ${LEAST_PRIV_ROLE} LOGIN PASSWORD '${LEAST_PRIV_PASSWORD}'`, + ); + await ownerPool.query(`GRANT USAGE ON SCHEMA public TO ${LEAST_PRIV_ROLE}`); + await ownerPool.query( + `GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO ${LEAST_PRIV_ROLE}`, + ); + // DML on tables the owner creates later (the per-field vector table). + await ownerPool.query( + `ALTER DEFAULT PRIVILEGES IN SCHEMA public ` + + `GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${LEAST_PRIV_ROLE}`, + ); + // The crux: deny DDL. (Legacy databases grant CREATE to PUBLIC.) + await ownerPool.query(`REVOKE CREATE ON SCHEMA public FROM PUBLIC`); + await ownerPool.query( + `REVOKE CREATE ON SCHEMA public FROM ${LEAST_PRIV_ROLE}`, + ); + + leastPrivPool = leastPrivConnection(); + await leastPrivPool.query("SELECT 1"); + postgresAvailable = true; + } catch { + postgresAvailable = false; + } +}); + +afterAll(async () => { + if (leastPrivPool) await leastPrivPool.end(); + if (ownerPool && postgresAvailable) { + // Restore PUBLIC's CREATE so we don't leave the shared test database + // more restrictive than other suites expect, then drop the role. + await ownerPool.query(`GRANT CREATE ON SCHEMA public TO PUBLIC`); + await ownerPool.query(`DROP OWNED BY ${LEAST_PRIV_ROLE}`).catch(() => { + // The role may not exist yet (first run) — nothing to drop. + }); + await ownerPool.query(`DROP ROLE IF EXISTS ${LEAST_PRIV_ROLE}`); + } + if (ownerPool) await ownerPool.end(); +}); + +describe.runIf(process.env.POSTGRES_URL)( + "PostgreSQL vector ops under a least-privilege (USAGE-only) role", + () => { + const transientPools: Pool[] = []; + + function ownerBackend(): ReturnType { + const pool = new Pool({ connectionString: TEST_DATABASE_URL }); + transientPools.push(pool); + return createPostgresBackend(drizzle(pool)); + } + + function runtimeBackend(): ReturnType { + const pool = leastPrivConnection(); + transientPools.push(pool); + return createPostgresBackend(drizzle(pool)); + } + + beforeEach(async () => { + if (!postgresAvailable || !ownerPool) return; + // Start each test genuinely un-provisioned for this graph. + await ownerPool.query(`DROP TABLE IF EXISTS "${PER_FIELD_TABLE}"`); + await ownerPool.query( + `DELETE FROM ${CONTRIB_MAT_TABLE} WHERE graph_id = $1`, + [VecGraph.id], + ); + await ownerPool.query(`DELETE FROM typegraph_nodes WHERE graph_id = $1`, [ + VecGraph.id, + ]); + }); + + afterEach(async () => { + while (transientPools.length > 0) { + const pool = transientPools.pop()!; + await pool.end(); + } + }); + + it("the runtime role genuinely lacks CREATE on schema public (control)", async () => { + await expect( + leastPrivPool!.query(`CREATE TABLE tg_lp_probe (id text)`), + ).rejects.toMatchObject({ code: "42501" }); + }); + + it("owner provisions; the USAGE-only role then upserts + searches with zero DDL", async () => { + // Privileged migrator (owner) creates the per-field table + marker. + await createStoreWithSchema(VecGraph, ownerBackend()); + + // The per-field vector table now exists, owned by the migrator. + const tableExists = await ownerPool!.query<{ exists: boolean }>( + `SELECT EXISTS ( + SELECT 1 FROM pg_tables + WHERE schemaname = 'public' AND tablename = $1 + ) AS exists`, + [PER_FIELD_TABLE], + ); + expect(tableExists.rows[0]?.exists).toBe(true); + + // Runtime attaches as the least-privilege role and verifies markers + // (SELECT-only) — exercising the createVerifiedStore vector gate. + const [store] = await createVerifiedStore(VecGraph, runtimeBackend()); + + // The write that reproduced `permission denied for schema public` + // before the fix now succeeds: assert marker (SELECT) + INSERT (DML). + const near = await store.nodes.Doc.create({ + title: "near", + embedding: [1, 0, 0, 0], + }); + await store.nodes.Doc.create({ title: "far", embedding: [0, 0, 0, 1] }); + + const hits = await store.search.vector("Doc", { + fieldPath: "embedding", + queryEmbedding: [1, 0, 0, 0], + limit: 2, + metric: "cosine", + }); + expect(hits.map((hit) => hit.node.id)).toContain(near.id); + expect(hits[0]?.node.title).toBe("near"); + }); + + it("an un-provisioned slot makes the runtime role throw StoreNotInitializedError, not a DDL permission error", async () => { + // No owner provisioning this time. The hot path asserts the (missing) + // durable marker via SELECT and refuses loudly — it must NOT attempt + // CREATE TABLE and surface a raw 42501. + const store = createStore(VecGraph, runtimeBackend()); + + const error = await store.nodes.Doc.create({ + title: "x", + embedding: [1, 0, 0, 0], + }).catch((error_: unknown) => error_); + + expect(error).toBeInstanceOf(StoreNotInitializedError); + expect((error as StoreNotInitializedError).code).toBe( + "STORE_NOT_INITIALIZED", + ); + }); + }, +); diff --git a/packages/typegraph/tests/backends/postgres/reclaim-removed-vector-fields.test.ts b/packages/typegraph/tests/backends/postgres/reclaim-removed-vector-fields.test.ts index e4daa4ca..c3411dea 100644 --- a/packages/typegraph/tests/backends/postgres/reclaim-removed-vector-fields.test.ts +++ b/packages/typegraph/tests/backends/postgres/reclaim-removed-vector-fields.test.ts @@ -65,8 +65,14 @@ afterAll(async () => { beforeEach(async () => { if (sharedPool === undefined) return; + // Clear the durable contribution markers (#135) alongside dropping the + // per-field tables below: this suite drops `tg_vec_*` tables directly + // (bypassing reclaim), so a marker left behind on this shared database + // would outlive its table and make a later evolve()/createStoreWithSchema + // trust the marker and skip re-creating the table. await sharedPool.query( `TRUNCATE typegraph_index_materializations, + typegraph_contribution_materializations, typegraph_node_uniques, typegraph_nodes, typegraph_edges, diff --git a/packages/typegraph/tests/backends/sqlite/libsql-vector-strategy.test.ts b/packages/typegraph/tests/backends/sqlite/libsql-vector-strategy.test.ts index 44f00e7f..9eb93ee3 100644 --- a/packages/typegraph/tests/backends/sqlite/libsql-vector-strategy.test.ts +++ b/packages/typegraph/tests/backends/sqlite/libsql-vector-strategy.test.ts @@ -212,30 +212,33 @@ describe("libsqlVectorStrategy (executed against @libsql/client)", () => { expect(ids).not.toContain("d2"); // orthogonal vector excluded from top-2 }); - it("backend ensure reruns DiskANN DDL when indexType changes at the same dimension", async () => { + it("provisioning a DiskANN (hnsw) slot materializes its index for vector_top_k via the backend methods", async () => { const { backend } = await createLibsqlBackend(client); - const base = { + const slot = { graphId: GRAPH, nodeKind: "Document", fieldPath: "embedding", dimensions: 3, metric: "cosine" as const, + indexType: "hnsw" as const, }; - // First write ensures only brute-force storage. A later schema change can - // keep the same dimension while switching the slot to ANN-backed `hnsw`; - // the backend must run the hnsw ensure path so `vector_top_k` has an index. + // The privileged migrator provisions the per-field table + DiskANN index + // (libsql folds the index DDL into `ownedTables` for ANN slots) and the + // durable marker. The runtime backend methods then assert the marker + // (never DDL) and `vector_top_k` has its index. A later none→hnsw change + // at the same dimension is a deliberate shape change handled by + // `store.reembedVectorField`, not a lazy re-ensure on the hot path. + await backend.ensureVectorSlotContribution!(slot); await backend.upsertEmbedding!({ - ...base, + ...slot, nodeId: "d1", embedding: [1, 0, 0], - indexType: "none", }); const result = await backend.vectorSearch!({ - ...base, + ...slot, queryEmbedding: [1, 0, 0], - indexType: "hnsw", limit: 1, }); diff --git a/packages/typegraph/tests/backends/sqlite/sqlite-backend.test.ts b/packages/typegraph/tests/backends/sqlite/sqlite-backend.test.ts index f33b005f..0ffc15e2 100644 --- a/packages/typegraph/tests/backends/sqlite/sqlite-backend.test.ts +++ b/packages/typegraph/tests/backends/sqlite/sqlite-backend.test.ts @@ -578,7 +578,8 @@ describe("SQLite embedding persistence via createSqliteBackend", () => { return; } - const { embedding, defineNode, defineGraph } = await import("../../../src"); + const { embedding, defineNode, defineGraph, createStoreWithSchema } = + await import("../../../src"); const Document = defineNode("Doc", { schema: z.object({ title: z.string(), @@ -602,7 +603,9 @@ describe("SQLite embedding persistence via createSqliteBackend", () => { }); expect(backend.upsertEmbedding).toBeDefined(); - const store = createStore(graph, backend); + // createStoreWithSchema provisions the per-field vector table + marker + // (privileged), so the embedding write below asserts cleanly. + const [store] = await createStoreWithSchema(graph, backend); await store.nodes.Doc.create({ title: "Similar", embedding: [1, 0, 0, 0], @@ -745,25 +748,51 @@ describe("SQLite backend.vectorSearch", () => { // Inject the store-resolved slot fields the backend now requires. // sqlite-vec storage is metric-agnostic and brute-force here, so the // defaults (`cosine` / `none` / dimension from the vector length) - // preserve every existing call site's intended behavior. + // preserve every existing call site's intended behavior. Each wrapper + // first provisions the per-field table + durable marker (#135) — the + // privileged step the store's `createStoreWithSchema` would normally do; + // these backend-facade tests drive the backend directly, so they stand + // in for the migrator. Idempotent and cached after the first call. function upsertEmbedding( params: Omit, ): Promise { - return rawUpsertEmbedding({ + const full = { ...params, - metric: "cosine", - indexType: "none", - }); + metric: "cosine" as const, + indexType: "none" as const, + }; + return (async () => { + await backend.ensureVectorSlotContribution?.({ + graphId: full.graphId, + nodeKind: full.nodeKind, + fieldPath: full.fieldPath, + dimensions: full.dimensions, + metric: full.metric, + indexType: full.indexType, + }); + await rawUpsertEmbedding(full); + })(); } function vectorSearch( params: Omit, ): Promise { - return rawVectorSearch({ + const full = { ...params, dimensions: params.queryEmbedding.length, - indexType: "none", - }); + indexType: "none" as const, + }; + return (async () => { + await backend.ensureVectorSlotContribution?.({ + graphId: full.graphId, + nodeKind: full.nodeKind, + fieldPath: full.fieldPath, + dimensions: full.dimensions, + metric: full.metric, + indexType: full.indexType, + }); + return rawVectorSearch(full); + })(); } async function seed( diff --git a/packages/typegraph/tests/contribution-materializer.test.ts b/packages/typegraph/tests/contribution-materializer.test.ts index 6305f090..76473e47 100644 --- a/packages/typegraph/tests/contribution-materializer.test.ts +++ b/packages/typegraph/tests/contribution-materializer.test.ts @@ -13,7 +13,11 @@ */ import { describe, expect, it, vi } from "vitest"; -import { fts5Strategy, StoreNotInitializedError } from "../src"; +import { + fts5Strategy, + pgvectorStrategy, + StoreNotInitializedError, +} from "../src"; import { type ContributionMaterializerDeps, createContributionMaterializer, @@ -23,6 +27,7 @@ import type { ContributionMaterializationRow, RecordContributionMaterializationParams, } from "../src/backend/types"; +import { type VectorStrategy } from "../src/query/dialect/vector-strategy"; const GRAPH_ID = "contrib-mat-unit"; const FULLTEXT_TABLE = "typegraph_node_fulltext"; @@ -95,6 +100,7 @@ function createMockMaterializer( getMarkerOverride?: ( identity: ContributionMaterializationIdentity, ) => Promise, + vectorStrategy?: VectorStrategy, ) { const ensureMarkerTable = vi.fn((): Promise => Promise.resolve()); const execDdl = vi.fn( @@ -113,23 +119,48 @@ function createMockMaterializer( return Promise.resolve(); }, ); + const deleteMarker = vi.fn( + (identity: ContributionMaterializationIdentity): Promise => { + markers.delete(markerKey(identity)); + return Promise.resolve(); + }, + ); const deps = { - dialect: "sqlite", + dialect: vectorStrategy === undefined ? "sqlite" : "postgres", fulltextStrategy: fts5Strategy, fulltextTableName: FULLTEXT_TABLE, + vectorStrategy, execDdl, ensureMarkerTable, getMarker, recordMarker, + deleteMarker, } satisfies ContributionMaterializerDeps; return { materializer: createContributionMaterializer(deps), - spies: { ensureMarkerTable, execDdl, getMarker, recordMarker }, + spies: { + ensureMarkerTable, + execDdl, + getMarker, + recordMarker, + deleteMarker, + }, }; } +// A representative embedding slot for the vector-contribution tests. Its +// physical table + marker identity derive from `pgvectorStrategy.ownedTables`. +const VECTOR_SLOT = { + graphId: GRAPH_ID, + nodeKind: "Document", + fieldPath: "embedding", + dimensions: 8, + metric: "cosine", + indexType: "hnsw", +} as const; + describe("#149 ensureRuntimeContributions is read-only when already materialized", () => { it("a fresh materializer over an already-materialized graph runs zero DDL", async () => { const markers = new Map(); @@ -200,10 +231,12 @@ describe("#149 ensureRuntimeContributions is read-only when already materialized dialect: "sqlite", fulltextStrategy: fts5Strategy, fulltextTableName: FULLTEXT_TABLE, + vectorStrategy: undefined, execDdl, ensureMarkerTable, getMarker, recordMarker, + deleteMarker: vi.fn((): Promise => Promise.resolve()), } satisfies ContributionMaterializerDeps; await createContributionMaterializer(deps).ensureRuntimeContributions( @@ -250,10 +283,12 @@ describe("#149 ensureRuntimeContributions is read-only when already materialized dialect: "postgres", fulltextStrategy: fts5Strategy, fulltextTableName: FULLTEXT_TABLE, + vectorStrategy: undefined, execDdl, ensureMarkerTable, getMarker, recordMarker, + deleteMarker: vi.fn((): Promise => Promise.resolve()), } satisfies ContributionMaterializerDeps; await expect( @@ -321,3 +356,128 @@ describe("assertInitialized verdicts after the readContributionState refactor", expect(rejection).not.toBeInstanceOf(StoreNotInitializedError); }); }); + +describe("vector slot contributions ride the same durable-marker machinery", () => { + it("ensureVectorSlot materializes the per-field table + marker on cold boot", async () => { + const markers = new Map(); + const cold = createMockMaterializer(markers, undefined, pgvectorStrategy); + + await cold.materializer.ensureVectorSlot(VECTOR_SLOT); + + expect(cold.spies.ensureMarkerTable).toHaveBeenCalledTimes(1); + expect(cold.spies.execDdl).toHaveBeenCalled(); + expect(markers.size).toBe(1); + }); + + it("assertVectorSlot throws StoreNotInitializedError before provisioning — never DDL", async () => { + const { materializer, spies } = createMockMaterializer( + new Map(), + () => Promise.reject(MISSING_MARKER_TABLE_ERROR), + pgvectorStrategy, + ); + + await expect( + materializer.assertVectorSlot(VECTOR_SLOT), + ).rejects.toMatchObject({ + name: "StoreNotInitializedError", + details: { reason: "missing" }, + }); + // The hot-path gate never tries to create anything on a USAGE-only role. + expect(spies.execDdl).not.toHaveBeenCalled(); + expect(spies.ensureMarkerTable).not.toHaveBeenCalled(); + }); + + it("assertVectorSlot resolves with zero DDL once a fresh instance reopens a provisioned slot", async () => { + const markers = new Map(); + await createMockMaterializer( + markers, + undefined, + pgvectorStrategy, + ).materializer.ensureVectorSlot(VECTOR_SLOT); + + const warm = createMockMaterializer(markers, undefined, pgvectorStrategy); + await expect( + warm.materializer.assertVectorSlot(VECTOR_SLOT), + ).resolves.toBeUndefined(); + expect(warm.spies.execDdl).not.toHaveBeenCalled(); + expect(warm.spies.ensureMarkerTable).not.toHaveBeenCalled(); + expect(warm.spies.getMarker).toHaveBeenCalled(); + }); + + it("ensureVectorSlot refuses a dimension change as drift, but { force: true } re-stamps it", async () => { + const markers = new Map(); + await createMockMaterializer( + markers, + undefined, + pgvectorStrategy, + ).materializer.ensureVectorSlot(VECTOR_SLOT); + + // A dimension change keeps the same (kind, field) table identity but + // changes the createDdl signature — drift. A fresh instance (no cache) + // must refuse it loudly rather than silently bless a stale table. + const changed = { ...VECTOR_SLOT, dimensions: 16 } as const; + await expect( + createMockMaterializer( + markers, + undefined, + pgvectorStrategy, + ).materializer.ensureVectorSlot(changed), + ).rejects.toThrow(/different signature/); + + // `reembedVectorField`'s path: force past the drift-guard after the + // table was recreated at the new dimension. + await createMockMaterializer( + markers, + undefined, + pgvectorStrategy, + ).materializer.ensureVectorSlot(changed, { force: true }); + + // The marker now reads back as initialized at the new shape. + const warm = createMockMaterializer(markers, undefined, pgvectorStrategy); + await expect( + warm.materializer.assertVectorSlot(changed), + ).resolves.toBeUndefined(); + }); + + it("dropVectorSlot forgets the marker so a later ensure re-creates the table", async () => { + const markers = new Map(); + const first = createMockMaterializer(markers, undefined, pgvectorStrategy); + await first.materializer.ensureVectorSlot(VECTOR_SLOT); + expect(markers.size).toBe(1); + + // Reclaim drops the physical table; dropVectorSlot must delete the marker + // so a stale "initialized" row can't outlive it. + await first.materializer.dropVectorSlot(VECTOR_SLOT); + expect(markers.size).toBe(0); + + // A fresh instance (cold cache) now sees "missing" and the assert refuses. + const warm = createMockMaterializer(markers, undefined, pgvectorStrategy); + await expect( + warm.materializer.assertVectorSlot(VECTOR_SLOT), + ).rejects.toBeInstanceOf(StoreNotInitializedError); + + // ...and a re-provision runs the CREATE again rather than trusting a marker. + const reprovision = createMockMaterializer( + markers, + undefined, + pgvectorStrategy, + ); + await reprovision.materializer.ensureVectorSlot(VECTOR_SLOT); + expect(reprovision.spies.execDdl).toHaveBeenCalled(); + expect(markers.size).toBe(1); + }); + + it("ensureVectorSlot/assertVectorSlot are no-ops when vector support is disabled", async () => { + // No vectorStrategy → the materializer can't address any slot; both + // methods resolve without touching the marker store. + const { materializer, spies } = createMockMaterializer(new Map()); + await expect( + materializer.ensureVectorSlot(VECTOR_SLOT), + ).resolves.toBeUndefined(); + await expect( + materializer.assertVectorSlot(VECTOR_SLOT), + ).resolves.toBeUndefined(); + expect(spies.getMarker).not.toHaveBeenCalled(); + expect(spies.execDdl).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/typegraph/tests/reembed-vector-field.test.ts b/packages/typegraph/tests/reembed-vector-field.test.ts index 095ccf4f..2825fef0 100644 --- a/packages/typegraph/tests/reembed-vector-field.test.ts +++ b/packages/typegraph/tests/reembed-vector-field.test.ts @@ -164,10 +164,11 @@ describe("store.reembedVectorField (sqlite-vec)", () => { ); }); - it("write-time: a wrong-dimension upsert surfaces EmbeddingDimensionChangedError", async () => { + it("write-time: a length-mismatched upsert surfaces EmbeddingDimensionChangedError", async () => { if ( backend.vectorStrategy === undefined || - backend.upsertEmbedding === undefined + backend.upsertEmbedding === undefined || + backend.ensureVectorSlotContribution === undefined ) { return; } @@ -178,21 +179,25 @@ describe("store.reembedVectorField (sqlite-vec)", () => { metric: "cosine" as const, indexType: "none" as const, }; - // First write fixes the per-field table at dimension 3. + // Provision + fix the per-field table at dimension 3 (the privileged step + // the migrator does; this test drives the backend directly). + await backend.ensureVectorSlotContribution({ ...base, dimensions: 3 }); await backend.upsertEmbedding({ ...base, nodeId: "n1", embedding: [1, 0, 0], dimensions: 3, }); - // A later 4-dim write (the field's dimension changed) must surface the - // typed error, not the raw engine "expected N dimensions" message. + // A vector whose length doesn't match the column's fixed dimension must + // surface the typed error (via mapVectorWriteError), not the raw engine + // "expected N dimensions" message. The marker assert passes (dimensions + // still 3); the engine rejects the 4-length vector. await expect( backend.upsertEmbedding({ ...base, nodeId: "n2", embedding: [1, 0, 0, 0], - dimensions: 4, + dimensions: 3, }), ).rejects.toBeInstanceOf(EmbeddingDimensionChangedError); }); @@ -283,9 +288,23 @@ describe("store.reembedVectorField (sqlite-vec)", () => { }); it("backend.vectorSearch rejects a non-finite minScore (#6)", async () => { - if (backend.vectorSearch === undefined) return; - // Direct backend call bypasses the facade — buildSearch must still reject a - // non-finite minScore instead of compiling `distance <= (1 - NaN)`. + if ( + backend.vectorSearch === undefined || + backend.ensureVectorSlotContribution === undefined + ) { + return; + } + // Provision the slot so the marker assert passes and execution reaches + // buildSearch — which must still reject a non-finite minScore instead of + // compiling `distance <= (1 - NaN)`. + await backend.ensureVectorSlotContribution({ + graphId: "g", + nodeKind: "Doc", + fieldPath: "embedding", + dimensions: 3, + metric: "cosine", + indexType: "none", + }); await expect( backend.vectorSearch({ graphId: "g",