From b1def85c366abf04c0dc2bc505ee5f66b316d2a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 14:22:30 +0000 Subject: [PATCH] =?UTF-8?q?feat(driver-turso):=20remote=20canonical=20temp?= =?UTF-8?q?oral=20backfill=20=E2=80=94=20batched,=20resumable,=20marked=20?= =?UTF-8?q?only=20when=20proved=20(#5770)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SqlDriver.backfillCanonicalDatetimes` / `backfillCanonicalTimes` are Knex paths, and TursoDriver's remote mode routes all DDL and CRUD over `@libsql/client`, so they never ran there. `canonicalDatetimeFields` / `canonicalTimeFields` stayed empty in remote mode, `needsLegacyDatetimeRepair` was permanently true, and every `Field.datetime` / `Field.time` filter compiled to the unindexable repair CASE with no exit (cloud#1005 consequence A). Adds `remote-canonical-backfill.ts` and `TursoDriver.backfillRemoteCanonicalTemporal()`, run automatically after remote `initObjects` / `syncSchema` — the position local mode calls its backfill from: - batched via `rowid IN (SELECT ... LIMIT ?)` (portable; `UPDATE ... LIMIT` needs a compile-time option libSQL does not promise); - resumable with no checkpoint state — the WHERE guard IS the checkpoint; - marks canonical only when a post-pass probe proves nothing is left for either phase to change, writing to the same maps `needsLegacyDatetimeRepair` reads; - recovers TEXT-affinity numeric epochs (consequence B) by feeding `cast(col as real)` back into the driver's OWN expression, so no second epoch conversion rule is introduced; - interprets only 1e12 <= v < 4102444800000, which puts every plausible epoch-SECONDS value out of band; anything outside is left on disk and counted. Correctness posture is unchanged (ADR-0053 D-B3 / cloud#1003): every failure mode degrades to "column unmarked, reads keep the repair, answers correct", and nothing may depend on the backfill having run. The two remote legacy conformance sweeps now clear the canonical marker explicitly, as `driver-sql`'s `LegacyStorageDriver.forgetCanonical` does, and assert the repair still applies — "un-backfilled" used to hold by accident because remote had no backfill at all. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01WyvqvKMG6asi9aXjKE6xtx --- ...urso-remote-canonical-temporal-backfill.md | 77 +++ packages/drivers/driver-turso/src/index.ts | 22 + .../src/remote-canonical-backfill.test.ts | 621 ++++++++++++++++++ .../src/remote-canonical-backfill.ts | 561 ++++++++++++++++ .../drivers/driver-turso/src/turso-driver.ts | 118 ++++ .../turso-remote-temporal-conformance.test.ts | 49 +- 6 files changed, 1445 insertions(+), 3 deletions(-) create mode 100644 .changeset/turso-remote-canonical-temporal-backfill.md create mode 100644 packages/drivers/driver-turso/src/remote-canonical-backfill.test.ts create mode 100644 packages/drivers/driver-turso/src/remote-canonical-backfill.ts 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) {