diff --git a/packages/storage/storage-domain/src/index.ts b/packages/storage/storage-domain/src/index.ts index d2c16a3d69..7e0be5e151 100644 --- a/packages/storage/storage-domain/src/index.ts +++ b/packages/storage/storage-domain/src/index.ts @@ -88,7 +88,10 @@ export class DomainFacility { * (`facet-unsupported`); open the unit projected from the spec (backend * `version-mismatch`/`malformed-medium` pass through); load and validate * every stored record against the spec's zod schemas (`invalid-record` - * with the offending table and key); construct the domain. + * with the offending table and key — unless the spec declares + * `invalidRecords: 'backup-and-skip'` and the unit can move documents aside, in + * which case the failing record is backed up, logged, and skipped); + * construct the domain. * * Lifecycle: the CALLER owns the returned handle and closes it via * `Domain.close()` (typically as its own `ctx.effect` disposer) — the @@ -118,7 +121,23 @@ export class DomainFacility { for (const [table, tableSpec] of Object.entries(spec.tables)) { const records = new Map() for (const [key, raw] of Object.entries(snapshot.tables[table] ?? {})) { - records.set(key, parseRecord(spec.name, table, key, () => tableSpec.valueSchema.parse(raw))) + let parsed: unknown + try { + parsed = parseRecord(spec.name, table, key, () => tableSpec.valueSchema.parse(raw)) + } catch (error) { + // Backup-and-skip policy (disposable derived data): move the record's + // document aside, log the concrete failure, and open without the + // record. Backends that cannot move a document keep the loud path. + if (spec.invalidRecords !== 'backup-and-skip' || unit.backupRecord === undefined) throw error + const moved = await unit.backupRecord(table, key) + // parseRecord always wraps the zod failure as the cause. + this.ctx.logger.error( + `domain '${spec.name}': stored record '${key}' in table '${table}' failed schema validation; ` + + `moved to '${moved}' and treated as absent. Cause: ${String((error as DomainError).cause)}`, + ) + continue + } + records.set(key, parsed) } tables.set(table, records) } diff --git a/packages/storage/storage-domain/src/spec.ts b/packages/storage/storage-domain/src/spec.ts index 74348918aa..f9267bcba5 100644 --- a/packages/storage/storage-domain/src/spec.ts +++ b/packages/storage/storage-domain/src/spec.ts @@ -45,6 +45,26 @@ export interface DomainSpec { * (a stale record document is discarded, never migrated). */ readonly layout?: 'single' | 'per-record' + /** + * Older domain versions whose stored records the current record schemas + * also accept (the declaring owner vouches for that, typically by + * declaring the fields older records lack as optional). `per-record` backends + * read documents stamped with a listed version instead of discarding them, + * and accept a legacy whole-unit file so stamped for the one-time + * bootstrap; writes always stamp {@link version}. + */ + readonly compatibleVersions?: readonly number[] + /** + * What `open` does with a stored table record that fails its zod schema. + * Absent (the default), the whole open rejects with `invalid-record` — + * right for authoritative data. `'backup-and-skip'` is for domains whose + * records are disposable derived data: the backend moves the record's + * document aside (`KvUnit.backupRecord`), the failure is logged with + * its cause, and the open continues with the record absent. A backend + * without `backupRecord` (no per-record document to move) falls back + * to the rejecting default. The global slot always rejects. + */ + readonly invalidRecords?: 'backup-and-skip' /** Optional global singleton slot. */ readonly global?: DomainGlobalSpec /** Table declarations keyed by table name; each name must match `UNIT_NAME_RE`. */ @@ -91,6 +111,13 @@ export function defineDomain(spec: S): S { if (!Number.isInteger(spec.version) || spec.version < 0) { throw new Error(`domain '${spec.name}' version must be a non-negative integer, got ${spec.version}`) } + for (const compat of spec.compatibleVersions ?? []) { + if (!Number.isInteger(compat) || compat < 0 || compat >= spec.version) { + throw new Error( + `domain '${spec.name}' compatibleVersions entries must be non-negative integers below version ${spec.version}, got ${compat}`, + ) + } + } if (spec.layout !== undefined) { // Runtime boundary: the union type is compile-time only — a spec built // from config could carry any value, and a bad one must fail loud here. @@ -99,6 +126,12 @@ export function defineDomain(spec: S): S { throw new Error(`domain '${spec.name}' layout must be 'single' or 'per-record', got ${layout}`) } } + if (spec.invalidRecords !== undefined) { + const policy: string = spec.invalidRecords + if (policy !== 'backup-and-skip') { + throw new Error(`domain '${spec.name}' invalidRecords must be 'backup-and-skip' when present, got ${policy}`) + } + } for (const table of Object.keys(spec.tables)) { if (!UNIT_NAME_RE.test(table)) { throw new Error(`domain '${spec.name}' table name '${table}' must match ${UNIT_NAME_RE}`) @@ -125,5 +158,6 @@ export function descriptorOf(spec: DomainSpec): KvUnitDescriptor { tables: Object.keys(spec.tables), hasGlobal: spec.global !== undefined, ...spec.layout === undefined ? {} : { layout: spec.layout }, + ...spec.compatibleVersions === undefined ? {} : { compatibleVersions: spec.compatibleVersions }, } } diff --git a/packages/storage/storage-domain/tests/domain.spec.ts b/packages/storage/storage-domain/tests/domain.spec.ts index 3b3c8fa679..b21ec0faf8 100644 --- a/packages/storage/storage-domain/tests/domain.spec.ts +++ b/packages/storage/storage-domain/tests/domain.spec.ts @@ -58,6 +58,25 @@ describe('defineDomain', () => { })).toThrow(/must not accept null/) }) + it('validates compatibleVersions entries and projects them onto the descriptor', () => { + expect(() => defineDomain({ name: 'ok', version: 2, compatibleVersions: [1.5], tables: {} })) + .toThrow(/compatibleVersions/) + expect(() => defineDomain({ name: 'ok', version: 2, compatibleVersions: [2], tables: {} })) + .toThrow(/below version/) + expect(() => defineDomain({ name: 'ok', version: 2, compatibleVersions: [-1], tables: {} })) + .toThrow(/compatibleVersions/) + expect(descriptorOf(defineDomain({ name: 'ok', version: 2, compatibleVersions: [0, 1], tables: {} }))) + .toMatchObject({ compatibleVersions: [0, 1] }) + // An undeclared set is absent from the descriptor. + expect(descriptorOf(spec)).not.toHaveProperty('compatibleVersions') + }) + + it('rejects an unknown invalidRecords policy', () => { + expect(() => defineDomain({ + name: 'ok', version: 1, invalidRecords: 'zap' as 'backup-and-skip', tables: {}, + })).toThrow(/invalidRecords/) + }) + it('rejects an invalid layout and projects the declared one onto the descriptor', () => { // A spec built from config can carry any value; the union type is // compile-time only, so the runtime boundary check must reject it. @@ -139,6 +158,28 @@ describe('DomainFacility.open', () => { }) }) + it('keeps the rejecting default under backup-and-skip when the backend cannot move documents', async () => { + // The memory backend has no backupRecord, so the declared policy cannot + // apply and the open falls back to failing loud. + const salvageSpec = defineDomain({ + name: 'salvage', + version: 1, + invalidRecords: 'backup-and-skip', + tables: { items: domainTable(itemSchema) }, + }) + const pool = new MemoryMediaPool() + { + const { facility } = await harness({ pool }) + await (await facility.open(salvageSpec)).table('items').put('bad', { label: 'x', count: 2 }) + } + pool.media.get('salvage')!.tables.get('items')!.set('bad', { label: 'x', count: 'NaN' }) + const { facility } = await harness({ pool }) + await expect(facility.open(salvageSpec)).rejects.toMatchObject({ + code: 'invalid-record', + detail: { table: 'items', key: 'bad' }, + }) + }) + it('rejects a stored global that fails its schema with the global marker', async () => { const pool = new MemoryMediaPool() pool.versions.set('demo', 1) diff --git a/packages/storage/storage-json/src/format.ts b/packages/storage/storage-json/src/format.ts index 045485eaba..22cbe8f703 100644 --- a/packages/storage/storage-json/src/format.ts +++ b/packages/storage/storage-json/src/format.ts @@ -100,16 +100,18 @@ export function serializeRecord(version: number, value: unknown): string { /** * Parse one per-record document, validating its version stamp. A document - * that is malformed or stamped with a different version is FOREIGN and reads - * as absent — the per-record contract: one bad or stale record file must not - * brick the whole unit, and a version bump discards stale records instead of - * migrating them (the whole-unit format rejects instead, because there is - * exactly one document). + * that is malformed or stamped with an unaccepted version is FOREIGN and + * reads as absent — the per-record contract: one bad or stale record file + * must not brick the whole unit, and a version bump discards stale records + * instead of migrating them (the whole-unit format rejects instead, because + * there is exactly one document). * @param text - Raw per-record document content. - * @param version - Expected unit version; a mismatch discards the document. + * @param versions - Accepted unit versions (the current one plus the + * descriptor's compatibleVersions); any other stamp discards the + * document. * @returns the record value, or `undefined` for a foreign document. */ -export function parseRecord(text: string, version: number): unknown { +export function parseRecord(text: string, versions: readonly number[]): unknown { let document: unknown try { document = JSON.parse(text) @@ -118,6 +120,6 @@ export function parseRecord(text: string, version: number): unknown { } if (typeof document !== 'object' || document === null) return undefined const { version: stamped, record } = document as Record - if (stamped !== version) return undefined + if (typeof stamped !== 'number' || !versions.includes(stamped)) return undefined return record } diff --git a/packages/storage/storage-json/src/per-record-unit.ts b/packages/storage/storage-json/src/per-record-unit.ts index b75f9f9452..346c7a0d6a 100644 --- a/packages/storage/storage-json/src/per-record-unit.ts +++ b/packages/storage/storage-json/src/per-record-unit.ts @@ -10,20 +10,23 @@ * memory unchanged. * * Per-record contract: a record document that is malformed or stamped with a - * different version reads as an absent record — one bad or stale file never - * bricks the whole unit, and a version bump discards stale records instead - * of migrating them. Record keys become path segments, so they must be - * path-safe (`[a-zA-Z0-9_-]+`); an unsafe key rejects at write. + * version outside the accepted set (the descriptor's current version plus + * its `compatibleVersions`) reads as an absent record — one bad or stale + * file never bricks the whole unit, and a version bump discards stale + * records instead of migrating them. Record keys become path segments, so + * they must be path-safe (`[a-zA-Z0-9_-]+`); an unsafe key rejects at write. * * Legacy bootstrap: when the new tree has no document path, a legacy * whole-unit file `/.json` (the pre-per-record layout) seeds - * per-record documents. Any new document path, including one whose contents - * are unreadable or stale, suppresses the bootstrap for the whole unit. The + * per-record documents, provided its stored unit version is in the accepted + * set — a legacy file stamped with any other version is left alone and reads + * as the empty unit. Any new document path, including one whose contents are + * unreadable or stale, suppresses the bootstrap for the whole unit. The * legacy file is never changed or deleted. * @module @deepseek-ai/dsh-storage-json/src/per-record-unit */ -import { mkdir, readFile, readdir, rm } from 'node:fs/promises' +import { mkdir, readFile, readdir, rename, rm } from 'node:fs/promises' import { dirname, join } from 'node:path' import type { Dirent } from 'node:fs' import { StorageError } from '@deepseek-ai/dsh-storage' @@ -64,6 +67,7 @@ export async function openPerRecordUnit( * @returns the authoritative state reconstructed from the tree. */ async function loadPerRecordState(descriptor: KvUnitDescriptor, dir: string): Promise { + const versions = acceptedStamps(descriptor) const state: UnitState = { version: descriptor.version, global: null, @@ -83,11 +87,11 @@ async function loadPerRecordState(descriptor: KvUnitDescriptor, dir: string): Pr if (entry.isDirectory()) { const records = state.tables.get(entry.name) if (records !== undefined) { - return loadTableRecords(records, descriptor.version, join(dir, entry.name)) + return loadTableRecords(records, versions, join(dir, entry.name)) } } if (entry.name === 'global.json' && descriptor.hasGlobal) { - const global = await readRecord(join(dir, entry.name), descriptor.version) + const global = await readRecord(join(dir, entry.name), versions) if (global !== undefined) state.global = global return true } @@ -97,12 +101,21 @@ async function loadPerRecordState(descriptor: KvUnitDescriptor, dir: string): Pr return state } +/** The version stamps this unit reads as its own: current plus declared compatible versions. */ +function acceptedStamps(descriptor: KvUnitDescriptor): readonly number[] { + return [descriptor.version, ...descriptor.compatibleVersions ?? []] +} + /** * Bootstrap an empty per-record tree from a legacy whole-unit file * (`/.json`, the pre-per-record layout). Every declared-table * record is copied into a current-version document, while the legacy file is * retained unchanged. A missing, foreign (another unit's name), malformed, - * or non-unit legacy file is left alone; other read failures propagate. + * or non-unit legacy file is left alone, and so is one whose stored unit + * version is outside the accepted set — migrating records the owner never + * vouched for would stamp them with the current version and turn a + * discardable stale cache into schema failures at the domain layer. Other + * read failures propagate. * @param descriptor - Static identity and shape of the unit. * @param dir - The per-record unit directory (`/`). * @param state - The empty tree state; bootstrapped records are added. @@ -116,16 +129,18 @@ async function bootstrapLegacyUnit(descriptor: KvUnitDescriptor, dir: string, st if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error return } - // The legacy document is runtime data: only `unit.name` and the tables map - // shape are checked here — the record values are migrated as-is and the - // domain layer's schemas judge them. - let document: { unit?: { name?: unknown }; tables?: unknown } + // The legacy document is runtime data: only `unit.name`, `unit.version`, + // and the tables map shape are checked here — the record values are + // migrated as-is and the domain layer's schemas judge them. + let document: { unit?: { name?: unknown; version?: unknown }; tables?: unknown } try { - document = JSON.parse(text) as { unit?: { name?: unknown }; tables?: unknown } + document = JSON.parse(text) as { unit?: { name?: unknown; version?: unknown }; tables?: unknown } } catch { return // Malformed legacy file: not ours to interpret or delete. } if (document.unit?.name !== descriptor.name) return + const stamped = document.unit.version + if (typeof stamped !== 'number' || !acceptedStamps(descriptor).includes(stamped)) return const tables = document.tables if (typeof tables !== 'object' || tables === null) return const recordsByTable = tables as Record> @@ -146,14 +161,14 @@ async function bootstrapLegacyUnit(descriptor: KvUnitDescriptor, dir: string, st * @returns whether the directory contains any `.json` document path, * independently of key safety, readability, or stored version. */ -async function loadTableRecords(records: Map, version: number, dir: string): Promise { +async function loadTableRecords(records: Map, versions: readonly number[], dir: string): Promise { const files = await readdir(dir, { withFileTypes: true }) const hasDocuments = files.some(file => file.name.endsWith('.json')) const loaded = await Promise.all(files.map(async (file) => { if (!file.name.endsWith('.json')) return const key = file.name.slice(0, -'.json'.length) if (!SAFE_KEY_RE.test(key)) return - const record = await readRecord(join(dir, file.name), version) + const record = await readRecord(join(dir, file.name), versions) if (record !== undefined) return [key, record] as const })) for (const record of loaded) { @@ -163,9 +178,9 @@ async function loadTableRecords(records: Map, version: number, } /** Read one record document; a foreign (unreadable or stale) one reads as absent. */ -async function readRecord(path: string, version: number): Promise { +async function readRecord(path: string, versions: readonly number[]): Promise { try { - return parseRecord(await readFile(path, 'utf8'), version) + return parseRecord(await readFile(path, 'utf8'), versions) } catch { return undefined } @@ -213,6 +228,22 @@ export class PerRecordJsonUnit implements KvUnit { await this.tracked(rm(join(this.tableDir(table), `${key}.json`), { force: true })) } + /** + * Move one record's document aside as `.json.bak.`. The + * moved file no longer ends in `.json`, so every later read ignores it; the + * bytes stay on disk for inspection. A same-minute backup of the same + * key overwrites the previous backup (the newer bytes are the ones worth + * keeping). + */ + async backupRecord(table: string, key: string): Promise { + this.assertOpen() + assertSafeKey(this.descriptor.name, key) + const path = join(this.tableDir(table), `${key}.json`) + const moved = `${path}.bak.${backupStamp(new Date())}` + await this.tracked(rename(path, moved)) + return moved + } + /** Durably replace the global singleton. Only valid when declared. */ async setGlobal(value: unknown): Promise { this.assertOpen() @@ -267,6 +298,12 @@ export class PerRecordJsonUnit implements KvUnit { } } +/** Local-time `YYYYMMDDHHmm` suffix for backed-up documents. */ +function backupStamp(now: Date): string { + const pad = (value: number): string => String(value).padStart(2, '0') + return `${String(now.getFullYear())}${pad(now.getMonth() + 1)}${pad(now.getDate())}${pad(now.getHours())}${pad(now.getMinutes())}` +} + /** Reject a record key that would be unsafe as a path segment. */ function assertSafeKey(unit: string, key: string): void { if (!SAFE_KEY_RE.test(key)) { diff --git a/packages/storage/storage-json/tests/json-backend.spec.ts b/packages/storage/storage-json/tests/json-backend.spec.ts index bafbf92ef4..57ad9d7633 100644 --- a/packages/storage/storage-json/tests/json-backend.spec.ts +++ b/packages/storage/storage-json/tests/json-backend.spec.ts @@ -329,10 +329,10 @@ describe('per-record layout', () => { it('bootstraps an empty per-record tree from a legacy whole-unit file and preserves it', async () => { const root = await freshRoot() - // A legacy single-layout file for the same unit (any older version); - // the extra table is not declared and must be skipped. + // A legacy single-layout file for the same unit, stamped with the current + // version; the extra table is not declared and must be skipped. const legacy = JSON.stringify({ - unit: { name: 'recs', version: 3 }, + unit: { name: 'recs', version: 2 }, global: null, tables: { t: { old1: { v: 1 }, old2: { v: 2 } }, undeclared: { k: { v: 0 } } }, }) @@ -347,6 +347,75 @@ describe('per-record layout', () => { await backend.close() }) + it('bootstraps from a legacy file only when its stored version is accepted', async () => { + // Version 3 is neither current (2) nor declared compat: the legacy file + // is left alone and the unit reads empty — migrating unvouched records + // would stamp them current and surface as schema failures at the domain + // layer instead of a discardable stale cache. + const root = await freshRoot() + const legacy = JSON.stringify({ + unit: { name: 'recs', version: 3 }, + global: null, + tables: { t: { old: { v: 1 } } }, + }) + await writeFile(join(root, 'recs.json'), legacy, 'utf8') + const backend = new JsonStorageBackend(root) + const unit = await backend.kv.open(descriptor) + expect(await unit.loadAll()).toEqual({ tables: { t: {} }, global: null }) + await expect(readFile(join(root, 'recs.json'), 'utf8')).resolves.toBe(legacy) + await unit.close() + await backend.close() + + // The same file bootstraps once version 3 is declared read-compatible… + const root2 = await freshRoot() + await writeFile(join(root2, 'recs.json'), legacy, 'utf8') + const backend2 = new JsonStorageBackend(root2) + const compat = { ...descriptor, version: 4, compatibleVersions: [3] } + const unit2 = await backend2.kv.open(compat) + expect(await unit2.loadAll()).toEqual({ tables: { t: { old: { v: 1 } } }, global: null }) + // …and the migrated documents are stamped with the CURRENT version. + expect(JSON.parse(await readFile(join(root2, 'recs', 't', 'old.json'), 'utf8'))) + .toEqual({ version: 4, record: { v: 1 } }) + await unit2.close() + await backend2.close() + }) + + it('backupRecord moves the document aside; reads see it absent and a write recreates it', async () => { + const root = await freshRoot() + const backend = new JsonStorageBackend(root) + const unit = await backend.kv.open(descriptor) + await unit.putRecord('t', 'k', { v: 1 }) + const moved = await unit.backupRecord!('t', 'k') + expect(moved).toMatch(/k\.json\.bak\.\d{12}$/) + expect(JSON.parse(await readFile(moved, 'utf8'))).toEqual({ version: 2, record: { v: 1 } }) + await expect(readFile(recordPath(root, 'k'), 'utf8')).rejects.toMatchObject({ code: 'ENOENT' }) + // The moved file no longer ends in .json, so it reads as absent… + expect(await unit.loadAll()).toEqual({ tables: { t: {} }, global: null }) + // …and the key is free for a fresh write. + await unit.putRecord('t', 'k', { v: 2 }) + expect(await unit.loadAll()).toEqual({ tables: { t: { k: { v: 2 } } }, global: null }) + await expect(unit.backupRecord!('t', 'a/b')).rejects.toThrow(/not path-safe/) + await unit.close() + await expect(unit.backupRecord!('t', 'k')).rejects.toMatchObject({ code: 'closed' }) + await backend.close() + }) + + it('reads per-record documents stamped with a declared compat version and stamps writes current', async () => { + const root = await freshRoot() + const backend = new JsonStorageBackend(root) + const compat = { ...descriptor, compatibleVersions: [1] } + await mkdir(join(root, 'recs', 't'), { recursive: true }) + await writeFile(recordPath(root, 'oldrec'), JSON.stringify({ version: 1, record: { v: 'old' } }), 'utf8') + await writeFile(recordPath(root, 'ancient'), JSON.stringify({ version: 0, record: { v: 'no' } }), 'utf8') + const unit = await backend.kv.open(compat) + // Version 1 is declared compat and served; version 0 is not and discards. + expect(await unit.loadAll()).toEqual({ tables: { t: { oldrec: { v: 'old' } } }, global: null }) + await unit.putRecord('t', 'oldrec', { v: 'new' }) + expect(JSON.parse(await readFile(recordPath(root, 'oldrec'), 'utf8'))) + .toEqual({ version: 2, record: { v: 'new' } }) + await backend.close() + }) + it('ignores the legacy whole-unit file when any new document path exists', async () => { const root = await freshRoot() const legacy = JSON.stringify({ @@ -400,7 +469,8 @@ describe('per-record layout', () => { await backend4.close() const root5 = await freshRoot() - await writeFile(join(root5, 'recs.json'), JSON.stringify({ unit: { name: 'recs' }, tables: 'not an object' }), 'utf8') + // A current-version stamp so the shapeless `tables` is what stops the bootstrap. + await writeFile(join(root5, 'recs.json'), JSON.stringify({ unit: { name: 'recs', version: 2 }, tables: 'not an object' }), 'utf8') const backend5 = new JsonStorageBackend(root5) const unit5 = await backend5.kv.open(descriptor) expect(await unit5.loadAll()).toEqual({ tables: { t: {} }, global: null }) diff --git a/packages/storage/storage/src/backend.ts b/packages/storage/storage/src/backend.ts index 9cba748655..4a8e0ac2d8 100644 --- a/packages/storage/storage/src/backend.ts +++ b/packages/storage/storage/src/backend.ts @@ -61,6 +61,16 @@ export interface KvUnitDescriptor { * foreign documents. */ readonly layout?: 'single' | 'per-record' + /** + * Older unit versions whose stored records are also readable under the + * declaring owner's current record schemas (the owner vouches for that — + * typically by declaring the fields old records lack as optional). Reads of + * a `per-record` unit accept documents stamped with any listed version, and + * the legacy whole-unit bootstrap accepts a legacy file stamped with one; + * writes always stamp {@link version}. `single`-layout reads stay + * exact-version. + */ + readonly compatibleVersions?: readonly number[] } /** @@ -99,6 +109,19 @@ export interface KvUnit { */ deleteRecord(table: string, key: string): Promise + /** + * Move one record's stored document out of the unit's readable set, + * preserving its bytes for inspection instead of deleting them. Backends + * whose medium has no per-record document to move (the `single` layout, a + * row store) omit this member, and the caller falls back to its + * reject-loud path. Absent after the move: a later {@link loadAll} reads + * the key as missing and a later {@link putRecord} recreates it fresh. + * @param table - Declared table name. + * @param key - Record key. + * @returns the medium location the document was moved to (diagnostics). + */ + backupRecord?(table: string, key: string): Promise + /** * Write the global singleton durably. Only valid when the descriptor * declared `hasGlobal`.