kevchess15/deepseek-harness
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(storage): version read compatibility and backup-and-skip salvage for per-record units

Browse source 
A DomainSpec may declare compatibleVersions: older domain versions whose
stored records the current record schemas still accept. The json backend's
per-record reads admit documents stamped with a declared version (writes
always stamp the current one), and the legacy whole-unit bootstrap migrates
only a file whose stored version is in the accepted set — previously it
migrated any version and stamped the records current, turning a discardable
stale cache into invalid-record failures that refused the whole domain at
open and permanently poisoned the new tree on first boot.

A DomainSpec may also declare invalidRecords: 'backup-and-skip' for domains
whose records are disposable derived data: a stored record failing its zod
schema is moved aside through the new optional KvUnit.backupRecord
(<key>.json.bak.<YYYYMMDDHHmm> under the json backend), logged with its
cause, and skipped, instead of rejecting the open. The default stays
fail-loud, and so do backends without backupRecord.
This commit is contained in:
imccyu 2026-09-02 14:07:58 +08:00
parent 4e84901e64
commit fcd109d29a
7 changed files with 259 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
packages/storage/storage-domain/src/index.ts
View file

@ -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<string, unknown>()
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)
}

34
packages/storage/storage-domain/src/spec.ts
View file

@ -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<unknown>
/** Table declarations keyed by table name; each name must match `UNIT_NAME_RE`. */
@ -91,6 +111,13 @@ export function defineDomain<S extends DomainSpec>(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<S extends DomainSpec>(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 },
}
}

41
packages/storage/storage-domain/tests/domain.spec.ts
View file

@ -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<string, Item>(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)

18
packages/storage/storage-json/src/format.ts
View file

@ -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<string, unknown>
if (stamped !== version) return undefined
if (typeof stamped !== 'number' || !versions.includes(stamped)) return undefined
return record
}

75
packages/storage/storage-json/src/per-record-unit.ts
View file

@ -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 `<root>/<name>.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<UnitState> {
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
* (`<root>/<name>.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 (`<root>/<name>`).
* @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<string, Record<string, unknown>>
@ -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<string, unknown>, version: number, dir: string): Promise<boolean> {
async function loadTableRecords(records: Map<string, unknown>, versions: readonly number[], dir: string): Promise<boolean> {
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<string, unknown>, version: number,
}
/** Read one record document; a foreign (unreadable or stale) one reads as absent. */
async function readRecord(path: string, version: number): Promise<unknown> {
async function readRecord(path: string, versions: readonly number[]): Promise<unknown> {
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 `<key>.json.bak.<YYYYMMDDHHmm>`. 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<string> {
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<void> {
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)) {

78
packages/storage/storage-json/tests/json-backend.spec.ts
View file

@ -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 })

23
packages/storage/storage/src/backend.ts
View file

@ -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<void>
/**
* 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<string>
/**
* Write the global singleton durably. Only valid when the descriptor
* declared `hasGlobal`.