diff --git a/.changeset/driver-turso-remote-declared-indexes.md b/.changeset/driver-turso-remote-declared-indexes.md new file mode 100644 index 0000000000..e4fb50a88b --- /dev/null +++ b/.changeset/driver-turso-remote-declared-indexes.md @@ -0,0 +1,21 @@ +--- +'@objectstack/driver-turso': patch +--- + +fix(driver-turso): remote mode materializes every declared object-level index, not only field-level `unique` (#17609) + +## What was wrong + +In remote mode (`libsql://` / `https://`), `TursoDriver` provisions tables through `RemoteTransport`, and the only index DDL that path could emit came from field-level `unique`. An object's declared `indexes: [...]` — unique or not — had no consumer there, so no remote database ever carried one. The local face (`SqlDriver`) created all of them, so nothing failed and no local test noticed: on a remote tenant database `sys_notification_delivery` (five declared indexes) and `sys_job_queue` (three) held only their primary-key autoindex, and the delivery claim query answered every poll with a full table scan (`SCAN sys_notification_delivery` + `USE TEMP B-TREE FOR ORDER BY`). + +## What changes + +- Remote mode now creates **every** declared index: field-level `unique` plus the object's own `indexes`, unique and non-unique, including `unique: 'organization'` with its NULL-safe `COALESCE(, '__global__')` key part. Names and keys come from the same shared normalizers `SqlDriver` and the drift differ use (`uniqueIndexesFromFields`, `normalizeDeclaredIndex`, `buildIndexName`), so both faces land the same index set — pinned by a new local/remote parity suite that compares `sqlite_master` on both. +- New tables get their indexes in the same batch as `CREATE TABLE`. +- **Existing tables are retrofitted on the next schema sync** with `CREATE [UNIQUE] INDEX IF NOT EXISTS`. No row is read-modified or rewritten. +- An index the retrofit cannot create is reported once at `error`, naming the index, the table and the database's own cause. A declared `unique` index over rows that already violate it is **not** forced and no data is repaired: de-duplicate the key's values and re-run schema sync. +- Steady-state cost goes down: a sync now reads the existing index names once (one statement, folded into the column-probe batch it already sends) and issues no index DDL when every declared index exists. Before, every boot re-sent one `CREATE UNIQUE INDEX IF NOT EXISTS` per field-level unique index on an existing table. + +## Upgrading + +Nothing to change in metadata or configuration. The first kernel build after upgrading creates the missing indexes on each existing remote database — on a large table that one build pays the index build time. Watch the boot log for `could not create the declared` lines at `error`: each names an index that is still absent and why. diff --git a/packages/drivers/driver-turso/src/remote-transport-unsafe-identifier-envelope.test.ts b/packages/drivers/driver-turso/src/remote-transport-unsafe-identifier-envelope.test.ts index 4bb698ef48..029a7852ab 100644 --- a/packages/drivers/driver-turso/src/remote-transport-unsafe-identifier-envelope.test.ts +++ b/packages/drivers/driver-turso/src/remote-transport-unsafe-identifier-envelope.test.ts @@ -10,7 +10,7 @@ * where an identifier is INLINED into SQL (SQLite cannot bind one): `object`, * `field` and the `groupBy` field / output key in `aggregate`, the table and * column names in `syncSchema` / `syncSchemasBatch` / `buildCreateTableSQL`, - * and the index name and columns in `syncUniqueIndexes`. It threw a bare + * and the index key columns in `buildDeclaredIndexDDL`. It threw a bare * `Error` — no `code`, no `status` — so `mapDataError` * (`packages/rest/src/error-response.ts`) reached none of its classifying * branches, fell through to its sanitised terminal and served a **500**. A @@ -57,9 +57,12 @@ * a separate card and stays open. [#14235] That card has since landed: the * `groupBy` OUT KEY position is ESCAPED rather than refused now, so its case * below asserts the quoted emission instead of a refusal — every OTHER - * position named above (`object`, `field`, the `groupBy` FIELD, the DDL table, - * column and index names) is untouched and still refuses with this envelope, - * and the accept-set `describe` below still drives the `groupBy` FIELD. + * position named above (`object`, `field`, the `groupBy` FIELD, the DDL table + * and column names, the index key columns) is untouched and still refuses with + * this envelope, and the accept-set `describe` below still drives the `groupBy` + * FIELD. [#17609] An author-declared index NAME moved the same way, for the + * same reason — escaped, not refused; pinned in + * `turso-local-remote-declared-index-parity.test.ts`. * * ## Reverse verification — direction predicted BEFORE it was run * diff --git a/packages/drivers/driver-turso/src/remote-transport.ts b/packages/drivers/driver-turso/src/remote-transport.ts index 7d0383821b..90cf0ed74c 100644 --- a/packages/drivers/driver-turso/src/remote-transport.ts +++ b/packages/drivers/driver-turso/src/remote-transport.ts @@ -46,9 +46,17 @@ import type { DriverQuery } from '@objectstack/spec/contracts'; // face creates are byte-identical to the ones `SqlDriver` creates locally and // the ones the drift differ looks for, so the two faces cannot fork on what the // declaration meant (#6203, which this driver has already paid for twice). +// +// [#17609] …and what an object's declared `indexes: [...]` become, through the +// same normalizer `SqlDriver.syncDeclaredIndexes` and the drift differ read +// (`normalizeDeclaredIndex`). Until then only the field-level half above was +// imported, so every object-level index — unique or not — had no consumer on +// this face and never reached a remote database at all. import { uniqueIndexesFromFields, + normalizeDeclaredIndex, organizationKeyPartSql, + type DeclaredIndexInput, type ExpectedIndex, } from '@objectstack/driver-sql'; import { nanoid } from 'nanoid'; @@ -63,6 +71,23 @@ const DEFAULT_ID_LENGTH = 16; */ const BUILTIN_COLUMNS = new Set(['id', 'created_at', 'updated_at']); +/** + * [#17609] Every index name the database already carries — ONE statement for a + * whole schema sync, read beside the column probe. SQLite index names are + * unique per database, so a name hit is the same "already exists" answer + * `SqlDriver.syncDeclaredIndexes` reads from its own per-table introspection; + * `IF NOT EXISTS` on the DDL still absorbs a concurrent creator. + */ +const EXISTING_INDEX_NAMES_SQL = `SELECT name FROM sqlite_master WHERE type='index'`; + +/** [#17609] One declared index, resolved to the DDL that materializes it. */ +interface PlannedIndex { + name: string; + table: string; + unique: boolean; + sql: string; +} + /** * Pattern for valid SQL identifiers (table and column names). * Prevents SQL injection in DDL statements where parameterized queries @@ -1142,16 +1167,21 @@ export class RemoteTransport { * `error` class, and routing it through the `warn` sink would file it under * exactly the level the rule exists to keep readable. * - * Absent, the degradation is not lost: {@link syncUniqueIndexes} still skips - * the index and the condition resurfaces as the enveloped refusal in - * {@link upsert}. `TursoDriver` wires this to `logger.error` at construction, + * [#17609] A declared NON-unique index that could not be created is the same + * class for the same reason: every query keeps answering, correctly, while + * each one scans the table — nothing is visibly wrong, and the cost arrives + * as read volume. + * + * Absent, the degradation is not lost: {@link retrofitDeclaredIndexes} still + * skips the index, and a missing UNIQUE resurfaces as the enveloped refusal + * in {@link upsert}. `TursoDriver` wires this to `logger.error` at construction, * the same way it wires the connect factory and the temporal column rule. */ private durabilitySink: ((message: string) => void) | null = null; /** - * Register where this transport reports a declared constraint it could not - * materialize (#8413). See {@link durabilitySink} for why it is not the + * Register where this transport reports a declared index it could not + * materialize (#8413, #17609). See {@link durabilitySink} for why it is not the * diagnostic sink. */ setDurabilitySink(sink: (message: string) => void): void { @@ -1822,7 +1852,7 @@ export class RemoteTransport { async syncSchema(object: string, schema: any): Promise { await this.ensureConnected(); - const objectDef = schema as { name: string; fields?: Record }; + const objectDef = schema as { name: string; fields?: Record; indexes?: unknown }; const tableName = object; this.assertSafeIdentifier(tableName); @@ -1835,11 +1865,11 @@ export class RemoteTransport { if (!exists) { await this.client!.execute(this.buildCreateTableSQL(tableName, objectDef)); - // [#8413] The table was created empty microseconds ago, so its unique - // indexes cannot fail on existing data — no isolation needed, and any - // error here is a real DDL fault that should surface. - for (const sql of this.buildUniqueIndexDDL(tableName, objectDef, this.materializedColumns(objectDef))) { - await this.client!.execute(sql); + // [#8413 · #17609] The table was created empty microseconds ago, so its + // declared indexes cannot fail on existing data — no isolation needed, + // and any error here is a real DDL fault that should surface. + for (const index of this.buildDeclaredIndexDDL(tableName, objectDef, this.materializedColumns(objectDef))) { + await this.client!.execute(index.sql); } } else { // ALTER TABLE — add missing columns @@ -1862,10 +1892,15 @@ export class RemoteTransport { materialized.add(name); } } - // [#8413] The retrofit leg — this table may already hold the duplicates - // the missing constraint admitted, so it is isolated and reported, never - // forced. See {@link syncUniqueIndexes}. - await this.syncUniqueIndexes(this.buildUniqueIndexDDL(tableName, objectDef, materialized)); + // [#8413 · #17609] The retrofit leg — this table may already hold the + // duplicates a missing UNIQUE admitted, so it is isolated and reported, + // never forced, and an index the database already carries is not + // re-issued. See {@link retrofitDeclaredIndexes}. + const planned = this.buildDeclaredIndexDDL(tableName, objectDef, materialized); + if (planned.length > 0) { + const existing = await this.client!.execute(EXISTING_INDEX_NAMES_SQL); + await this.retrofitDeclaredIndexes(this.missingIndexes(planned, existing.rows)); + } } } @@ -1888,11 +1923,15 @@ export class RemoteTransport { /** * Batch-synchronize multiple object schemas using batched libsql calls. * - * Collects all DDL statements (CREATE TABLE / ALTER TABLE ADD COLUMN) - * for every schema and uses `client.batch()` to minimize network - * round-trips. The process may perform up to three batch calls: - * one to introspect existing tables, one to introspect columns for - * existing tables, and one to apply DDL statements. + * Collects all DDL statements (CREATE TABLE / ALTER TABLE ADD COLUMN / + * CREATE INDEX) for every schema and uses `client.batch()` to minimize + * network round-trips. The process may perform up to four batch calls: one + * to introspect existing tables, one to introspect the existing tables' + * columns together with the index names the database already carries, one to + * apply DDL statements, and — only when an existing table is missing a + * declared index — one to retrofit those indexes. Once every declared index + * exists (the steady state of every boot after the first) the fourth call is + * not made and no index DDL is sent (#17609). * * This method does not implement an internal fallback to sequential * `syncSchema()`. Any fallback behavior is expected to be handled @@ -1930,45 +1969,52 @@ export class RemoteTransport { const ddlStatements: InStatement[] = []; for (const { object, schema } of newSchemas) { - const objectDef = schema as { name: string; fields?: Record }; + const objectDef = schema as { name: string; fields?: Record; indexes?: unknown }; ddlStatements.push(this.buildCreateTableSQL(object, objectDef)); - // [#8413] Rides the SAME batch, immediately behind its own CREATE TABLE - // (order matters — the index cannot precede the table). A brand-new table - // is empty, so these cannot fail on existing data and need none of the - // isolation the retrofit leg below gets: they belong in the batch, and - // cost this path zero extra round trips. - ddlStatements.push( - ...this.buildUniqueIndexDDL(object, objectDef, this.materializedColumns(objectDef)), - ); + // [#8413 · #17609] Every declared index — field-level `unique` and the + // object's own `indexes`, unique or not — rides the SAME batch, + // immediately behind its own CREATE TABLE (order matters — the index + // cannot precede the table). A brand-new table is empty, so these cannot + // fail on existing data and need none of the isolation the retrofit leg + // below gets: they belong in the batch, and cost this path zero extra + // round trips. + for (const index of this.buildDeclaredIndexDDL(object, objectDef, this.materializedColumns(objectDef))) { + ddlStatements.push(index.sql); + } } - // [#8413] Existing tables' unique indexes are collected here and applied - // AFTER the main batch — they must follow their table's `ALTER TABLE ADD - // COLUMN` (the column may be brand new), and they must not share a - // transaction with it: on a libsql `write` batch one statement's failure - // rolls back every other statement in the batch, so a single table holding - // duplicates would silently undo the schema sync of every OTHER object in - // the boot. That is the blast radius the separation exists to prevent. - const retrofitStatements: string[] = []; - - // Phase 2b: for existing tables, introspect columns in one batch + // [#8413 · #17609] Existing tables' missing declared indexes are collected + // here and applied AFTER the main batch — they must follow their table's + // `ALTER TABLE ADD COLUMN` (the column may be brand new), and they must not + // share a transaction with it: on a libsql `write` batch one statement's + // failure rolls back every other statement in the batch, so a single table + // holding duplicates would silently undo the schema sync of every OTHER + // object in the boot. That is the blast radius the separation exists to + // prevent. + let retrofit: PlannedIndex[] = []; + + // Phase 2b: for existing tables, introspect columns — and, as the LAST + // statement of the same read batch, every index name the database already + // carries (#17609), so the retrofit re-issues only what is missing. That + // one statement is the whole steady-state cost of index sync: it adds no + // round trip, and it does not grow with the number of declared indexes. if (existingSchemas.length > 0) { const pragmaStmts: InStatement[] = existingSchemas.map((s) => ({ sql: `PRAGMA table_info("${s.object}")`, args: [], })); - const pragmaResults = await this.client!.batch(pragmaStmts, 'read'); + const introspection = await this.client!.batch([...pragmaStmts, EXISTING_INDEX_NAMES_SQL], 'read'); + const planned: PlannedIndex[] = []; for (let i = 0; i < existingSchemas.length; i++) { const { object, schema } = existingSchemas[i]; - const objectDef = schema as { name: string; fields?: Record }; - if (!objectDef.fields) continue; + const objectDef = schema as { name: string; fields?: Record; indexes?: unknown }; - const existingColumns = new Set(pragmaResults[i].rows.map((r: any) => r.name)); + const existingColumns = new Set(introspection[i].rows.map((r: any) => r.name)); const materialized = new Set(BUILTIN_COLUMNS); for (const c of existingColumns) materialized.add(String(c)); - for (const [name, field] of Object.entries(objectDef.fields)) { + for (const [name, field] of Object.entries(objectDef.fields ?? {})) { if (existingColumns.has(name)) continue; const type = (field as any).type || 'string'; if (type === 'formula') continue; @@ -1978,8 +2024,9 @@ export class RemoteTransport { materialized.add(name); } - retrofitStatements.push(...this.buildUniqueIndexDDL(object, objectDef, materialized)); + planned.push(...this.buildDeclaredIndexDDL(object, objectDef, materialized)); } + retrofit = this.missingIndexes(planned, introspection[existingSchemas.length].rows); } // Phase 3: execute all DDL in a single batch @@ -1987,10 +2034,11 @@ export class RemoteTransport { await this.client!.batch(ddlStatements, 'write'); } - // Phase 4 [#8413]: retrofit the existing tables' declared unique indexes, - // outside the batch above for the blast-radius reason stated at its - // declaration. Failures here are reported, never forced and never repaired. - await this.syncUniqueIndexes(retrofitStatements); + // Phase 4 [#8413 · #17609]: retrofit the existing tables' missing declared + // indexes, outside the batch above for the blast-radius reason stated at + // its declaration. Failures here are reported, never forced and never + // repaired. + await this.retrofitDeclaredIndexes(retrofit); } async dropTable(object: string): Promise { @@ -2043,7 +2091,7 @@ export class RemoteTransport { * identifier — `object`, `field` and the `groupBy` `outKey` in * {@link RemoteTransport.aggregate}, the table and column names in * `syncSchema` / `syncSchemasBatch` / `buildCreateTableSQL`, and the index - * name and columns in `syncUniqueIndexes` — so the envelope is decided once + * key columns in `buildDeclaredIndexDDL` — so the envelope is decided once * for all of them. See {@link unsafeIdentifierError} for which envelope and * why. The predicate and the message are unchanged. */ @@ -2094,6 +2142,9 @@ export class RemoteTransport { * `alias` is escaped rather than concatenated. * * ⛔ NOT for column references — those keep {@link assertSafeIdentifier}. + * + * [#17609] It also spells a declared index NAME in + * {@link buildDeclaredIndexDDL}: one name, never a reference. */ private aliasIdentifierSql(alias: string): string { return `"${String(alias).replace(/"/g, '""')}"`; @@ -2122,8 +2173,23 @@ export class RemoteTransport { } /** - * [#8413] The UNIQUE indexes a schema's field-level `unique` declarations ask - * for, as executable DDL. + * [#8413 · #17609] Every index an object's metadata declares, as executable + * DDL: field-level `unique` (tenancy-aware) PLUS the object's own + * `indexes: [...]`, unique or not. + * + * # One answer, three readers + * + * The set is composed from the two shared normalizers that + * `SqlDriver.syncDeclaredIndexes` and the drift differ's `expectedIndexes` + * compose — `uniqueIndexesFromFields` for field-level `unique`, + * `normalizeDeclaredIndex` for each declared entry — so the index NAME + * (`buildIndexName`, or the author's own `name`) and the KEY this face + * creates are the ones the local face creates and the differ looks for. + * Nothing about naming, scope or key order is decided here. Until #17609 only + * the first normalizer was read, so every object-level index had no consumer + * on this face and no remote database carried one — `sqlite_master` on a + * production tenant held the primary-key autoindexes and nothing else, and + * the hot polling tables answered every claim query with a full scan. * * # Why a companion index and not an inline `UNIQUE` column constraint * @@ -2134,121 +2200,162 @@ export class RemoteTransport { * 1. **Retrofit.** SQLite cannot add a column constraint to an existing * table — `ALTER TABLE` has no `ADD CONSTRAINT`, so an inline `UNIQUE` * reaches an already-created table only through a full table rebuild - * (create-copy-drop-rename). A `CREATE UNIQUE INDEX` is a single - * statement that touches no row. Since the tables this defect has been - * filling with duplicates all already exist, the inline form would have - * made the fix unreachable exactly where it is needed. - * 2. **Parity.** `SqlDriver` materializes field-level `unique` as a UNIQUE - * INDEX (`syncDeclaredIndexes`), never inline. Matching it means the two - * faces converge on the same index NAME (`buildIndexName`) and the same - * key, so `sqlite_master` on a remote database and on a local one read - * alike — and the drift differ, which looks for those names, does not - * report a remote database as drifted from its own declaration. + * (create-copy-drop-rename). A `CREATE [UNIQUE] INDEX` is a single + * statement that touches no row. Since the tables these indexes were + * missing from all already exist, the inline form would have made the + * fix unreachable exactly where it is needed. + * 2. **Parity.** `SqlDriver` materializes every declared index as an INDEX, + * never inline, so `sqlite_master` on a remote database and on a local + * one read alike — and the drift differ, which looks for those names, + * does not report a remote database as drifted from its own declaration. * * # NULL semantics are inherited, not chosen here * * SQL UNIQUE is NULL-distinct, so rows with a NULL in the key stay mutually - * unconstrained; the tenant-scoped arm gets the NULL-SAFE key part + * unconstrained; an organization-scoped unique gets the NULL-SAFE key part * (`COALESCE(, '__global__')`) from the shared helper for the reason * ADR-0120 D3 records. Neither rule is re-decided here. * * Columns that were never materialized (a virtual `formula` field) are * skipped rather than emitted — the same choice `SqlDriver.syncDeclaredIndexes` * makes, and for the same reason: DDL naming a column that does not exist - * fails the whole sync over an index nothing could have used. + * fails the whole sync over an index nothing could have used. A name declared + * twice is emitted once, as the local face creates it once. */ - private buildUniqueIndexDDL( + private buildDeclaredIndexDDL( tableName: string, - objectDef: { fields?: Record; tenancy?: any }, + objectDef: { fields?: Record; tenancy?: any; indexes?: unknown }, materializedColumns: Set, - ): string[] { + ): PlannedIndex[] { const tenantField = this.tenantFieldResolver ? this.tenantFieldResolver(objectDef) : null; const expected: ExpectedIndex[] = uniqueIndexesFromFields( tableName, objectDef.fields ?? {}, tenantField, ); + const declared = Array.isArray(objectDef.indexes) ? (objectDef.indexes as DeclaredIndexInput[]) : []; + for (const entry of declared) { + const normalized = normalizeDeclaredIndex(tableName, entry, tenantField); + if (normalized) expected.push(normalized); + } - const statements: string[] = []; + const planned: PlannedIndex[] = []; + const emitted = new Set(); for (const index of expected) { const missing = index.columns.filter((c) => !materializedColumns.has(c)); if (missing.length > 0) { this.diagnosticSink?.( - `[RemoteTransport] skipping declared unique index on "${tableName}" — ` + + `[RemoteTransport] skipping declared index "${index.name}" on "${tableName}" — ` + `column(s) not materialized: ${missing.join(', ')}`, ); continue; } - this.assertSafeIdentifier(index.name); + if (emitted.has(index.name)) continue; + // The KEY columns are references and stay gated. The index NAME is not a + // reference but one name by definition — the class #14113 escapes for an + // alias rather than refuses. It is the one position here an author types + // freely (`IndexSchema.name` is any string, and the local face quotes + // whatever it is given), so refusing it would fail this face's WHOLE + // schema sync over a declaration the other face accepts. for (const column of index.columns) this.assertSafeIdentifier(column); + const nameSql = this.aliasIdentifierSql(index.name); const nullSafe = new Set(index.nullSafeColumns ?? []); const parts = index.columns.map((c) => nullSafe.has(c) ? organizationKeyPartSql(`"${c}"`) : `"${c}"`, ); - // `IF NOT EXISTS` is what makes every sync after the first a no-op - // server-side, so re-syncing an object costs a statement rather than an - // error — and it is also what makes the per-statement retry in - // {@link syncUniqueIndexes} safe under EITHER libsql batch semantic + // `IF NOT EXISTS` is what makes a concurrent creator (two instances + // booting against one database) a no-op rather than an error — and it is + // also what makes the per-statement retry in + // {@link retrofitDeclaredIndexes} safe under EITHER libsql batch semantic // (transactional: nothing was applied; non-transactional: re-applying is // a no-op). - statements.push( - `CREATE UNIQUE INDEX IF NOT EXISTS "${index.name}" ON "${tableName}" (${parts.join(', ')})`, - ); + planned.push({ + name: index.name, + table: tableName, + unique: index.unique, + sql: + `CREATE ${index.unique ? 'UNIQUE ' : ''}INDEX IF NOT EXISTS ${nameSql} ` + + `ON "${tableName}" (${parts.join(', ')})`, + }); + emitted.add(index.name); } - return statements; + return planned; + } + + /** + * [#17609] The planned indexes the database does not already carry, by name — + * the remote twin of the `existing.has(name)` skip in + * `SqlDriver.syncDeclaredIndexes`. `rows` is the answer to + * {@link EXISTING_INDEX_NAMES_SQL}. + */ + private missingIndexes(planned: PlannedIndex[], rows: ReadonlyArray): PlannedIndex[] { + const existing = new Set(rows.map((row) => String((row as { name?: unknown }).name))); + return planned.filter((index) => !existing.has(index.name)); } /** - * [#8413] Materialize unique indexes against a table that ALREADY EXISTS — - * the retrofit path, and the one that can legitimately fail. + * [#8413 · #17609] Materialize declared indexes against tables that ALREADY + * EXIST — the retrofit path, and the one that can legitimately fail. * * ⛔ **This must never repair data, and never gives up quietly.** Creating a * UNIQUE index over a table that already holds duplicates fails, and those - * duplicates are precisely what this defect has been producing. Deleting, + * duplicates are precisely what a missing constraint admits. Deleting, * merging or rewriting any of those rows is a destructive migration and an - * operator's decision — never a side effect of a driver booting. So the only - * two outcomes here are *the index now exists* and *the index does not exist - * and somebody was told at `error`*, which is the level AGENTS.md's - * degradation rule requires for a declared constraint that is not enforced: - * from the outside nothing looks wrong, and the loss surfaces a release later. - * - * Reporting it is not the whole remedy, and is not meant to be — the operator - * with duplicates also gets {@link refuseUnbackedConflictTarget} on any - * `conflictKeys` upsert against the same table, which is a refusal at the - * moment of use rather than a log line at boot. - * - * **Round-trip cost, stated rather than hidden** (#7099 asked which trips are - * already paid): the happy path is ONE extra batch per sync that touches an - * existing table, and only when that table declares a unique field at all. - * The per-statement fallback runs only after a batch has already failed, i.e. - * only on a database that really does have a violated constraint — so the - * cost of precision is paid by the deployment that needs the diagnosis, not - * by every boot. + * operator's decision — never a side effect of a driver booting. A plain + * index cannot fail on data, but it can still fail (a server-side limit, a + * timeout on a large table). So the only two outcomes here are *the index + * now exists* and *the index does not exist and somebody was told at + * `error`*, which is the level AGENTS.md's degradation rule requires: DDL the + * metadata declares did not run, and from the outside nothing looks wrong — + * a missing UNIQUE keeps admitting duplicates, a missing access path keeps + * answering every query by scanning the table. Both surface a release later, + * the second one as read volume. + * + * Reporting a missing UNIQUE is not the whole remedy, and is not meant to be + * — the operator with duplicates also gets {@link refuseUnbackedConflictTarget} + * on any `conflictKeys` upsert against the same table, which is a refusal at + * the moment of use rather than a log line at boot. + * + * **Round-trip cost, stated rather than hidden:** `planned` holds only the + * indexes the database does not already carry — both callers filter by name + * first — so the steady state issues NO statement here and no round trip. A + * boot that does have indexes to add pays ONE batch for all of them. The + * per-statement fallback runs only after that batch has failed, i.e. only on + * a database where some index really is unbuildable, so the cost of naming + * the culprit is paid by the deployment that needs the diagnosis, not by + * every boot. On a transactional libsql `write` batch one failure rolls back + * its siblings as well; the fallback is also what lands those. */ - private async syncUniqueIndexes(statements: string[]): Promise { - if (statements.length === 0) return; + private async retrofitDeclaredIndexes(planned: PlannedIndex[]): Promise { + if (planned.length === 0) return; try { - await this.client!.batch(statements, 'write'); + await this.client!.batch(planned.map((index) => index.sql), 'write'); return; } catch { // The batch told us SOMETHING failed, not which. Re-issue one at a time // so the report names the index an operator has to act on — `IF NOT // EXISTS` makes the ones that already succeeded no-ops either way. } - for (const sql of statements) { + for (const index of planned) { try { - await this.client!.execute(sql); + await this.client!.execute(index.sql); } catch (e) { + const cause = e instanceof Error ? e.message : String(e); this.durabilitySink?.( - `[RemoteTransport] could not create the declared unique index — ` + - `${sql}. The constraint is NOT enforced on this table: existing rows already ` + - `violate it (the duplicates this face accepted while it emitted no UNIQUE at all), ` + - `or the index is otherwise unbuildable. Nothing looks broken from the outside and ` + - `duplicates will keep accumulating. Fix by de-duplicating the column's existing ` + - `values and re-running schema sync — this driver deliberately does NOT rewrite ` + - `stored rows to force the index through. Until then a conflictKeys upsert on this ` + - `table is refused rather than crashing. Cause: ${e instanceof Error ? e.message : String(e)}`, + index.unique + ? `[RemoteTransport] could not create the declared unique index "${index.name}" on ` + + `"${index.table}" — ${index.sql}. The constraint is NOT enforced on this table: existing rows ` + + `already violate it (duplicates this face accepted while the index was absent), or the index is ` + + `otherwise unbuildable. Nothing looks broken from the outside and duplicates will keep ` + + `accumulating. Fix by de-duplicating the key's existing values and re-running schema sync — this ` + + `driver deliberately does NOT rewrite stored rows to force the index through. Until then a ` + + `conflictKeys upsert on this table is refused rather than crashing. Cause: ${cause}` + : `[RemoteTransport] could not create the declared index "${index.name}" on "${index.table}" — ` + + `${index.sql}. Every query this index exists to serve is answered by scanning the whole table ` + + `instead: nothing looks broken from the outside and results stay correct, but each such read ` + + `costs a full scan that grows with the table. Fix the cause below and re-run schema sync — the ` + + `next schema sync retries it, and no row is touched either way. Cause: ${cause}`, ); } } diff --git a/packages/drivers/driver-turso/src/turso-driver.ts b/packages/drivers/driver-turso/src/turso-driver.ts index 2ca35697bd..3b7c14c902 100644 --- a/packages/drivers/driver-turso/src/turso-driver.ts +++ b/packages/drivers/driver-turso/src/turso-driver.ts @@ -689,7 +689,9 @@ export class TursoDriver extends SqlDriver { // DURABILITY degradation, not a functional one: writes keep succeeding, // reads keep returning rows, and the only thing that changed is that a // constraint the metadata declares is not enforced — the "looks normal - // from the outside" shape AGENTS.md grades at `error`. It is a SEPARATE + // from the outside" shape AGENTS.md grades at `error`. [#17609] A declared + // PLAIN index it could not create lands here too, for the same reason: + // every query still answers, by scanning the table. It is a SEPARATE // sink from the `warn` one above precisely so this class does not have to // share a level with the diagnostics that are merely informative. this.remoteTransport.setDurabilitySink((message) => diff --git a/packages/drivers/driver-turso/src/turso-local-remote-declared-index-parity.test.ts b/packages/drivers/driver-turso/src/turso-local-remote-declared-index-parity.test.ts new file mode 100644 index 0000000000..fb35c1e827 --- /dev/null +++ b/packages/drivers/driver-turso/src/turso-local-remote-declared-index-parity.test.ts @@ -0,0 +1,509 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#17609] ONE `TursoDriver`, ONE index set — an object's declared + * `indexes: [...]` materialized on BOTH faces and read back off + * `sqlite_master`, the two readings held against each other. + * + * # What was broken + * + * Remote mode (`libsql://`) provisions tables through + * `RemoteTransport.syncSchemasBatch`, and every index DDL that file could emit + * came from field-level `unique` alone. An object's declared `indexes` — + * unique or not — had no consumer on that path, so on every remote tenant + * database the hot tables carried only their primary-key autoindex: + * `sys_notification_delivery` declares five indexes and had none, and its + * claim query full-scanned the table on every dispatcher tick. The local face + * (`SqlDriver.syncDeclaredIndexes`) created all of them, so no local suite + * could see it; the loss surfaced as row-read volume, not as a failure. + * + * # Why a parity file + * + * The same lesson `turso-local-remote-unique-parity.test.ts` records: a + * per-face suite cannot fail on the DIFFERENCE between faces, and the + * difference is the defect. So each pin here drives one object set through + * BOTH faces and compares what physically landed — index names, uniqueness, + * origin and key parts (NULL-safe organization expressions included) — rather + * than asserting either face against a literal. Each parity pin is ANCHORED + * by names computed through the shared `buildIndexName`, because two faces + * that both lost every index would agree perfectly. + * + * The remote face here is the real `@libsql/client` over `file::memory:`, + * not the better-sqlite3 stub: a libsql `write` batch is transactional, and + * the retrofit's failure disposition depends on exactly that semantic. + * + * # The fixtures + * + * `sys_notification_delivery` and `sys_job_queue` are the two tables the card + * measured on production, reproduced here as SHAPES — their fields and their + * `indexes` verbatim — because this package does not (and should not) depend + * on `service-messaging` or `platform-objects`. `os17609_scoped` covers what + * those two do not: a tenant column, a field-level `unique` scoped by it, a + * declared `unique: 'organization'` (NULL-safe key part), an explicit + * `unique: 'global'`, two author-named indexes — one of them no SQL + * identifier at all (a space and a double quote), which both faces must + * accept — and an index over a virtual `formula` field that neither face may + * materialize. + */ + +import { describe, it, expect, afterEach, vi } from 'vitest'; +import { createClient, type Client, type InStatement } from '@libsql/client'; +import { Field } from '@objectstack/spec/data'; +import { buildIndexName } from '@objectstack/driver-sql'; +import { TursoDriver } from './turso-driver.js'; + +type ObjectDef = { + name: string; + fields: Record; + indexes?: Array<{ name?: string; fields: string[]; unique?: boolean | 'global' | 'organization' }>; +}; + +const DELIVERY: ObjectDef = { + name: 'sys_notification_delivery', + fields: { + id: Field.text({ label: 'Delivery ID', required: true, readonly: true }), + notification_id: Field.text({ label: 'Notification Event', required: true, maxLength: 255 }), + recipient_id: Field.text({ label: 'Recipient User', required: true, maxLength: 255 }), + channel: Field.text({ label: 'Channel', required: true, maxLength: 64 }), + topic: Field.text({ label: 'Topic' }), + digest_key: Field.text({ label: 'Digest Key', maxLength: 331 }), + payload: Field.json({ label: 'Payload' }), + status: Field.select(['pending', 'in_flight', 'success', 'failed', 'dead', 'suppressed'], { + label: 'Status', + required: true, + defaultValue: 'pending', + }), + attempts: Field.number({ label: 'Attempts', defaultValue: 0 }), + partition_key: Field.number({ label: 'Partition Key', defaultValue: 0 }), + claimed_by: Field.text({ label: 'Claimed By' }), + claimed_at: Field.number({ label: 'Claimed At (ms)' }), + next_attempt_at: Field.number({ label: 'Next Attempt At (ms)' }), + last_attempted_at: Field.number({ label: 'Last Attempted At (ms)' }), + error: Field.textarea({ label: 'Error' }), + created_at: Field.datetime({ label: 'Created At', readonly: true }), + updated_at: Field.datetime({ label: 'Updated At' }), + }, + indexes: [ + { fields: ['notification_id', 'recipient_id', 'channel'], unique: true }, + { fields: ['status', 'partition_key', 'next_attempt_at'] }, + { fields: ['status', 'claimed_at'] }, + { fields: ['notification_id'] }, + { fields: ['digest_key', 'status', 'next_attempt_at'] }, + ], +}; + +const JOB_QUEUE: ObjectDef = { + name: 'sys_job_queue', + fields: { + id: Field.text({ label: 'Message ID', required: true, readonly: true }), + queue: Field.text({ label: 'Queue', required: true, maxLength: 255 }), + idempotency_key: Field.text({ label: 'Idempotency Key', required: false, maxLength: 255 }), + payload_json: Field.textarea({ label: 'Payload (JSON)', required: false }), + metadata_json: Field.textarea({ label: 'Metadata (JSON)', required: false }), + status: Field.select(['pending', 'running', 'completed', 'failed', 'dlq'], { + label: 'Status', + required: true, + defaultValue: 'pending', + }), + priority: Field.number({ label: 'Priority', required: false, defaultValue: 100 }), + attempts: Field.number({ label: 'Attempts', required: false, defaultValue: 0 }), + max_attempts: Field.number({ label: 'Max Attempts', required: false, defaultValue: 3 }), + backoff_type: Field.select(['fixed', 'exponential'], { + label: 'Backoff', + required: false, + defaultValue: 'exponential', + }), + backoff_delay_ms: Field.number({ label: 'Backoff Base (ms)', required: false, defaultValue: 1000 }), + backoff_max_delay_ms: Field.number({ label: 'Backoff Cap (ms)', required: false }), + scheduled_for: Field.datetime({ label: 'Scheduled For', required: false }), + locked_by: Field.text({ label: 'Locked By', required: false, maxLength: 255 }), + locked_until: Field.datetime({ label: 'Locked Until', required: false }), + last_error: Field.textarea({ label: 'Last Error', required: false }), + completed_at: Field.datetime({ label: 'Completed At', required: false }), + created_at: Field.datetime({ label: 'Created At', required: true, readonly: true }), + updated_at: Field.datetime({ label: 'Updated At', required: false }), + }, + indexes: [ + { fields: ['queue', 'status', 'scheduled_for'] }, + { fields: ['idempotency_key', 'queue'] }, + { fields: ['status'] }, + ], +}; + +const SCOPED: ObjectDef = { + name: 'os17609_scoped', + fields: { + organization_id: { type: 'text', maxLength: 64 }, + code: { type: 'text', maxLength: 64, unique: true }, + slug: { type: 'text', maxLength: 64 }, + external_id: { type: 'text', maxLength: 64 }, + region: { type: 'text', maxLength: 64 }, + total: { type: 'formula', expression: 'region' }, + }, + indexes: [ + { fields: ['slug'], unique: 'organization' }, + { fields: ['external_id'], unique: 'global' }, + { fields: ['region', 'slug'] }, + { name: 'os17609_scoped_by_region', fields: ['region'] }, + { name: 'os17609 scoped-by "slug"', fields: ['slug'] }, + { fields: ['total'] }, + ], +}; + +const OBJECTS = [DELIVERY, JOB_QUEUE, SCOPED]; + +/** The card's claim query, verbatim. */ +const CLAIM_SQL = + `SELECT id FROM sys_notification_delivery WHERE status = 'pending' AND partition_key = ? ` + + `AND next_attempt_at <= ? ORDER BY next_attempt_at LIMIT 50`; + +/** The index names each fixture must carry — through the SHARED namer, never a second spelling. */ +const EXPECTED_NAMES: Record = { + [DELIVERY.name]: [ + buildIndexName(DELIVERY.name, ['notification_id', 'recipient_id', 'channel'], true), + buildIndexName(DELIVERY.name, ['status', 'partition_key', 'next_attempt_at'], false), + buildIndexName(DELIVERY.name, ['status', 'claimed_at'], false), + buildIndexName(DELIVERY.name, ['notification_id'], false), + buildIndexName(DELIVERY.name, ['digest_key', 'status', 'next_attempt_at'], false), + `sqlite_autoindex_${DELIVERY.name}_1`, + ].sort(), + [JOB_QUEUE.name]: [ + buildIndexName(JOB_QUEUE.name, ['queue', 'status', 'scheduled_for'], false), + buildIndexName(JOB_QUEUE.name, ['idempotency_key', 'queue'], false), + buildIndexName(JOB_QUEUE.name, ['status'], false), + `sqlite_autoindex_${JOB_QUEUE.name}_1`, + ].sort(), + [SCOPED.name]: [ + buildIndexName(SCOPED.name, ['organization_id', 'code'], true), + buildIndexName(SCOPED.name, ['organization_id', 'slug'], true), + buildIndexName(SCOPED.name, ['external_id'], true), + buildIndexName(SCOPED.name, ['region', 'slug'], false), + 'os17609_scoped_by_region', + 'os17609 scoped-by "slug"', + `sqlite_autoindex_${SCOPED.name}_1`, + ].sort(), +}; + +const CLAIM_INDEX = buildIndexName(DELIVERY.name, ['status', 'partition_key', 'next_attempt_at'], false); +const DEDUP_INDEX = buildIndexName(DELIVERY.name, ['notification_id', 'recipient_id', 'channel'], true); + +/** A fresh copy per sync — neither face may see a definition the other one mutated. */ +const fresh = (o: ObjectDef): ObjectDef => ({ + ...o, + fields: Object.fromEntries(Object.entries(o.fields).map(([k, v]) => [k, { ...v }])), + ...(o.indexes ? { indexes: o.indexes.map((i) => ({ ...i, fields: [...i.fields] })) } : {}), +}); + +/** The same object as the pre-fix remote face provisioned it: no object-level indexes. */ +const withoutDeclaredIndexes = (o: ObjectDef): ObjectDef => { + const { indexes: _dropped, ...rest } = fresh(o); + return rest; +}; + +type Row = Record; + +/** A SQL identifier, quoted — the fixture carries an index name that is not a bare identifier. */ +const quoteIdent = (id: string) => `"${id.replace(/"/g, '""')}"`; +type Query = (sql: string, args?: unknown[]) => Promise; + +interface IndexShape { + name: string; + unique: boolean; + partial: boolean; + origin: string; + /** Key parts in order; an expression part is its normalized COALESCE text. */ + keys: string[]; +} + +/** + * What physically landed for one table, read the same way off either face: + * `PRAGMA index_list` for name / uniqueness / origin / partiality, + * `PRAGMA index_xinfo` for the key columns, and the stored `CREATE` text for + * expression parts (whose column name `index_xinfo` reports as NULL). + */ +async function indexShapes(query: Query, table: string): Promise { + const out: IndexShape[] = []; + for (const row of await query(`PRAGMA index_list("${table}")`)) { + const name = String(row.name); + const [master] = await query(`SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?`, [name]); + const expressions = [ + ...String(master?.sql ?? '').matchAll(/COALESCE\(\s*[`"]?(\w+)[`"]?\s*,\s*'([^']*)'\s*\)/gi), + ].map((m) => `COALESCE(${m[1]}, '${m[2]}')`); + const keys = (await query(`PRAGMA index_xinfo(${quoteIdent(name)})`)) + .filter((k) => Number(k.key) === 1) + .sort((a, b) => Number(a.seqno) - Number(b.seqno)) + .map((k) => (Number(k.cid) === -2 ? (expressions.shift() ?? '') : String(k.name))); + out.push({ + name, + unique: Number(row.unique) === 1, + partial: Number(row.partial) === 1, + origin: String(row.origin), + keys, + }); + } + // Code-unit order, the same order `EXPECTED_NAMES`' `.sort()` uses. + return out.sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0)); +} + +const indexNames = async (query: Query, table: string) => (await indexShapes(query, table)).map((i) => i.name); + +const cleanups: Array<() => Promise | void> = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function localFace(): Promise<{ driver: TursoDriver; query: Query }> { + const driver = new TursoDriver({ url: ':memory:' }); + await driver.connect(); + expect(driver.transportMode).toBe('local'); + cleanups.push(() => driver.disconnect()); + return { driver, query: async (sql, args) => (await driver.execute(sql, args ?? [])) as Row[] }; +} + +async function remoteFace(client: Client = createClient({ url: 'file::memory:' })): Promise<{ + driver: TursoDriver; + query: Query; +}> { + const driver = new TursoDriver({ url: 'libsql://declared-index-parity.turso.io', client }); + await driver.connect(); + expect(driver.transportMode).toBe('remote'); + cleanups.push(() => driver.disconnect()); + return { + driver, + query: async (sql, args) => (await client.execute({ sql, args: (args ?? []) as any[] })).rows as unknown as Row[], + }; +} + +/** Every statement the driver hands the client, in order, with the call that carried it. */ +function countingClient(inner: Client): { client: Client; calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> } { + const calls: Array<{ via: 'execute' | 'batch'; sql: string[] }> = []; + const sqlOf = (s: InStatement) => (typeof s === 'string' ? s : s.sql); + const client = new Proxy(inner, { + get(target, prop) { + if (prop === 'execute') { + return (stmt: InStatement, ...rest: unknown[]) => { + calls.push({ via: 'execute', sql: [sqlOf(stmt)] }); + return (target.execute as (...a: unknown[]) => unknown).call(target, stmt, ...rest); + }; + } + if (prop === 'batch') { + return (stmts: InStatement[], ...rest: unknown[]) => { + calls.push({ via: 'batch', sql: stmts.map(sqlOf) }); + return (target.batch as (...a: unknown[]) => unknown).call(target, stmts, ...rest); + }; + } + const value = Reflect.get(target, prop, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + return { client, calls }; +} + +const INDEX_DDL = /^\s*CREATE\s+(UNIQUE\s+)?INDEX\b/i; +const INDEX_NAME_READ = /FROM\s+sqlite_master\s+WHERE\s+type\s*=\s*'index'/i; + +/** The durability channel (`logger.error`) and the diagnostic one (`logger.warn`), captured. */ +function captureLogs(driver: TursoDriver) { + const logger = (driver as unknown as { logger: { warn: (m: string) => void; error: (m: string) => void } }).logger; + return { + error: vi.spyOn(logger, 'error').mockImplementation(() => undefined), + warn: vi.spyOn(logger, 'warn').mockImplementation(() => undefined), + }; +} + +describe('[#17609] declared object-level indexes land identically on both TursoDriver faces', () => { + it('a fresh database carries the same index set — names, uniqueness, origin and key parts', async () => { + const local = await localFace(); + const remote = await remoteFace(); + + await local.driver.initObjects(OBJECTS.map(fresh)); + await remote.driver.initObjects(OBJECTS.map(fresh)); + + for (const { name } of OBJECTS) { + const localShapes = await indexShapes(local.query, name); + const remoteShapes = await indexShapes(remote.query, name); + // Names first, for a readable diff; then the whole shape. + expect(remoteShapes.map((i) => i.name), name).toEqual(localShapes.map((i) => i.name)); + expect(remoteShapes, name).toEqual(localShapes); + // The anchor: agreement alone is satisfied by both faces losing them together. + expect(localShapes.map((i) => i.name), name).toEqual(EXPECTED_NAMES[name]); + } + + // The tenant key part is the NULL-safe expression on both faces, not the bare column. + const scoped = await indexShapes(remote.query, SCOPED.name); + expect(scoped.find((i) => i.name === buildIndexName(SCOPED.name, ['organization_id', 'slug'], true))).toEqual( + expect.objectContaining({ unique: true, keys: ["COALESCE(organization_id, '__global__')", 'slug'] }), + ); + }); + + it('the single-object `syncSchema` path lands the same set as the batch path', async () => { + const local = await localFace(); + const remote = await remoteFace(); + + for (const o of OBJECTS) { + await local.driver.initObjects([fresh(o)]); + await remote.driver.syncSchema(o.name, fresh(o)); + } + for (const { name } of OBJECTS) { + expect(await indexShapes(remote.query, name), name).toEqual(await indexShapes(local.query, name)); + } + }); + + it('retrofits every declared index onto tables that already exist — and changes no row', async () => { + const local = await localFace(); + const { client, calls } = countingClient(createClient({ url: 'file::memory:' })); + const remote = await remoteFace(client); + + // The production state: tables the pre-fix remote face provisioned, holding rows. + await remote.driver.initObjects(OBJECTS.map(withoutDeclaredIndexes)); + expect(await indexNames(remote.query, DELIVERY.name)).toEqual([`sqlite_autoindex_${DELIVERY.name}_1`]); + expect(await indexNames(remote.query, JOB_QUEUE.name)).toEqual([`sqlite_autoindex_${JOB_QUEUE.name}_1`]); + for (let i = 0; i < 3; i++) { + await remote.driver.create(DELIVERY.name, { + notification_id: `n${i}`, + recipient_id: 'u1', + channel: 'inbox', + status: 'pending', + partition_key: i, + next_attempt_at: 1000 + i, + }); + await remote.driver.create(JOB_QUEUE.name, { queue: 'q', status: 'pending', idempotency_key: `k${i}` }); + } + const snapshot = async () => ({ + delivery: await remote.query(`SELECT * FROM "${DELIVERY.name}" ORDER BY id`), + jobs: await remote.query(`SELECT * FROM "${JOB_QUEUE.name}" ORDER BY id`), + changes: (await remote.query('SELECT total_changes() AS n'))[0].n, + }); + const before = await snapshot(); + + calls.length = 0; + await remote.driver.initObjects(OBJECTS.map(fresh)); + const retrofitDdl = calls.flatMap((c) => c.sql).filter((s) => INDEX_DDL.test(s)); + + // ⛔ Zero data change: the same rows, byte for byte, and no row-level write at all. + expect(await snapshot()).toEqual(before); + // `IF NOT EXISTS` on every retrofit statement — a concurrent boot creating the same index is a no-op. + expect(retrofitDdl.length).toBeGreaterThan(0); + for (const sql of retrofitDdl) expect(sql).toMatch(/\bINDEX IF NOT EXISTS\b/i); + + await local.driver.initObjects(OBJECTS.map(fresh)); + for (const { name } of OBJECTS) { + expect(await indexShapes(remote.query, name), name).toEqual(await indexShapes(local.query, name)); + expect(await indexNames(remote.query, name), name).toEqual(EXPECTED_NAMES[name]); + } + }); + + it('steady state costs zero index DDL — one index-name read, riding a round trip the sync already pays', async () => { + const { client, calls } = countingClient(createClient({ url: 'file::memory:' })); + const remote = await remoteFace(client); + await remote.driver.initObjects(OBJECTS.map(fresh)); + + // A whole kernel build's `initObjects`, once every index exists. + calls.length = 0; + await remote.driver.initObjects(OBJECTS.map(fresh)); + const boot = calls.flatMap((c) => c.sql); + expect(boot.filter((s) => INDEX_DDL.test(s))).toEqual([]); + + // The DDL seam alone, measured exactly: N table probes (one batch) + N column + // probes and ONE index-name read (one batch) — two round trips, 2N + 1 statements, + // independent of how many indexes the objects declare. + calls.length = 0; + await remote.driver.syncSchemasBatch(OBJECTS.map((o) => ({ object: o.name, schema: fresh(o) }))); + const N = OBJECTS.length; + expect(calls.map((c) => c.via)).toEqual(['batch', 'batch']); + expect(calls.flatMap((c) => c.sql)).toHaveLength(2 * N + 1); + expect(calls[1].sql.filter((s) => INDEX_NAME_READ.test(s))).toHaveLength(1); + expect(calls[1].sql.filter((s) => /^PRAGMA table_info/i.test(s))).toHaveLength(N); + }); + + it('serves the delivery claim query from the declared index — the same plan on both faces', async () => { + const local = await localFace(); + const remote = await remoteFace(); + await local.driver.initObjects([fresh(DELIVERY)]); + await remote.driver.initObjects([fresh(DELIVERY)]); + + const plan = async (query: Query) => + (await query(`EXPLAIN QUERY PLAN ${CLAIM_SQL}`, [0, 0])).map((r) => String(r.detail)).join('\n'); + const remotePlan = await plan(remote.query); + + expect(remotePlan).toContain( + `SEARCH sys_notification_delivery USING INDEX ${CLAIM_INDEX} (status=? AND partition_key=? AND next_attempt_at { + it('an object-level UNIQUE over existing duplicates: not created, named on the durability channel, no row repaired', async () => { + const remote = await remoteFace(); + await remote.driver.initObjects([withoutDeclaredIndexes(DELIVERY)]); + // The duplicates a table with no dedup index accepted. + for (const attempt of [1, 2]) { + await remote.driver.create(DELIVERY.name, { + notification_id: 'n1', + recipient_id: 'u1', + channel: 'email', + status: 'pending', + attempts: attempt, + }); + } + const logs = captureLogs(remote.driver); + + await expect(remote.driver.initObjects([fresh(DELIVERY)])).resolves.toBeUndefined(); + + const names = await indexNames(remote.query, DELIVERY.name); + // The unique index is absent — and the four plain ones DID land, although a libsql + // `write` batch rolls back every statement beside the one that failed. + expect(names).not.toContain(DEDUP_INDEX); + expect(names).toEqual(EXPECTED_NAMES[DELIVERY.name].filter((n) => n !== DEDUP_INDEX)); + + // Reported once, on the `error` channel, naming the index and carrying SQLite's own cause. + const reports = logs.error.mock.calls.map((c) => String(c[0])).filter((m) => m.includes(DEDUP_INDEX)); + expect(reports).toHaveLength(1); + expect(reports[0]).toContain(DELIVERY.name); + expect(reports[0]).toMatch(/UNIQUE constraint failed/i); + expect(logs.warn.mock.calls.some((c) => String(c[0]).includes(DEDUP_INDEX))).toBe(false); + + // ⛔ Nothing repaired: both duplicates are still there. + expect((await remote.query(`SELECT COUNT(*) AS n FROM "${DELIVERY.name}"`))[0].n).toBe(2); + }); + + it('a PLAIN index the server refuses: named on the durability channel with its cause; the rest still land', async () => { + const inner = createClient({ url: 'file::memory:' }); + const refused = (sql: string) => INDEX_DDL.test(sql) && sql.includes(`"${CLAIM_INDEX}"`); + const refusal = () => Object.assign(new Error('SQLITE_FULL: database or disk is full'), { code: 'SQLITE_FULL' }); + const client = new Proxy(inner, { + get(target, prop) { + if (prop === 'execute') { + return async (stmt: InStatement) => { + if (refused(typeof stmt === 'string' ? stmt : stmt.sql)) throw refusal(); + return target.execute(stmt); + }; + } + if (prop === 'batch') { + return async (stmts: InStatement[], mode?: 'write' | 'read' | 'deferred') => { + if (stmts.some((s) => refused(typeof s === 'string' ? s : s.sql))) throw refusal(); + return target.batch(stmts, mode); + }; + } + const value = Reflect.get(target, prop, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + const remote = await remoteFace(client); + await remote.driver.initObjects([withoutDeclaredIndexes(DELIVERY)]); + const logs = captureLogs(remote.driver); + + await expect(remote.driver.initObjects([fresh(DELIVERY)])).resolves.toBeUndefined(); + + const names = await indexNames(remote.query, DELIVERY.name); + expect(names).not.toContain(CLAIM_INDEX); + expect(names).toEqual(EXPECTED_NAMES[DELIVERY.name].filter((n) => n !== CLAIM_INDEX)); + + const reports = logs.error.mock.calls.map((c) => String(c[0])).filter((m) => m.includes(CLAIM_INDEX)); + expect(reports).toHaveLength(1); + expect(reports[0]).toContain(DELIVERY.name); + expect(reports[0]).toContain('SQLITE_FULL'); + }); +});