diff --git a/.changeset/turso-remote-canonical-temporal-backfill.md b/.changeset/turso-remote-canonical-temporal-backfill.md new file mode 100644 index 0000000000..a9f7063d82 --- /dev/null +++ b/.changeset/turso-remote-canonical-temporal-backfill.md @@ -0,0 +1,77 @@ +--- +"@objectstack/driver-turso": minor +--- + +feat(driver-turso): remote 模式补上 canonical 时间列 backfill 通道(分批、可恢复、完成标记) (#5770) + +`SqlDriver.backfillCanonicalDatetimes` / `backfillCanonicalTimes` 是 Knex 路径, +remote 模式的 DDL 与 CRUD 全部走 `@libsql/client`,永远到不了它们。于是 +`canonicalDatetimeFields` / `canonicalTimeFields` 在 remote 恒空, +`needsLegacyDatetimeRepair` 恒为 true。在 `origin/main`(`d82b85fee`)上实测: + +``` +canonicalDatetimeFields['probe'] -> undefined +needsLegacyDatetimeRepair('probe','at') -> true +temporalFilterColumnSql('probe','at','"at"') + -> (case when typeof("at") in ('integer','real') + then strftime('%Y-%m-%dT%H:%M:%fZ', "at"/1000.0, 'unixepoch') + else coalesce(strftime('%Y-%m-%dT%H:%M:%fZ', "at"), "at") end) +``` + +每一次对 `Field.datetime` / `Field.time` 的 filter 都编译成这个表达式 —— 正确,但 +**不可索引**。local 跑完 backfill 能退回 `col >= ?`,remote 此前没有这个出口, +代价是永久的(cloud#1005 后果 A)。 + +**后果 B 实测比原单描述更尖锐。** `RemoteTransport.mapFieldTypeToSQL` 把时间列声明为 +TEXT,#942 之前的 remote 写路径原样透传数字,于是 epoch 毫秒落盘成 +`'1753660800000.0'`。共享修复表达式按 `typeof(col) in ('integer','real')` 分派, +TEXT 亲和列永不命中该支;`strftime` 也解析不了这串数字,`coalesce` 把原值还回来 —— +该行是修复表达式的**不动点**,永远转不过来,并且按 TEXT 比较。原单记为「任何 filter +都匹配不到」;实测是**更坏**的形态 —— `'1753660800000.0'` 字典序排在所有 `'2…'` +之前,所以该行既被自己所属的窗口漏掉,又被并不包含它的窗口命中: + +``` +where at between '2025-07-01T…' and '2025-08-01T…' -> ['ok'] (legacy 行丢失) +where at <= '2030-01-01T…' -> ['legacy','ok'] (不该命中却命中) +``` + +## 本次落地(维护者 2026-08-03 裁定的方案 1) + +新增 `remote-canonical-backfill.ts` 与 +`TursoDriver.backfillRemoteCanonicalTemporal()`,在 remote 的 `initObjects` / +`syncSchema` 之后自动运行 —— 与 local 在 `initObjects` 里调用 backfill 的位置对应: + +- **分批**:每条 `UPDATE` 至多改 `batchSize` 行,借 + `rowid IN (SELECT … LIMIT ?)` 子查询限量(`UPDATE … LIMIT` 需要 libSQL 不保证的 + 编译选项)。 +- **可恢复**:不需要任何断点状态。WHERE 守卫本身就是断点 —— 它选中的正是尚未 + canonical 的行,中断在任何位置都不回滚已转换的行,下次从余量继续;已收敛的列 + 重跑只花一条语句、零写入。 +- **完成标记**:只有在收尾探针测到「两个阶段都无事可做」时才标 canonical,写进 + `canonicalDatetimeFields` / `canonicalTimeFields` —— 与 local 完全相同的消费点 + (`needsLegacyDatetimeRepair`),因此两种 transport 靠同一条规则拿回可索引形态。 + 被批次预算截断或报错的列**不标记**,保留读侧修复。 +- **后果 B 的可解部分**:对元数据声明为 `Field.datetime` / `Field.time` 的列,把 + 纯数字文本按 `cast(col as real)` 喂回**驱动自己的**表达式(typeof 变成 'real', + 于是走它原本的 integer/real 分支)。因此本仓不新增第二套 epoch 转换规则。 +- **不可解残留如实记录**:只解释 1e12 ≤ v < 4102444800000(2001-09-09 ~ 2100-01-01) + 的值。下界是为了让 epoch **秒**永不入界 —— 2100 年前的秒值最大约 4.1e9,若按毫秒 + 解释会把 `'1753660800'`(2025-07-28)静默改写成 1970-01-21,实测确认。界外的行 + 原样留在盘上并计入 `unresolvedEpochTextRows`,不猜。 + +方案 2(DDL 亲和性对齐)按裁定等 staging 存量探针另议,不在本次范围;方案 3(在 +`@objectstack/driver-sql` 公共表达式里加启发式)维护者已否决 —— 上面的恢复限于 +一次性迁移、且只作用于元数据声明为时间类型的列,与那条被否决的读路径启发式不是 +一回事。 + +## 正确性姿态不变(ADR-0053 D-B3 / cloud#1003) + +backfill 是**性能出口,不是正确性前提**。读写路径不依赖它跑过:任何失败(远端不可 +达、标识符非法、预算耗尽)都只导致该列不被标记、读侧继续带修复、答案照旧正确, +且不会让 boot 失败。新增 20 条用例覆盖两个后果、分批/断点续跑/失败中断、完成标记 +的三个门、不可解残留、以及「标记与不标记答案一致」的 D-B3 断言。 + +`turso-remote-temporal-conformance.test.ts` 的两条 legacy sweep 现在显式清除 +canonical 标记(与 driver-sql 的 `LegacyStorageDriver.forgetCanonical` 同一做法), +并断言修复确实仍在生效 —— 此前 remote 「未 backfill」是因为压根没有 backfill 而 +**碰巧**成立,现在它是 fixture 必须自己声明的状态。 diff --git a/packages/drivers/driver-turso/src/index.ts b/packages/drivers/driver-turso/src/index.ts index 878b5cf84d..218d651a5c 100644 --- a/packages/drivers/driver-turso/src/index.ts +++ b/packages/drivers/driver-turso/src/index.ts @@ -37,6 +37,28 @@ import { TursoDriver } from './turso-driver.js'; export { TursoDriver, type TursoDriverConfig, type TursoTransportMode } from './turso-driver.js'; export { RemoteTransport, type FilterColumnSqlResolver } from './remote-transport.js'; +// The remote canonical temporal backfill (#5770 / cloud#1005) — the remote-mode +// exit from the unindexable read-side repair. `TursoDriver` runs it after every +// remote schema sync; these are exported so an operator can drive a large table +// to completion with a bigger budget and read what it actually converged. +export { + backfillRemoteCanonicalColumns, + backfillRemoteCanonicalColumn, + probeRemoteCanonicalColumns, + REMOTE_BACKFILL_DEFAULT_BATCH_SIZE, + REMOTE_BACKFILL_DEFAULT_MAX_BATCHES, + REMOTE_BACKFILL_EPOCH_MS_MIN, + REMOTE_BACKFILL_EPOCH_MS_MAX, + type RemoteBackfillClient, + type RemoteBackfillColumn, + type RemoteBackfillColumnReport, + type RemoteBackfillKind, + type RemoteBackfillLogger, + type CanonicalSqlFor, + type RemoteCanonicalBackfillOptions, + type RemoteCanonicalBackfillReport, +} from './remote-canonical-backfill.js'; + // Spec / Studio metadata for the Turso driver — published from this package // so a host exposes Turso configuration UI without the driver-specific shape // landing in the shared `@objectstack/spec` surface. diff --git a/packages/drivers/driver-turso/src/remote-canonical-backfill.test.ts b/packages/drivers/driver-turso/src/remote-canonical-backfill.test.ts new file mode 100644 index 0000000000..bda705bd10 --- /dev/null +++ b/packages/drivers/driver-turso/src/remote-canonical-backfill.test.ts @@ -0,0 +1,621 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The REMOTE canonical temporal backfill (#5770, cloud#1005 方案 1). + * + * These are ROW-RESULT assertions over a real SQLite database wearing the + * `@libsql/client` interface (`makeLibsqlSqliteStub`), not SQL-string + * assertions. The whole claim under test is about what is ON DISK and what a + * filter therefore matches, and a mis-built UPDATE leaves the SQL perfectly + * valid while converting the wrong rows — the blind spot framework#4081 found + * one layer up. libsql IS SQLite, so the stub gives the transport the real TEXT + * affinity that produced 后果 B in the first place. + * + * ## The measured before-state (verified on origin/main @ d82b85fee) + * + * With a `Field.datetime` column `at` on a remote-mode driver: + * + * | Measured | Value | + * |---|---| + * | `canonicalDatetimeFields['probe']` | `undefined` — nothing ever marks it | + * | `needsLegacyDatetimeRepair('probe','at')` | `true`, forever | + * | `temporalFilterColumnSql('probe','at','"at"')` | the full `case when typeof("at") in ('integer','real') …` CASE — correct, unindexable | + * | a raw `'1753660800000.0'` row (pre-#942 epoch under TEXT affinity) | MISSED by the Jul–Aug 2025 window it belongs to, and MATCHED by `$lte '2030-…'` which it does not | + * + * That last line is worth stating precisely, because it is sharper than + * cloud#1005's summary of "matched by no filter": the row is compared as TEXT, + * so `'1753660800000.0'` sorts before every `'2…'` spelling. It is not + * invisible — it is in the wrong windows, both ways. Same cause, same fix. + */ + +import { describe, it, expect } from 'vitest'; +import { TursoDriver } from './turso-driver.js'; +import { makeLibsqlSqliteStub, type LibsqlSqliteStub } from './libsql-sqlite-stub.testkit.js'; +import { + backfillRemoteCanonicalColumn, + REMOTE_BACKFILL_EPOCH_MS_MAX, + type RemoteBackfillClient, +} from './remote-canonical-backfill.js'; + +const DATETIME_OBJECT = { + name: 'probe', + fields: { at: { type: 'datetime' }, why: { type: 'string' } }, +}; + +const TIME_OBJECT = { + name: 'tprobe', + fields: { at: { type: 'time' }, why: { type: 'string' } }, +}; + +/** 2025-07-28T00:00:00.000Z, the instant cloud#1005 measured on staging. */ +const EPOCH_MS = 1_753_660_800_000; +const EPOCH_ISO = '2025-07-28T00:00:00.000Z'; +/** How TEXT affinity actually stored that bound number, pre-#942. */ +const EPOCH_TEXT = '1753660800000.0'; + +async function makeRemoteDriver(schema: Record, stub?: LibsqlSqliteStub) { + const client = stub ?? makeLibsqlSqliteStub(); + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: client as never }); + await driver.connect(); + expect(driver.transportMode).toBe('remote'); + await driver.syncSchema(schema.name as string, schema); + return { driver, stub: client }; +} + +const rowsOf = (stub: LibsqlSqliteStub, table: string) => + stub.raw.prepare(`select id, at, typeof(at) as t from "${table}" order by id`).all() as Array<{ + id: string; + at: unknown; + t: string; + }>; + +const idsOf = (rows: unknown) => (rows as Array<{ id: string }>).map((r) => r.id).sort(); + +/** Un-mark a column, i.e. put the driver back in the pre-#5770 state. */ +const unmark = (driver: TursoDriver, table: string, field: string, kind = 'datetime') => { + const key = kind === 'datetime' ? 'canonicalDatetimeFields' : 'canonicalTimeFields'; + (driver as never as Record>>)[key][table]?.delete(field); +}; + +const repairSql = (driver: TursoDriver, table: string, field: string) => + ( + driver as never as { + temporalFilterColumnSql: (o: string, f: string, c: string) => string; + } + ).temporalFilterColumnSql(table, field, `"${field}"`); + +const isMarked = (driver: TursoDriver, table: string, field: string, kind = 'datetime') => { + const key = kind === 'datetime' ? 'canonicalDatetimeFields' : 'canonicalTimeFields'; + return ( + (driver as never as Record>>)[key][table]?.has(field) === true + ); +}; + +// ───────────────────────────────────────────────────────────────────────────── +// 后果 A — the unindexable repair now has an exit +// ───────────────────────────────────────────────────────────────────────────── + +describe('#5770 后果 A — remote temporal columns can reach the indexable form', () => { + it('marks a freshly created remote table canonical, so filters emit a plain column', async () => { + const { driver } = await makeRemoteDriver(DATETIME_OBJECT); + + // The whole point: BEFORE this change `needsLegacyDatetimeRepair` was true + // here forever and this returned the CASE expression. + expect(isMarked(driver, 'probe', 'at')).toBe(true); + expect(repairSql(driver, 'probe', 'at')).toBe('"at"'); + + await driver.disconnect(); + }); + + it('reverse-verification: un-marking restores the pre-#5770 unindexable CASE', async () => { + const { driver } = await makeRemoteDriver(DATETIME_OBJECT); + unmark(driver, 'probe', 'at'); + + // This is the measured origin/main behaviour, reproduced by deleting the + // one fact this PR adds. Direction is the ordinary one: remove the fix, + // the diagnostic (the repair wrapper) comes back. + const sql = repairSql(driver, 'probe', 'at'); + expect(sql).toContain(`typeof("at") in ('integer','real')`); + expect(sql).toContain('strftime'); + expect(sql).not.toBe('"at"'); + + await driver.disconnect(); + }); + + it('converges pre-existing legacy rows at the next boot and marks the column', async () => { + // Boot 1 creates the table. + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + + // A pre-convention writer leaves rows the remote table can really hold: + // zone-naive `datetime('now')` output and an offset-bearing ISO string. + stub.raw + .prepare(`insert into probe (id, at, why) values ('naive','2025-07-28 00:00:00','naive')`) + .run(); + stub.raw + .prepare(`insert into probe (id, at, why) values ('offset','2025-07-28T08:00:00+08:00','off')`) + .run(); + stub.raw.prepare(`insert into probe (id, at, why) values ('ok', ?, 'canonical')`).run(EPOCH_ISO); + + // Boot 2 sees the existing table WITH legacy rows. + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + + // Every row is now the one canonical spelling, on disk. + expect(rowsOf(stub, 'probe').map((r) => r.at)).toEqual([EPOCH_ISO, EPOCH_ISO, EPOCH_ISO]); + // …and only because that is TRUE is the repair dropped. + expect(isMarked(driver, 'probe', 'at')).toBe(true); + expect(repairSql(driver, 'probe', 'at')).toBe('"at"'); + + // The filter answers the same thing it did through the repair — the + // performance exit changed the plan, not the result. + const hit = await driver.find('probe', { + where: { at: { $gte: '2025-07-01T00:00:00.000Z', $lte: '2025-08-01T00:00:00.000Z' } }, + }); + expect(idsOf(hit)).toEqual(['naive', 'offset', 'ok']); + + await driver.disconnect(); + stub.close(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// 后果 B — TEXT-affinity numeric epochs +// ───────────────────────────────────────────────────────────────────────────── + +describe('#5770 后果 B — TEXT-affinity numeric epochs are recovered', () => { + it('a raw epoch-text row is in the WRONG windows before, and the right ones after', async () => { + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + + stub.raw.prepare(`insert into probe (id, at, why) values ('legacy', ?, 'epoch')`).run(EPOCH_TEXT); + stub.raw.prepare(`insert into probe (id, at, why) values ('ok', ?, 'canonical')`).run(EPOCH_ISO); + + // BEFORE: a driver that has not run the backfill (the origin/main state). + const before = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await before.connect(); + await before.syncSchema('probe', DATETIME_OBJECT); + unmark(before, 'probe', 'at'); + // Restore the pre-fix disk state the backfill just converged, so the + // "before" half is measured against the real legacy row. + stub.raw.prepare(`update probe set at = ? where id = 'legacy'`).run(EPOCH_TEXT); + + const beforeWindow = await before.find('probe', { + where: { at: { $gte: '2025-07-01T00:00:00.000Z', $lte: '2025-08-01T00:00:00.000Z' } }, + }); + // Missed by the window it belongs to … + expect(idsOf(beforeWindow)).toEqual(['ok']); + const beforeWrong = await before.find('probe', { + where: { at: { $lte: '2030-01-01T00:00:00.000Z' } }, + }); + // … and matched by one it does not, because it is compared as TEXT. + expect(idsOf(beforeWrong)).toEqual(['legacy', 'ok']); + // (not disconnected: `disconnect()` closes the shared stub) + + // AFTER: the backfill runs. + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + + expect(rowsOf(stub, 'probe')).toEqual([ + { id: 'legacy', at: EPOCH_ISO, t: 'text' }, + { id: 'ok', at: EPOCH_ISO, t: 'text' }, + ]); + const after = await driver.find('probe', { + where: { at: { $gte: '2025-07-01T00:00:00.000Z', $lte: '2025-08-01T00:00:00.000Z' } }, + }); + expect(idsOf(after)).toEqual(['legacy', 'ok']); + + await driver.disconnect(); + stub.close(); + }); + + it('recovers an epoch-text value in a Field.time column to a canonical wall clock', async () => { + const { driver: first, stub } = await makeRemoteDriver(TIME_OBJECT); + void first; + stub.raw.prepare(`insert into tprobe (id, at, why) values ('legacy', ?, 'epoch')`).run(EPOCH_TEXT); + // 2025-07-28T00:00:00Z folds to midnight — the same answer reads always gave. + stub.raw.prepare(`insert into tprobe (id, at, why) values ('ok','00:00:00','canonical')`).run(); + + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('tprobe', TIME_OBJECT); + + expect(rowsOf(stub, 'tprobe').map((r) => r.at)).toEqual(['00:00:00', '00:00:00']); + expect(isMarked(driver, 'tprobe', 'at', 'time')).toBe(true); + + await driver.disconnect(); + stub.close(); + }); + + it('leaves out-of-band digits-only text alone, counts it, and still marks the column', async () => { + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + // Below the band (would be read as 1970 if interpreted as milliseconds) and + // above it — both are the unresolvable remainder of 后果 B. + stub.raw.prepare(`insert into probe (id, at, why) values ('small','12','tiny')`).run(); + stub.raw + .prepare(`insert into probe (id, at, why) values ('huge', ?, 'far-future')`) + .run(String(REMOTE_BACKFILL_EPOCH_MS_MAX + 1)); + stub.raw.prepare(`insert into probe (id, at, why) values ('ok', ?, 'canonical')`).run(EPOCH_ISO); + + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + // `syncSchema` already ran the pass and marked the column, so the driver + // method would skip it (that is the cheap-re-entry rule). Re-probe the + // converged column directly to read what it left behind — which also + // exercises the fast path's reporting. + const report = { + columns: [ + await backfillRemoteCanonicalColumn( + instrument(stub), + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + ), + ], + }; + expect(report.columns[0].canonical).toBe(true); + + const disk = rowsOf(stub, 'probe'); + + // `'huge'` is out of band for the epoch recovery AND a fixpoint of the + // shared repair (`strftime` declines it), so nothing touches it. Untouched + // on disk is the honest outcome: the backfill does not guess. + expect(disk.find((r) => r.id === 'huge')!.at).toBe(String(REMOTE_BACKFILL_EPOCH_MS_MAX + 1)); + + // `'small'` is a different animal and worth being exact about. It is out of + // band for the epoch limb too — but it is NOT a fixpoint of the shared + // repair, because SQLite reads a bare small number as a JULIAN DAY. So the + // shared convergence rewrites it, and the value it writes is precisely what + // a repaired READ already returned for that row. That is the invariant this + // whole module rests on: the backfill materialises the repair, it never + // changes an answer. + const asRepairRead = ( + stub.raw + .prepare(`select coalesce(strftime('%Y-%m-%dT%H:%M:%fZ','12'),'12') as v`) + .all() as Array<{ v: string }> + )[0].v; + expect(asRepairRead).toBe('-4713-12-06T12:00:00.000Z'); + expect(disk.find((r) => r.id === 'small')!.at).toBe(asRepairRead); + + // Only the still-digits-only row is reported as the unresolvable remainder. + const col = report.columns.find((c) => c.field === 'at'); + expect(col?.unresolvedEpochTextRows).toBe(1); + expect(col?.pendingEpochTextRows).toBe(0); + + // Still marked: these rows are fixpoints of the shared repair, so reading + // them with it and without it give the identical answer — dropping the + // repair cannot change which rows they match. + expect(isMarked(driver, 'probe', 'at')).toBe(true); + + await driver.disconnect(); + stub.close(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// Batching, resumability, and the completion marker +// ───────────────────────────────────────────────────────────────────────────── + +/** Wrap a stub to count statements and (optionally) fail on the Nth write. */ +function instrument(stub: LibsqlSqliteStub, failWriteAt?: number) { + const executed: string[] = []; + let writes = 0; + const client: RemoteBackfillClient & { executed: string[] } = { + executed, + async execute(stmt) { + const sql = typeof stmt === 'string' ? stmt : stmt.sql; + executed.push(sql); + if (/^\s*update/i.test(sql)) { + writes++; + if (failWriteAt !== undefined && writes === failWriteAt) { + throw new Error('simulated remote failure mid-backfill'); + } + } + return stub.execute(stmt) as never; + }, + async batch(stmts) { + // The stub takes no mode argument — it is a local SQLite database, and + // libsql's read/write batch mode is a wire-protocol concern. + for (const s of stmts) executed.push(typeof s === 'string' ? s : s.sql); + return stub.batch(stmts) as never; + }, + }; + return client; +} + +const canonicalFor = (driver: TursoDriver) => (kind: 'datetime' | 'time', columnSql: string) => + kind === 'datetime' + ? (driver as never as { sqliteCanonicalDatetimeSql: (c: string) => string }) + .sqliteCanonicalDatetimeSql(columnSql) + : (driver as never as { sqliteCanonicalTimeSql: (c: string) => string }) + .sqliteCanonicalTimeSql(columnSql); + +async function seedLegacy(rows: number) { + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + const insert = stub.raw.prepare(`insert into probe (id, at, why) values (?, ?, 'legacy')`); + for (let i = 0; i < rows; i++) insert.run(`r${String(i).padStart(4, '0')}`, EPOCH_TEXT); + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + return { driver, stub }; +} + +describe('#5770 batching / resumability / completion marker', () => { + it('splits a large table across several UPDATEs and converges all of it', async () => { + const { driver, stub } = await seedLegacy(25); + const client = instrument(stub); + + const report = await backfillRemoteCanonicalColumn( + client, + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + { batchSize: 10, maxBatches: 50 }, + ); + + expect(report.epochTextRowsConverted).toBe(25); + expect(report.residualRows).toBe(0); + expect(report.canonical).toBe(true); + expect(report.budgetExhausted).toBe(false); + // 25 rows at 10 per statement cannot have been one UPDATE. + const updates = client.executed.filter((s) => /^\s*update/i.test(s)); + expect(updates.length).toBeGreaterThanOrEqual(3); + // Every statement carries the LIMIT that makes it a batch. + for (const u of updates) expect(u).toContain('limit ?'); + + expect(rowsOf(stub, 'probe').every((r) => r.at === EPOCH_ISO)).toBe(true); + await driver.disconnect(); + stub.close(); + }); + + it('a budget-stopped run withholds the mark, and a re-run finishes and marks it', async () => { + const { driver, stub } = await seedLegacy(25); + const client = instrument(stub); + const column = { table: 'probe', field: 'at', kind: 'datetime' as const }; + + const partial = await backfillRemoteCanonicalColumn(client, column, canonicalFor(driver), { + batchSize: 10, + maxBatches: 1, + }); + expect(partial.budgetExhausted).toBe(true); + expect(partial.epochTextRowsConverted).toBe(10); + // NOTE the count that is zero here, because it is the trap this design had + // to be corrected for: an un-converted TEXT epoch is a FIXPOINT of the + // shared repair, so `residualRows` reads 0 even with 15 rows still on the + // legacy form. Gating the mark on `residual` alone would have declared the + // column done and — since a marked column is skipped next run — stranded + // those 15 rows permanently. + expect(partial.residualRows).toBe(0); + expect(partial.pendingEpochTextRows).toBe(15); + // The load-bearing half: NOT marked, so the caller keeps the repair and the + // 15 rows still on the legacy form keep answering correctly. + expect(partial.canonical).toBe(false); + + const finish = await backfillRemoteCanonicalColumn(client, column, canonicalFor(driver), { + batchSize: 10, + maxBatches: 50, + }); + // Resumed exactly where it stopped — no checkpoint state, the WHERE guard IS + // the checkpoint. + expect(finish.epochTextRowsConverted).toBe(15); + expect(finish.residualRows).toBe(0); + expect(finish.canonical).toBe(true); + expect(rowsOf(stub, 'probe').every((r) => r.at === EPOCH_ISO)).toBe(true); + + await driver.disconnect(); + stub.close(); + }); + + it('a mid-run failure converts nothing further, reports, and never marks', async () => { + const { driver, stub } = await seedLegacy(25); + const client = instrument(stub, 2); // blow up on the second UPDATE + + const report = await backfillRemoteCanonicalColumn( + client, + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + { batchSize: 10, maxBatches: 50 }, + ); + + expect(report.error).toMatch(/simulated remote failure/); + expect(report.canonical).toBe(false); + // The first batch's work survives — that is what makes the retry cheap. + const converged = rowsOf(stub, 'probe').filter((r) => r.at === EPOCH_ISO); + expect(converged).toHaveLength(10); + + await driver.disconnect(); + stub.close(); + }); + + it('is idempotent: a converged column costs one statement and zero writes', async () => { + const { driver, stub } = await makeRemoteDriver(DATETIME_OBJECT); + stub.raw.prepare(`insert into probe (id, at, why) values ('ok', ?, 'canonical')`).run(EPOCH_ISO); + const client = instrument(stub); + + const report = await backfillRemoteCanonicalColumn( + client, + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + ); + + expect(report.canonical).toBe(true); + expect(report.rowsConverted).toBe(0); + expect(report.epochTextRowsConverted).toBe(0); + expect(client.executed).toHaveLength(1); + expect(client.executed[0]).toMatch(/^select /); + + await driver.disconnect(); + stub.close(); + }); + + it('probes every column of a boot in ONE round-trip', async () => { + const stub = makeLibsqlSqliteStub(); + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.initObjects([ + { name: 'a', fields: { at: { type: 'datetime' }, tod: { type: 'time' } } }, + { name: 'b', fields: { at: { type: 'datetime' } } }, + ]); + + // All three temporal columns marked, and a re-run finds nothing to do. + expect(isMarked(driver, 'a', 'at')).toBe(true); + expect(isMarked(driver, 'a', 'tod', 'time')).toBe(true); + expect(isMarked(driver, 'b', 'at')).toBe(true); + expect(await driver.backfillRemoteCanonicalTemporal()).toEqual({ columns: [] }); + + await driver.disconnect(); + stub.close(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// D-B3 — correctness never depends on this having run +// ───────────────────────────────────────────────────────────────────────────── + +describe('#5770 D-B3 — the backfill is a performance exit, not a correctness prerequisite', () => { + it('answers identically with the column marked and un-marked', async () => { + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + stub.raw + .prepare(`insert into probe (id, at, why) values ('naive','2025-07-28 00:00:00','naive')`) + .run(); + stub.raw + .prepare(`insert into probe (id, at, why) values ('offset','2025-07-28T08:00:00+08:00','off')`) + .run(); + + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + + const window = { at: { $gte: '2025-07-01T00:00:00.000Z', $lte: '2025-08-01T00:00:00.000Z' } }; + const marked = idsOf(await driver.find('probe', { where: window })); + + unmark(driver, 'probe', 'at'); + const unmarked = idsOf(await driver.find('probe', { where: window })); + + expect(marked).toEqual(['naive', 'offset']); + expect(unmarked).toEqual(marked); + + await driver.disconnect(); + stub.close(); + }); + + it('an unreachable remote leaves the driver usable and every column un-marked', async () => { + const stub = makeLibsqlSqliteStub(); + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + unmark(driver, 'probe', 'at'); + + // Every statement the backfill issues fails. + const dead: RemoteBackfillClient = { + async execute() { + throw new Error('remote unreachable'); + }, + async batch() { + throw new Error('remote unreachable'); + }, + }; + const report = await backfillRemoteCanonicalColumn( + dead, + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + ); + + expect(report.error).toMatch(/remote unreachable/); + expect(report.canonical).toBe(false); + // The repair is still there, so reads are still correct. + expect(repairSql(driver, 'probe', 'at')).toContain('strftime'); + + await driver.disconnect(); + stub.close(); + }); + + it('never marks a column whose rows it could not converge', async () => { + const { driver: first, stub } = await makeRemoteDriver(DATETIME_OBJECT); + void first; + for (let i = 0; i < 5; i++) { + stub.raw + .prepare(`insert into probe (id, at, why) values (?, '2025-07-28 00:00:00','naive')`) + .run(`r${i}`); + } + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + await driver.syncSchema('probe', DATETIME_OBJECT); + unmark(driver, 'probe', 'at'); + // Undo the automatic pass so there is genuinely work left. + stub.raw.prepare(`update probe set at = '2025-07-28 00:00:00'`).run(); + + const report = await backfillRemoteCanonicalColumn( + instrument(stub), + { table: 'probe', field: 'at', kind: 'datetime' }, + canonicalFor(driver), + { batchSize: 2, maxBatches: 1 }, + ); + expect(report.canonical).toBe(false); + expect(report.residualRows).toBe(3); + + await driver.disconnect(); + stub.close(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// Identifier safety — identifiers are inlined, so they are checked +// ───────────────────────────────────────────────────────────────────────────── + +describe('#5770 identifier safety', () => { + it.each([ + ['probe"; drop table probe; --', 'at'], + ['probe', 'at"; drop table probe; --'], + ['1probe', 'at'], + ])('refuses an unsafe identifier (%s.%s)', async (table, field) => { + const stub = makeLibsqlSqliteStub(); + const driver = new TursoDriver({ url: 'libsql://probe.turso.io', client: stub as never }); + await driver.connect(); + + const report = await backfillRemoteCanonicalColumn( + instrument(stub), + { table, field, kind: 'datetime' }, + canonicalFor(driver), + ); + // Reported, not thrown — the caller treats it as "leave the repair on". + expect(report.error).toMatch(/unsafe identifier rejected/); + expect(report.canonical).toBe(false); + + await driver.disconnect(); + stub.close(); + }); +}); + +// ───────────────────────────────────────────────────────────────────────────── +// The non-remote guard +// ───────────────────────────────────────────────────────────────────────────── + +describe('#5770 local mode is untouched', () => { + it('backfillRemoteCanonicalTemporal is a no-op outside remote mode', async () => { + const driver = new TursoDriver({ url: ':memory:' }); + await driver.connect(); + expect(driver.transportMode).toBe('local'); + // Local reaches the same state through the INHERITED Knex backfill; this + // method must not double up on it. + expect(await driver.backfillRemoteCanonicalTemporal()).toEqual({ columns: [] }); + await driver.disconnect(); + }); + + it('local mode still marks its own columns via the inherited Knex backfill', async () => { + const driver = new TursoDriver({ url: ':memory:' }); + await driver.connect(); + await driver.initObjects([DATETIME_OBJECT]); + // Proves the two transports converge on the same consumption point. + expect(isMarked(driver, 'probe', 'at')).toBe(true); + await driver.disconnect(); + }); +}); + +it('EPOCH_MS constants describe the band the recovery documents', () => { + expect(new Date(EPOCH_MS).toISOString()).toBe(EPOCH_ISO); + expect(Number(EPOCH_TEXT)).toBe(EPOCH_MS); +}); diff --git a/packages/drivers/driver-turso/src/remote-canonical-backfill.ts b/packages/drivers/driver-turso/src/remote-canonical-backfill.ts new file mode 100644 index 0000000000..10f8f0a8ef --- /dev/null +++ b/packages/drivers/driver-turso/src/remote-canonical-backfill.ts @@ -0,0 +1,561 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The REMOTE-mode canonical temporal backfill (objectstack#5770, cloud#1005). + * + * `SqlDriver.backfillCanonicalDatetimes` / `backfillCanonicalTimes` converge a + * SQLite table's `Field.datetime` / `Field.time` columns on the canonical + * storage form (#3912 / #3994) and then mark the column clean, which is what + * lets `needsLegacyDatetimeRepair` drop the read-side repair expression and + * emit a plain, INDEXABLE `col >= ?` again. Both are Knex paths. TursoDriver's + * remote mode never touches Knex — its DDL and CRUD go out over + * `@libsql/client` (see `RemoteTransport`) — so in remote mode the backfill has + * never run, `canonicalDatetimeFields` / `canonicalTimeFields` stay empty, and + * every temporal filter compiles to the unindexable repair CASE forever. Local + * mode has an exit from that cost; remote had none. This module is that exit. + * + * ## What it is NOT + * + * It is NOT a correctness prerequisite. ADR-0053 D-B3 / cloud#1003 fix the + * posture: reads and writes are correct whether or not a backfill has ever run, + * because the read-side repair is always there until a column is PROVED + * canonical. Everything here is a performance exit, and every failure mode + * below degrades to "column stays unmarked, reads keep the repair, answers stay + * correct" — never to a wrong answer and never to a failed boot. + * + * ## Batched, resumable, and marked only when proved (the maintainer's 方案 1) + * + * - **Batched.** Each phase converts at most `batchSize` rows per statement, + * selected by `rowid` (`RemoteTransport.buildCreateTableSQL` never emits + * `WITHOUT ROWID`, so every table it makes has one). `UPDATE ... LIMIT` + * needs a compile-time option libSQL does not promise, so the limit rides on + * a `rowid IN (SELECT ... LIMIT n)` subquery, which is portable SQLite. + * - **Resumable.** No checkpoint state exists or is needed: the WHERE guard + * selects exactly the rows that are not yet canonical, so a run interrupted + * anywhere leaves converted rows converted and the next run picks up the + * remainder. Re-running a converged column costs one statement and zero + * writes. + * - **Marked only when proved.** The column is reported `canonical` only when a + * post-pass probe measures nothing left for either phase to change. A run + * stopped by its batch budget, or by an error, reports `canonical: false` and + * the caller leaves the repair in place. This is stricter than the local twin, + * which marks clean on a successful `UPDATE` because that statement is + * unbatched and therefore total. + * + * "Nothing left to change" deliberately takes TWO counts, not one. An + * un-converted TEXT epoch is a FIXPOINT of the shared repair (see 后果 B + * below), so a table of nothing but legacy epoch rows measures zero residual + * and would look finished. Gating on that alone would mark the column, drop + * the repair, and — because a marked column is skipped on the next run — + * strand exactly the rows this module exists to rescue. The in-band epoch + * count is therefore part of the gate. + * + * ## The 后果 B limb — TEXT-affinity numeric epochs, handled HERE and only here + * + * `RemoteTransport.mapFieldTypeToSQL` declares every temporal column `TEXT`, so + * the pre-#942 remote write path (which passed a bound number through verbatim) + * left epoch milliseconds on disk as TEXT — measured as `'1753660800000.0'`. + * The shared repair dispatches on `typeof(col) in ('integer','real')`, a branch + * a TEXT-affinity column can never take, and `strftime` cannot parse that + * string either, so `coalesce` hands the original back: the row is a fixpoint + * of the repair, is never converged, and is compared as TEXT. It is not + * invisible — it is worse: `'1753660800000.0'` sorts before `'2025-…'`, so the + * row is missed by the window it belongs to AND matched by windows it does not. + * + * The maintainer's 2026-08-03 ruling on cloud#1005 REJECTED teaching the shared + * `@objectstack/driver-sql` expression to recognise epoch-looking numeric text + * (option 3): that expression is a public contract for every SQLite consumer + * and runs on every read, where the heuristic would misread a legitimate + * numeric-string column. The same recovery is safe here, and the difference is + * not a matter of degree: + * + * - it runs ONCE, as an explicit migration, not on every read; + * - it only ever touches a column the METADATA declares `Field.datetime` or + * `Field.time` — the shape is not guessed from the value; + * - a digits-only string is not a spelling of any canonical temporal value, + * so in such a column it is unambiguously a pre-#942 write; + * - it is bounded to a plausible epoch-millisecond band and re-checked to + * have produced TEXT before a single row is written. + * + * Rows outside that band are LEFT ALONE and counted into + * `unresolvedEpochTextRows` rather than guessed at — the "不可解残留如实记录" + * half of the ruling. They do not block the canonical mark, because they are + * fixpoints of the shared repair: reading them through it and reading them + * without it give the identical answer, so dropping the repair cannot change + * what they match. + * + * ## One definition of "canonical" + * + * This module never spells the repair expression. The caller passes the + * driver's own `sqliteCanonicalDatetimeSql` / `sqliteCanonicalTimeSql` in, so + * the SET expression, the WHERE guard, the convergence probe and the read path + * are the same rule by construction and cannot drift — the property the local + * twin's contract calls out, kept across the transport boundary. + */ + +/** The `@libsql/client` surface this module uses — nothing more. */ +export interface RemoteBackfillClient { + execute(stmt: { sql: string; args?: unknown[] } | string): Promise<{ + rows: unknown[]; + rowsAffected: number; + }>; + batch( + stmts: Array<{ sql: string; args?: unknown[] } | string>, + mode?: string, + ): Promise>; +} + +/** Minimal log sink — the driver hands its own in. */ +export interface RemoteBackfillLogger { + warn: (msg: string, meta?: unknown) => void; + info?: (msg: string, meta?: unknown) => void; +} + +/** Which canonical form a column is being converged on. */ +export type RemoteBackfillKind = 'datetime' | 'time'; + +/** A column to converge, and the driver's rule for reading it canonically. */ +export interface RemoteBackfillColumn { + table: string; + field: string; + kind: RemoteBackfillKind; +} + +/** + * Renders the driver's canonical-read expression around an arbitrary SQL + * column reference — `SqlDriver.sqliteCanonicalDatetimeSql` / + * `sqliteCanonicalTimeSql`, handed in rather than copied. + */ +export type CanonicalSqlFor = (kind: RemoteBackfillKind, columnSql: string) => string; + +export interface RemoteCanonicalBackfillOptions { + /** + * Rows rewritten per `UPDATE` statement. Default + * {@link REMOTE_BACKFILL_DEFAULT_BATCH_SIZE}. + */ + batchSize?: number; + /** + * Maximum `UPDATE` statements per phase per column in ONE invocation — + * the budget that keeps an automatic run from turning a huge legacy table + * into an unbounded boot. Default {@link REMOTE_BACKFILL_DEFAULT_MAX_BATCHES}. + * Exhausting it is not a failure: the column stays unmarked, the reads stay + * repaired, and the next invocation resumes exactly where this one stopped. + */ + maxBatches?: number; +} + +/** What one column's convergence attempt actually did. */ +export interface RemoteBackfillColumnReport extends RemoteBackfillColumn { + /** Rows rewritten by the TEXT-affinity epoch recovery (后果 B). */ + epochTextRowsConverted: number; + /** Rows rewritten by the shared canonical convergence (后果 A). */ + rowsConverted: number; + /** Rows still not a fixpoint of the shared repair when this run stopped. */ + residualRows: number; + /** + * In-band epoch-text rows still awaiting conversion when this run stopped — + * nonzero only when a batch budget or an error cut the run short. Blocks the + * canonical mark: these rows ARE convertible, and a marked column is skipped + * next time. + */ + pendingEpochTextRows: number; + /** + * Digits-only TEXT rows this run deliberately declined to interpret because + * they fall outside {@link REMOTE_BACKFILL_EPOCH_MS_MIN} … + * {@link REMOTE_BACKFILL_EPOCH_MS_MAX} — the unresolvable remainder of 后果 B, + * recorded rather than guessed at. + * + * Does NOT block the canonical mark: such a row is a fixpoint of the shared + * repair, so reading it with the repair and without it give the identical + * answer, and dropping the repair cannot change which rows it matches. + */ + unresolvedEpochTextRows: number; + /** + * The column is PROVED converged: no row is left for either phase to change, + * so the caller may drop the read-side repair. `false` whenever that was not + * measured, for any reason at all. + */ + canonical: boolean; + /** A batch budget stopped this run before convergence. */ + budgetExhausted: boolean; + /** Present when the column was skipped because a statement failed. */ + error?: string; +} + +export interface RemoteCanonicalBackfillReport { + columns: RemoteBackfillColumnReport[]; +} + +/** Rows rewritten per `UPDATE`. */ +export const REMOTE_BACKFILL_DEFAULT_BATCH_SIZE = 500; + +/** `UPDATE` statements per phase per column, per invocation. */ +export const REMOTE_BACKFILL_DEFAULT_MAX_BATCHES = 20; + +/** + * The band of digits-only values the 后果 B recovery will interpret as epoch + * MILLISECONDS, inclusive lower / exclusive upper: 2001-09-09 through + * 2100-01-01. + * + * The floor is chosen for one reason and it is not cosmetic. A digits-only + * string is only *unambiguously* milliseconds once it is too large to be a + * plausible epoch-SECONDS value: seconds for any instant before 2100 are at + * most ~4.1e9, so a floor of 1e12 puts every one of them out of band. Without + * it, `'1753660800'` (2025-07-28 in seconds) would be read as milliseconds and + * silently rewritten to 1970-01-21 — measured, and exactly the kind of + * irreversible mis-repair the maintainer refused to put in the shared + * expression. + * + * Values outside the band are LEFT ON DISK and counted, never guessed at. + */ +export const REMOTE_BACKFILL_EPOCH_MS_MIN = 1_000_000_000_000; +export const REMOTE_BACKFILL_EPOCH_MS_MAX = 4_102_444_800_000; + +/** + * Identifiers are INLINED into these statements (SQLite cannot bind an + * identifier), so every one is checked against the same allowlist + * `RemoteTransport` uses for its DDL before it reaches a string template. + */ +const SAFE_IDENTIFIER = /^[a-zA-Z_][a-zA-Z0-9_]*$/; + +function assertSafeIdentifier(name: string): void { + if (!SAFE_IDENTIFIER.test(name)) { + throw new Error(`remote canonical backfill: unsafe identifier rejected: "${name}"`); + } +} + +/** + * "This TEXT value is a digits-only string" — the SHAPE half of the 后果 B + * guard, with no opinion yet on whether it is interpretable. + * + * `GLOB '[0-9]*'` requires a leading digit and `NOT GLOB '*[^0-9.]*'` forbids + * every character but digits and the decimal point, so no canonical temporal + * spelling can match: an ISO instant carries `-`, `:`, `T` and `Z`, and a + * canonical wall clock carries `:`. + */ +function epochTextShapeSql(columnSql: string): string { + return ( + `${columnSql} is not null and typeof(${columnSql}) = 'text' ` + + `and ${columnSql} glob '[0-9]*' and ${columnSql} not glob '*[^0-9.]*'` + ); +} + +/** {@link epochTextShapeSql} plus the plausible-epoch-millisecond band. */ +function epochTextInBandSql(columnSql: string): string { + return ( + `${epochTextShapeSql(columnSql)} ` + + `and cast(${columnSql} as real) >= ${REMOTE_BACKFILL_EPOCH_MS_MIN} ` + + `and cast(${columnSql} as real) < ${REMOTE_BACKFILL_EPOCH_MS_MAX}` + ); +} + +/** + * "This row is not yet canonical" — the same null-safe, type-aware `IS NOT` + * guard the local twin uses as its whole WHERE, over the driver's own + * expression. + */ +function notCanonicalSql(columnSql: string, canonicalSql: string): string { + return `${columnSql} is not null and ${columnSql} is not ${canonicalSql}`; +} + +const readCount = (result: { rows: unknown[] }, key: string): number => { + const row = result.rows[0] as Record | undefined; + const raw = row?.[key]; + return typeof raw === 'number' ? raw : Number(raw ?? 0) || 0; +}; + +/** + * One statement that measures everything a decision needs. + * + * THREE counts, and the third is the one that makes the completion marker + * honest. `residual` alone cannot decide it, because a TEXT epoch is a FIXPOINT + * of the shared repair — `strftime` declines it and `coalesce` hands it back — + * so a table of nothing but un-converted 后果 B rows measures `residual = 0`. + * Marking on that would declare the column done, drop the repair, and (because + * a marked column is skipped on the next run) strand those rows on the legacy + * form permanently. `epoch_in_band` is what remains to be converted; the mark + * waits for it. + * + * This is also the whole cost of re-running against a converged table — the + * counts come back zero and nothing is written. + */ +function buildProbeSql(column: RemoteBackfillColumn, canonicalSqlFor: CanonicalSqlFor): string { + assertSafeIdentifier(column.table); + assertSafeIdentifier(column.field); + const col = `"${column.field}"`; + const canonical = canonicalSqlFor(column.kind, col); + return ( + `select ` + + `sum(case when ${notCanonicalSql(col, canonical)} then 1 else 0 end) as residual, ` + + `sum(case when ${epochTextShapeSql(col)} then 1 else 0 end) as epoch_text, ` + + `sum(case when ${epochTextInBandSql(col)} then 1 else 0 end) as epoch_in_band ` + + `from "${column.table}"` + ); +} + +/** What one probe statement measured. */ +interface RemoteBackfillProbe { + /** Rows that are not a fixpoint of the shared repair. */ + residual: number; + /** Rows carrying the digits-only TEXT shape, in band or not. */ + epochText: number; + /** Rows the 后果 B recovery will interpret — the work still outstanding. */ + epochInBand: number; +} + +const readProbe = (result: { rows: unknown[] }): RemoteBackfillProbe => ({ + residual: readCount(result, 'residual'), + epochText: readCount(result, 'epoch_text'), + epochInBand: readCount(result, 'epoch_in_band'), +}); + +/** + * Probe every column in ONE round-trip. + * + * Boot calls this for all of a driver's temporal columns at once; on a database + * that is already converged (the steady state) that single batch is the entire + * cost of the backfill, and nothing else in this module runs. + */ +export async function probeRemoteCanonicalColumns( + client: RemoteBackfillClient, + columns: RemoteBackfillColumn[], + canonicalSqlFor: CanonicalSqlFor, +): Promise> { + if (columns.length === 0) return []; + let stmts: string[]; + try { + stmts = columns.map((c) => buildProbeSql(c, canonicalSqlFor)); + } catch (err) { + // An unsafe identifier anywhere in the set — report per column and let + // `backfillRemoteCanonicalColumn` re-raise it for the offending one. + const message = err instanceof Error ? err.message : String(err); + return columns.map(() => ({ error: message })); + } + try { + const results = await client.batch(stmts, 'read'); + return results.map(readProbe); + } catch { + // A batch is all-or-nothing; fall back to per-column probes so one + // unreadable table (dropped, permission-denied) cannot mask the rest. + const out: Array = []; + for (let i = 0; i < columns.length; i++) { + try { + out.push(readProbe(await client.execute(stmts[i]))); + } catch (inner) { + out.push({ error: inner instanceof Error ? inner.message : String(inner) }); + } + } + return out; + } +} + +/** + * Run one batched `UPDATE` phase until it converges, the budget runs out, or a + * statement stops making progress. + * + * Returns the rows rewritten and whether the budget was the reason it stopped. + * The no-progress break is a safety valve, not an expected path: each batch + * turns the rows it selects into fixpoints of the guard, so the matching set + * strictly shrinks and the loop terminates on its own. + */ +async function runBatchedUpdate( + client: RemoteBackfillClient, + updateSql: string, + batchSize: number, + maxBatches: number, +): Promise<{ rowsConverted: number; budgetExhausted: boolean }> { + let rowsConverted = 0; + for (let batch = 0; batch < maxBatches; batch++) { + const res = await client.execute({ sql: updateSql, args: [batchSize] }); + const affected = res.rowsAffected ?? 0; + rowsConverted += affected; + // Short of the limit means the guard had nothing more to match, so the + // phase is done — and a zero-row batch is the same statement, said once. + if (affected < batchSize) return { rowsConverted, budgetExhausted: false }; + } + return { rowsConverted, budgetExhausted: true }; +} + +/** + * Converge ONE column and report what happened — the unit every caller + * ultimately goes through. + * + * Phase order is load-bearing: the 后果 B recovery runs FIRST so the rows it + * rewrites are already canonical text by the time the shared convergence sweeps, + * which then finds them fixpoints and does not touch them again. + */ +export async function backfillRemoteCanonicalColumn( + client: RemoteBackfillClient, + column: RemoteBackfillColumn, + canonicalSqlFor: CanonicalSqlFor, + options: RemoteCanonicalBackfillOptions = {}, + probed?: RemoteBackfillProbe, +): Promise { + const batchSize = Math.max(1, options.batchSize ?? REMOTE_BACKFILL_DEFAULT_BATCH_SIZE); + const maxBatches = Math.max(1, options.maxBatches ?? REMOTE_BACKFILL_DEFAULT_MAX_BATCHES); + + const base: RemoteBackfillColumnReport = { + ...column, + epochTextRowsConverted: 0, + rowsConverted: 0, + residualRows: 0, + pendingEpochTextRows: 0, + unresolvedEpochTextRows: 0, + canonical: false, + budgetExhausted: false, + }; + + try { + // Inside the `try` on purpose: a rejected identifier is REPORTED like every + // other failure, so the caller's response is the same one it has for an + // unreachable remote — leave the read-side repair on — rather than an + // exception that could take a boot down (D-B3). + assertSafeIdentifier(column.table); + assertSafeIdentifier(column.field); + const table = `"${column.table}"`; + const col = `"${column.field}"`; + const canonical = canonicalSqlFor(column.kind, col); + // Reading the TEXT epoch AS A NUMBER is the whole trick: `cast(col as real)` + // makes `typeof()` report 'real', so the driver's OWN expression takes its + // integer/real branch and produces the canonical form. The recovery therefore + // reuses the shared rule instead of restating the epoch conversion. + const canonicalFromEpochText = canonicalSqlFor(column.kind, `cast(${col} as real)`); + + const probe = + probed ?? readProbe(await client.execute(buildProbeSql(column, canonicalSqlFor))); + + // Fast path — and the steady state. A converged column, and a table this + // very boot created (its counts are NULL, hence zero), costs one statement + // and no writes, and is marked canonical on measured evidence rather than + // on the assumption that it must be empty. Out-of-band digits-only rows are + // reported but do not hold the mark back — see the field's contract. + if (probe.residual === 0 && probe.epochInBand === 0) { + return { ...base, canonical: true, unresolvedEpochTextRows: probe.epochText }; + } + + if (probe.epochInBand > 0) { + // The extra `typeof(...) = 'text'` on the produced value is belt-and-braces + // over the band: no row is rewritten unless the expression has actually + // yielded a string for it, so a `strftime` that declined can never write a + // bare number into a temporal column. + const epochSql = + `update ${table} set ${col} = ${canonicalFromEpochText} where rowid in (` + + `select rowid from ${table} where ${epochTextInBandSql(col)} ` + + `and typeof(${canonicalFromEpochText}) = 'text' limit ?)`; + const phase = await runBatchedUpdate(client, epochSql, batchSize, maxBatches); + base.epochTextRowsConverted = phase.rowsConverted; + base.budgetExhausted ||= phase.budgetExhausted; + } + + if (probe.residual > 0 || base.epochTextRowsConverted > 0) { + // The local twin's statement, batched: same SET expression, same guard. + const convergeSql = + `update ${table} set ${col} = ${canonical} where rowid in (` + + `select rowid from ${table} where ${notCanonicalSql(col, canonical)} limit ?)`; + const phase = await runBatchedUpdate(client, convergeSql, batchSize, maxBatches); + base.rowsConverted = phase.rowsConverted; + base.budgetExhausted ||= phase.budgetExhausted; + } + + const after = readProbe(await client.execute(buildProbeSql(column, canonicalSqlFor))); + base.residualRows = after.residual; + base.pendingEpochTextRows = after.epochInBand; + // Everything digits-only that is left and is NOT still convertible. + base.unresolvedEpochTextRows = Math.max(0, after.epochText - after.epochInBand); + // The ONLY thing that earns the mark: nothing is left for either phase to + // change. `residual` alone would not do — an un-converted TEXT epoch is a + // fixpoint of the shared repair and measures zero there, so a budget-stopped + // run would look complete and strand its remaining rows forever. + base.canonical = after.residual === 0 && after.epochInBand === 0; + return base; + } catch (err) { + return { ...base, error: err instanceof Error ? err.message : String(err) }; + } +} + +/** + * Converge every given column, probing them all in one round-trip first. + * + * Never throws: a column whose statements fail comes back with `error` set and + * `canonical: false`, which the caller reads as "leave the read-side repair + * alone". A migration must never be able to take a boot down (D-B3). + */ +export async function backfillRemoteCanonicalColumns( + client: RemoteBackfillClient, + columns: RemoteBackfillColumn[], + canonicalSqlFor: CanonicalSqlFor, + options: RemoteCanonicalBackfillOptions = {}, + logger?: RemoteBackfillLogger, +): Promise { + if (columns.length === 0) return { columns: [] }; + + const probes = await probeRemoteCanonicalColumns(client, columns, canonicalSqlFor); + const reports: RemoteBackfillColumnReport[] = []; + + for (let i = 0; i < columns.length; i++) { + const column = columns[i]; + const probe = probes[i]; + if (probe && 'error' in probe) { + reports.push({ + ...column, + epochTextRowsConverted: 0, + rowsConverted: 0, + residualRows: 0, + pendingEpochTextRows: 0, + unresolvedEpochTextRows: 0, + canonical: false, + budgetExhausted: false, + error: probe.error, + }); + continue; + } + const report = await backfillRemoteCanonicalColumn( + client, + column, + canonicalSqlFor, + options, + probe, + ); + reports.push(report); + + const where = `${column.table}.${column.field}`; + if (report.error) { + logger?.warn( + `[driver-turso] could not canonicalise remote ${column.kind} storage for ${where}; ` + + `queries stay correct via the read-side repair`, + { error: report.error }, + ); + } else if (report.rowsConverted || report.epochTextRowsConverted) { + logger?.info?.( + `[driver-turso] canonicalised remote ${column.kind} storage (#5770) for ${where}`, + { + rowsConverted: report.rowsConverted, + epochTextRowsConverted: report.epochTextRowsConverted, + canonical: report.canonical, + }, + ); + } + if (report.budgetExhausted) { + logger?.warn( + `[driver-turso] remote ${column.kind} backfill for ${where} stopped on its batch ` + + `budget; the column stays un-marked and resumes on the next run`, + { + rowsConverted: report.rowsConverted, + epochTextRowsConverted: report.epochTextRowsConverted, + residualRows: report.residualRows, + pendingEpochTextRows: report.pendingEpochTextRows, + }, + ); + } + if (report.unresolvedEpochTextRows > 0) { + logger?.warn( + `[driver-turso] ${report.unresolvedEpochTextRows} row(s) in ${where} hold digits-only ` + + `text outside the interpretable epoch-millisecond band and were left untouched ` + + `(cloud#1005 后果 B, unresolvable remainder)`, + { table: column.table, field: column.field }, + ); + } + } + + return { columns: reports }; +} diff --git a/packages/drivers/driver-turso/src/turso-driver.ts b/packages/drivers/driver-turso/src/turso-driver.ts index e178edba36..3007579fc6 100644 --- a/packages/drivers/driver-turso/src/turso-driver.ts +++ b/packages/drivers/driver-turso/src/turso-driver.ts @@ -22,6 +22,13 @@ import { SqlDriver, type SqlDriverConfig } from '@objectstack/driver-sql'; import type { Client } from '@libsql/client'; import { RemoteTransport } from './remote-transport.js'; +import { + backfillRemoteCanonicalColumns, + type RemoteBackfillClient, + type RemoteBackfillColumn, + type RemoteCanonicalBackfillOptions, + type RemoteCanonicalBackfillReport, +} from './remote-canonical-backfill.js'; // ── Transport Mode ─────────────────────────────────────────────────────────── @@ -833,6 +840,108 @@ export class TursoDriver extends SqlDriver { } } + /** + * Converge this driver's REMOTE `Field.datetime` / `Field.time` columns on the + * canonical storage form and mark the ones that are PROVED converged, so their + * filters stop compiling to the unindexable repair expression + * (objectstack#5770, cloud#1005 后果 A; the maintainer's 2026-08-03 方案 1). + * + * The remote-mode counterpart of `SqlDriver.backfillCanonicalDatetimes` / + * `backfillCanonicalTimes`, which are Knex paths and therefore never ran here. + * Same semantics, same expression, same consumption point: this marks + * `canonicalDatetimeFields` / `canonicalTimeFields`, which is exactly what + * `needsLegacyDatetimeRepair` / `needsLegacyTimeRepair` read to drop the + * repair — so remote and local reach the indexable form through one rule + * rather than two. It additionally recovers the TEXT-affinity numeric epochs + * only a remote table can hold (后果 B); see `remote-canonical-backfill.ts` + * for why that limb is safe HERE and was rejected in the shared expression. + * + * Called automatically after remote schema sync, and public so an operator can + * drive a large table to completion with a bigger budget: + * + * ```typescript + * await driver.backfillRemoteCanonicalTemporal({ batchSize: 2000, maxBatches: 5000 }); + * ``` + * + * Never throws and never marks on anything but measured evidence. A column + * that errors, or that a batch budget stopped short, stays unmarked and keeps + * its read-side repair — correct answers, just unindexed. Nothing in any read + * or write path may depend on this having run (ADR-0053 D-B3 / cloud#1003). + * + * A no-op outside remote mode: local and replica reach the same state through + * the inherited Knex backfill during `initObjects`. + */ + async backfillRemoteCanonicalTemporal( + options?: RemoteCanonicalBackfillOptions, + ): Promise { + if (!this.isRemote) return { columns: [] }; + const client = this.remoteTransport?.getClient() as RemoteBackfillClient | null | undefined; + if (!client) return { columns: [] }; + + // Only tables this driver synced remotely, and only columns not already + // marked — re-running is cheap by design, but skipping a proved column + // makes it free. + const columns: RemoteBackfillColumn[] = []; + for (const table of this.remoteManagedObjects) { + for (const field of this.datetimeFields[table] ?? []) { + if (this.canonicalDatetimeFields[table]?.has(field)) continue; + columns.push({ table, field, kind: 'datetime' }); + } + for (const field of this.timeFields[table] ?? []) { + if (this.canonicalTimeFields[table]?.has(field)) continue; + columns.push({ table, field, kind: 'time' }); + } + } + if (columns.length === 0) return { columns: [] }; + + const report = await backfillRemoteCanonicalColumns( + client, + columns, + // The driver's OWN repair expression — handed over, never copied, so the + // backfill cannot drift from the read path it is retiring. + (kind, columnSql) => + kind === 'datetime' + ? this.sqliteCanonicalDatetimeSql(columnSql) + : this.sqliteCanonicalTimeSql(columnSql), + options, + this.logger, + ); + + for (const column of report.columns) { + if (!column.canonical) continue; + const marks = + column.kind === 'datetime' + ? (this.canonicalDatetimeFields[column.table] ??= new Set()) + : (this.canonicalTimeFields[column.table] ??= new Set()); + marks.add(column.field); + } + + return report; + } + + /** + * Run {@link backfillRemoteCanonicalTemporal} off the back of a remote schema + * sync, swallowing everything. + * + * The local twin is invoked from inside `SqlDriver.initObjects` for the same + * reason and with the same posture: schema sync is the moment the driver knows + * which columns are temporal, and a migration must never be able to fail a + * boot. `backfillRemoteCanonicalTemporal` already reports rather than throws; + * this catch covers the paths that could still reject (a client lost between + * the sync and here). + */ + private async backfillRemoteCanonicalTemporalQuietly(): Promise { + try { + await this.backfillRemoteCanonicalTemporal(); + } catch (err) { + this.logger.warn( + `[driver-turso] remote canonical temporal backfill failed; ` + + `queries stay correct via the read-side repair`, + { error: err instanceof Error ? err.message : String(err) }, + ); + } + } + // =================================== // Bulk Operations (remote mode overrides) // =================================== @@ -911,6 +1020,10 @@ export class TursoDriver extends SqlDriver { // Key strictly by `object` (what find()/formatOutput look up) — never let a // stray `schema.name` shadow it. this.registerRemoteFieldMetadata({ ...(schema as Record), name: object }); + // #5770: the remote twin of the `backfillCanonicalDatetimes` call + // `SqlDriver.initObjects` makes at exactly this point. Must run AFTER the + // registration above — that is what tells it which columns are temporal. + await this.backfillRemoteCanonicalTemporalQuietly(); return; } return super.syncSchema(object, schema, options); @@ -942,6 +1055,11 @@ export class TursoDriver extends SqlDriver { // a boolean reads back as raw 0/1, JSON as a string, dates as raw text. // (Root cause of the 2026-07-06 case_escalation `1 != true` incident.) for (const obj of objects) this.registerRemoteFieldMetadata(obj); + // #5770: the remote twin of the `backfillCanonicalDatetimes` / + // `backfillCanonicalTimes` calls `SqlDriver.initObjects` makes per table. + // One batched probe covers every column synced here, so the steady state + // (nothing to converge) costs a single round-trip for the whole boot. + await this.backfillRemoteCanonicalTemporalQuietly(); return; } return super.initObjects(objects); diff --git a/packages/drivers/driver-turso/src/turso-remote-temporal-conformance.test.ts b/packages/drivers/driver-turso/src/turso-remote-temporal-conformance.test.ts index cd801afa84..3b7a96a415 100644 --- a/packages/drivers/driver-turso/src/turso-remote-temporal-conformance.test.ts +++ b/packages/drivers/driver-turso/src/turso-remote-temporal-conformance.test.ts @@ -213,15 +213,49 @@ describe('TursoDriver remote — Field.time conformance', () => { } }); +/** + * Drop a column's "already backfilled" marker — the remote twin of + * `LegacyStorageDriver.forgetCanonical` in `driver-sql`, and needed here for + * the same reason since objectstack#5770 gave remote mode a backfill. + * + * A legacy sweep needs BOTH halves to be honest: rows seeded raw, AND the + * canonical marker cleared. Without the second, the driver would correctly + * conclude the column is clean, skip the read-side repair, and the sweep would + * measure the wrong thing. + */ +const forgetCanonical = (driver: TursoDriver, table: string, field: string, kind: 'datetime' | 'time') => { + const key = kind === 'datetime' ? 'canonicalDatetimeFields' : 'canonicalTimeFields'; + (driver as never as Record>>)[key][table]?.delete(field); +}; + +/** + * The driver's own answer to "does this column still owe the read-side repair?" + * — the remote twin of `LegacyStorageDriver.legacyDatetimeRepairApplies`, + * exposed so a sweep can ASSERT its premise instead of stating it in a comment. + */ +const repairApplies = (driver: TursoDriver, table: string, field: string, kind: 'datetime' | 'time') => { + const method = kind === 'datetime' ? 'needsLegacyDatetimeRepair' : 'needsLegacyTimeRepair'; + return (driver as never as Record boolean>)[method].call( + driver, + table, + field, + ); +}; + /** * Sweep 3 — the same matrix over un-backfilled legacy `datetime` storage. * * Seeded through the stub's raw handle: BELOW the transport, so no write path * can converge the forms (the remote twin of the local suite's * `LegacyStorageTursoDriver`, which inserts through Knex for the same reason). - * Nothing needs un-marking afterwards the way local does — remote mode has no - * backfill to mark a column canonical in the first place, which is exactly why - * this sweep is not hypothetical here. + * + * The marker is then cleared, exactly as the local twin does. Until + * objectstack#5770 that was unnecessary here — remote mode had no backfill, so + * no column was ever marked canonical, and this sweep was un-backfilled by + * accident. Now that a remote backfill exists, "un-backfilled" is a state the + * fixture must DECLARE rather than inherit: it is the state of a database whose + * backfill hit its batch budget or could not run, which D-B3 requires to keep + * answering correctly. That is precisely what this sweep measures. */ describe('TursoDriver remote — temporal conformance on un-backfilled legacy storage', () => { let driver: TursoDriver; @@ -250,6 +284,7 @@ describe('TursoDriver remote — temporal conformance on un-backfilled legacy st : r.at.replace('T', ' ').replace('Z', ''); insert.run(r.id, at, r.on, r.why); } + forgetCanonical(driver, 'conformance', 'at', 'datetime'); }); afterAll(async () => { @@ -270,6 +305,11 @@ describe('TursoDriver remote — temporal conformance on un-backfilled legacy st expect(row.at, `${row.id}.at`).not.toBe(r.at); expect(row.at, `${row.id}.at`).toMatch(r.writerForm === 'native' ? /\+08:00$/ : / /); } + // The other half of the premise (#5770): the driver must still OWE this + // column its repair. Without this, a future change that re-marked the + // column would leave the sweep below passing for the wrong reason — it + // would be measuring canonical storage under a legacy-storage name. + expect(repairApplies(driver, 'conformance', 'at', 'datetime')).toBe(true); }); // Literal spellings only, as in the local legacy sweep: the token axis is @@ -301,6 +341,7 @@ describe('TursoDriver remote — Field.time conformance on un-backfilled legacy r.writerForm === 'native' ? `2026-07-28T${r.at}Z` : `2026-07-28 ${r.at}`; insert.run(r.id, at, r.why); } + forgetCanonical(driver, 'time_conformance', 'at', 'time'); }); afterAll(async () => { @@ -319,6 +360,8 @@ describe('TursoDriver remote — Field.time conformance on un-backfilled legacy // defect this axis is about — so none of them is canonical wall clock. expect(row.at, `${row.id}.at`).toMatch(/^2026-07-28[T ]/); } + // See the `datetime` twin above (#5770). + expect(repairApplies(driver, 'time_conformance', 'at', 'time')).toBe(true); }); for (const c of TEMPORAL_TIME_CASES) {