From 4c9b61e8f7c5bf9c8e75a4dbe4dc085df92e7c60 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Thu, 30 Jul 2026 16:34:27 -0600 Subject: [PATCH 1/4] =?UTF-8?q?fix(migration):=20write=20the=20version/met?= =?UTF-8?q?adata=20prefix=20on=20LMDB=E2=86=92RocksDB=20migrated=20records?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Since #1307 the RecordEncoder encode hook's non-versioned opt-out read this.useVersions off whatever encoder it was grafted onto; copyDb grafts it onto the migration target's plain msgpackr encoder (useVersions undefined), so every migrated record was stored prefix-less: version silently dropped, and point reads returned prototype-less plain objects while scans repaired them (harper#2012, the dev CM invisible-clusters bug). - RecordEncoder: opt-out now requires explicit useVersions === false - PrimaryRocksDatabase: METADATA presence check (flags word 0 no longer leaks the decode wrapper as the value); metadata-less values get the structPrototype repair on point reads, matching getRange and the LMDB wrapper, healing already-migrated instances - copyDb: version-less source records encode plain instead of an undecodable flags-only prefix; post-copy tripwire fails the migration loudly if a versioned record round-trips without its prefix Co-Authored-By: Claude Fable 5 --- bin/copyDb.ts | 61 +++++++++++--- resources/PrimaryRocksDatabase.ts | 16 ++-- resources/RecordEncoder.ts | 6 +- .../bin/migrationCanonicalStructures.test.js | 15 +++- unitTests/bin/migrationMetadataPrefix.test.js | 84 +++++++++++++++++++ .../primaryRocksMetadataRepair.test.js | 71 ++++++++++++++++ 6 files changed, 233 insertions(+), 20 deletions(-) create mode 100644 unitTests/bin/migrationMetadataPrefix.test.js create mode 100644 unitTests/resources/primaryRocksMetadataRepair.test.js diff --git a/bin/copyDb.ts b/bin/copyDb.ts index 4d8fa640c..53f2651bd 100644 --- a/bin/copyDb.ts +++ b/bin/copyDb.ts @@ -20,7 +20,13 @@ import { encodeBlobsWithFilePath, endPendingMigrationBlobSaves, } from '../resources/blob.ts'; -import { RecordEncoder, setNextEncoding, lastMetadata, METADATA } from '../resources/RecordEncoder.ts'; +import { + RecordEncoder, + setNextEncoding, + clearNextEncoding, + lastMetadata, + METADATA, +} from '../resources/RecordEncoder.ts'; export async function compactOnStart() { hdbLogger.notify('Running compact on start'); @@ -527,7 +533,29 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar copyStructures(sourceDbi, key); console.log('migrating', key, 'from', sourceDatabase, 'to RocksDB'); - await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); + const firstVersioned = await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); + if (firstVersioned !== undefined) { + // Invariant tripwire (#2012): a versioned record must round-trip with its 8-byte + // version prefix. The full big-endian float64 is compared to the source version — + // a first-byte check alone is ambiguous, since a prefix-less classic record can + // also lead with 0x42 (structure ref id 2). Failing here keeps the migration + // incomplete (LMDB source intact, migrateOnStart flag retained) instead of + // shipping a database whose records silently lost versions and record prototypes. + await firstVersioned.written; + await written; + const roundTrip = targetDbi.getBinarySync(firstVersioned.key); + const roundTripVersion = + roundTrip && roundTrip.length >= 8 + ? new DataView(roundTrip.buffer, roundTrip.byteOffset, roundTrip.byteLength).getFloat64(0) + : undefined; + if (roundTripVersion !== firstVersioned.version) { + throw new Error( + `Migration of ${sourceDatabase} wrote record ${JSON.stringify(firstVersioned.key)} without its ` + + `version/metadata prefix (expected version ${firstVersioned.version}, read back ${roundTripVersion}) — ` + + `records would lose versions and prototypes` + ); + } + } // Persist the canonical v5 classic structures the observer built, so every v5 runtime worker // adopts one agreed dictionary on startup instead of minting its own from an empty durable and @@ -598,6 +626,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar async function copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder?) { let recordsCopied = 0; let skippedRecord = 0; + let firstVersioned; const MAX_RETRIES = 1000; let retries = MAX_RETRIES; let start = null; @@ -626,18 +655,30 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // lastMetadata is set by RecordEncoder.decode for unpatched stores; // entry fields are set by handleLocalTimeForGets for patched stores const sourceMeta = lastMetadata; - setNextEncoding( - version, - entryMetadataFlags ?? sourceMeta?.[METADATA] ?? 0, - entryExpiresAt ?? sourceMeta?.expiresAt ?? -1, - entryNodeId ?? sourceMeta?.nodeId ?? -1, - entryResidencyId ?? sourceMeta?.residencyId ?? 0 - ); + if (version) { + setNextEncoding( + version, + entryMetadataFlags ?? sourceMeta?.[METADATA] ?? 0, + entryExpiresAt ?? sourceMeta?.expiresAt ?? -1, + entryNodeId ?? sourceMeta?.nodeId ?? -1, + entryResidencyId ?? sourceMeta?.residencyId ?? 0 + ); + } else { + // A flags-word-only prefix (no leading timestamp) is misparsed by the RocksDB + // decode heuristic (8 bytes consumed as a timestamp), so a version-less source + // record must be encoded plain; the read path repairs its prototype (#2012). + // Cleared even though the plain-encode hook would not consume stale globals: + // they may have leaked from a prior record whose encode was skipped or threw. + clearNextEncoding(); + } written = encodeBlobsWithFilePath( () => targetDbi.put(key, value, version), typeof key === 'number' ? key : recordsCopied, sourceRootStore ); + // Capture the put promise so the tripwire awaits THIS record's write, not + // an ordering assumption about later puts in the batch. + if (version) firstVersioned ??= { key, version, written }; // Feed only the record's SHAPE to the observer so it accumulates the canonical // classic structure (key list) for this shape; the encoded output is discarded. // A classic/named structure depends only on the keys, so we stub every leaf value @@ -694,7 +735,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar } } console.log('finish migrating, copied', recordsCopied, 'entries, skipped', skippedRecord, 'delete records'); - return; + return firstVersioned; } catch (err) { console.error( `Error iterating dbi for ${sourceDatabase} near key ${JSON.stringify(start)}, retrying (${retries} retries left):`, diff --git a/resources/PrimaryRocksDatabase.ts b/resources/PrimaryRocksDatabase.ts index 92caeb3ab..cd8b4c25a 100644 --- a/resources/PrimaryRocksDatabase.ts +++ b/resources/PrimaryRocksDatabase.ts @@ -72,15 +72,17 @@ export class PrimaryRocksDatabase extends RocksDatabase { #processEntry(raw: any, id: any): Entry | undefined { if (raw == null) return undefined; - if (raw[METADATA]) { + // Presence check, not truthiness: a [timestamp][no flags word] record decodes with + // metadataFlags === 0, and a falsy gate would hand the metadata wrapper itself back as + // the record value (harper#2012). + if (raw[METADATA] !== undefined) { raw.metadataFlags = raw[METADATA]; return this.#withEntry(raw, id); } - const entry = { value: raw, key: id } as Entry; - // map metadata-less values too, so getEntry guarantees the value→Entry - // mapping for every object value it returns (getSync/get rely on this) - if (typeof raw === 'object') entryMap.set(raw, entry); - return entry; + // Metadata-less values (e.g. records a broken migration stored without the prefix) still + // get the prototype repair and value→Entry mapping, matching getRange and the LMDB + // wrapper — only the version is unrecoverable (harper#2012). + return this.#withEntry({ value: raw } as Entry, id); } /** @@ -155,7 +157,7 @@ export class PrimaryRocksDatabase extends RocksDatabase { if (!this.#enc.isRocksDB) return iterable; const enc = this.#enc; return iterable.map((entry: any) => { - if (entry.value?.[METADATA]) { + if (entry.value?.[METADATA] !== undefined) { entry.metadataFlags = entry.value[METADATA]; Object.assign(entry, entry.value); } diff --git a/resources/RecordEncoder.ts b/resources/RecordEncoder.ts index f6aaacc7d..a229f83cb 100644 --- a/resources/RecordEncoder.ts +++ b/resources/RecordEncoder.ts @@ -171,7 +171,11 @@ export class RecordEncoder extends StructonEncoder { if (!options.randomAccessStructure) this._writeStruct = () => 0; const superEncode = this.encode; this.encode = function (record, options?) { - if (!this.useVersions) { + // Explicit opt-out only: `this` may be a foreign encoder this hook was grafted onto + // (copyDb patches the migration target's plain msgpackr encoder with it), where + // useVersions is undefined. Treating undefined as non-versioned silently stripped the + // metadata prefix from every LMDB→RocksDB migrated record (harper#2012). + if (this.useVersions === false) { // harper#1307: this store does not carry version metadata, so it never prefixes its records. // Encode plainly and LEAVE any in-flight *NextEncoding globals untouched for their real owner: // they belong to a versioned write (recordUpdater staged the primary's metadata and a nested diff --git a/unitTests/bin/migrationCanonicalStructures.test.js b/unitTests/bin/migrationCanonicalStructures.test.js index 75b2aa347..c1fad2f17 100644 --- a/unitTests/bin/migrationCanonicalStructures.test.js +++ b/unitTests/bin/migrationCanonicalStructures.test.js @@ -54,13 +54,23 @@ describe('migration: records still decode after the canonical-structures change // Open the migrated primary CF the same way the v5 runtime would and read records back. // With the canonical structures persisted, every migrated record (including the bare // structure-id references after the first of each shape) must decode, not null out. - const cf = RocksDatabase.open(targetPath, { name: 'CacheStruct/', sharedStructuresKey: Symbol.for('structures') }); + // PrimaryRocksDatabase + RecordEncoder are required since #2012: migrated records carry + // the version/metadata prefix again, which a plain msgpackr decoder cannot strip. + const { RecordEncoder } = require('#src/resources/RecordEncoder'); + const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); + const root = RocksDatabase.open(targetPath, {}); + const cf = new PrimaryRocksDatabase(targetPath, { + name: 'CacheStruct/', + sharedStructuresKey: Symbol.for('structures'), + encoder: { Encoder: RecordEncoder }, + }).open(); + cf.initStore(root); try { const failures = []; for (const id of ['a', 'b', 'c']) { let rec; try { - rec = cf.get(id); + rec = cf.getSync(id); } catch (e) { rec = { __threw: e.message }; } @@ -74,6 +84,7 @@ describe('migration: records still decode after the canonical-structures change ); } finally { cf.close(); + root.close(); } }); }); diff --git a/unitTests/bin/migrationMetadataPrefix.test.js b/unitTests/bin/migrationMetadataPrefix.test.js new file mode 100644 index 000000000..f87523255 --- /dev/null +++ b/unitTests/bin/migrationMetadataPrefix.test.js @@ -0,0 +1,84 @@ +// Regression guard for HarperFast/harper#2012: the LMDB→RocksDB migration must write each +// record with the [8-byte version][flags word] metadata prefix. The #1307 opt-out guard read +// `this.useVersions` off the migration target's plain msgpackr encoder (undefined), so every +// migrated record was stored prefix-less: no version, and point reads returned prototype-less +// plain objects (scans repaired the prototype, hiding it from search-based tests). +const fs = require('fs-extra'); +const assert = require('node:assert'); +const path = require('path'); +const { table } = require('#src/resources/databases'); +const { setMainIsWorker } = require('#js/server/threads/manageThreads'); +const { get: envGet } = require('#src/utility/environment/environmentManager'); +const { CONFIG_PARAMS } = require('#src/utility/hdbTerms'); + +describe('migration: records carry the version/metadata prefix (#2012)', function () { + if ((process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb') return; + const { setupTestDBPath } = require('../testUtils'); + const copyDB = require('#src/bin/copyDb'); + const { RocksDatabase } = require('@harperfast/rocksdb-js'); + const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); + const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); + + let rootPath, targetPath, Tbl; + const sourceVersions = new Map(); + + before(async function () { + rootPath = setupTestDBPath(); + setMainIsWorker(true); + Tbl = table({ + table: 'PrefixGuard', + database: 'pgtest', + attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'name' }, { name: 'rank' }], + }); + await Tbl.put({ id: 'a', name: 'alpha', rank: 1 }); + await Tbl.put({ id: 'b', name: 'beta', rank: 2 }); + await Tbl.put({ id: 'c', name: 'gamma', rank: 3 }); + for (const id of ['a', 'b', 'c']) { + sourceVersions.set(id, Tbl.primaryStore.getEntry(id).version); + } + + targetPath = path.join(rootPath, 'rocks-migrated-prefix', 'pgtest'); + await fs.remove(targetPath); + await copyDB.copyDbToRocks(Tbl.primaryStore.rootStore, 'pgtest', targetPath); + }); + + after(async function () { + await fs.remove(path.join(rootPath, 'rocks-migrated-prefix')); + }); + + it('migrated record bytes start with the 8-byte version prefix', function () { + const cf = RocksDatabase.open(targetPath, { name: 'PrefixGuard/', encoding: false }); + try { + for (const id of ['a', 'b', 'c']) { + const raw = cf.getBinarySync(id); + assert(raw, `record ${id} missing from migrated CF`); + // float64 of a current-era ms timestamp starts with 0x42 + assert.equal(raw[0], 0x42, `record ${id} first byte ${raw[0]} — metadata prefix missing`); + } + } finally { + cf.close(); + } + }); + + it('migrated records point-read with record prototype and source version via the runtime path', function () { + const root = RocksDatabase.open(targetPath, {}); + const cf = new PrimaryRocksDatabase(targetPath, { + name: 'PrefixGuard/', + encoder: { Encoder: RecordEncoder }, + sharedStructuresKey: Symbol.for('structures'), + }).open(); + cf.initStore(root); + try { + for (const id of ['a', 'b', 'c']) { + const entry = cf.getEntry(id); + assert(entry?.value, `record ${id} missing via getEntry`); + assert.equal(entry.value.name, { a: 'alpha', b: 'beta', c: 'gamma' }[id]); + assert(entry.value instanceof RecordObject, `record ${id} decoded without the record prototype`); + assert.equal(entry.version, sourceVersions.get(id), `record ${id} lost its source version`); + } + } finally { + cf.close(); + root.close(); + } + }); +}); diff --git a/unitTests/resources/primaryRocksMetadataRepair.test.js b/unitTests/resources/primaryRocksMetadataRepair.test.js new file mode 100644 index 000000000..d22e8fe2d --- /dev/null +++ b/unitTests/resources/primaryRocksMetadataRepair.test.js @@ -0,0 +1,71 @@ +// Regression guards for the PrimaryRocksDatabase read-path halves of HarperFast/harper#2012: +// 1. a [timestamp][no flags word] record (metadataFlags === 0) must not leak the decode +// wrapper as the record value — the gate on METADATA must be presence, not truthiness; +// 2. a metadata-less record (e.g. stored by the broken migration) must still get the +// structPrototype repair on point reads, matching getRange and the LMDB wrapper. +require('../testUtils'); +const assert = require('node:assert'); +const path = require('node:path'); +const fs = require('fs-extra'); +const { setupTestDBPath } = require('../testUtils'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); +const { Packr } = require('msgpackr'); +const { RecordEncoder, RecordObject, setNextEncoding } = require('#src/resources/RecordEncoder'); +const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); + +const isLMDB = process.env.HARPER_STORAGE_ENGINE === 'lmdb'; + +describe('PrimaryRocksDatabase metadata repair (#2012)', function () { + let dbPath, root, store, rawStore; + + before(function () { + if (isLMDB) return this.skip(); + dbPath = path.join(setupTestDBPath(), 'rocks-metadata-repair'); + fs.removeSync(dbPath); + root = RocksDatabase.open(dbPath, {}); + store = new PrimaryRocksDatabase(dbPath, { + name: 'RepairTest/', + encoder: { Encoder: RecordEncoder }, + }).open(); + store.initStore(root); + rawStore = RocksDatabase.open(dbPath, { name: 'RepairTest/', encoding: false }); + }); + + after(function () { + rawStore?.close(); + store?.close(); + root?.close(); + }); + + it('timestamp-only record (flags word absent) reads as a record, not the decode wrapper', async function () { + const version = Date.now() + 0.5; + setNextEncoding(version, -1, -1, -1, 0); + await store.put('noflags', { alpha: 1, beta: 2 }, version); + + const entry = store.getEntry('noflags'); + assert.equal(entry.version, version); + assert.deepEqual(Object.keys(entry.value), ['alpha', 'beta']); + assert.equal(entry.value.value, undefined, 'decode wrapper leaked as the record value'); + assert(entry.value instanceof RecordObject); + + for (const rangeEntry of store.getRange({ start: 'noflags', end: 'noflags~' })) { + assert.equal(rangeEntry.version, version); + assert.deepEqual(Object.keys(rangeEntry.value), ['alpha', 'beta']); + assert(rangeEntry.value instanceof RecordObject); + } + }); + + it('metadata-less record gets the prototype repair on point reads', async function () { + // plain msgpackr record-ext bytes with no prefix — the layout the broken migration produced + const packr = new Packr(); + await rawStore.put('legacy', packr.encode({ gamma: 3, delta: 4 })); + + const entry = store.getEntry('legacy'); + assert.deepEqual({ ...entry.value }, { gamma: 3, delta: 4 }); + assert(entry.value instanceof RecordObject, 'point read returned a prototype-less plain object'); + assert.equal(entry.version, undefined); + + const viaGet = store.getSync('legacy'); + assert(viaGet instanceof RecordObject); + }); +}); From 848b38504d2c8091c86845ca052a1eee93080307 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Thu, 30 Jul 2026 17:42:09 -0600 Subject: [PATCH 2/4] =?UTF-8?q?fix(migration):=20stage=20LMDB=E2=86=92Rock?= =?UTF-8?q?sDB=20migration=20and=20rename=20into=20place=20only=20after=20?= =?UTF-8?q?verification?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failure part-way through copyDbToRocks previously left a partial RocksDB at the database's real path; on the next boot both .mdb and / were discovered and whichever bound last won — a partial RocksDB that won was promoted (migrateOnStart saw 'already RocksDB', cleared the flag) while the intact LMDB was abandoned. The migration now writes into .migrating (excluded from database discovery), closes every target handle, and atomically renames into place only after the copy fully verifies; stale staging dirs are removed before retrying, so an interrupted migration (throw or SIGKILL) always recovers from the untouched LMDB source. Also: tripwire read-back now awaits the checked record's own put promise and uses readDoubleBE; strict assertions in the new tests (review notes). Co-Authored-By: Claude Fable 5 --- bin/copyDb.ts | 52 +++++++-- resources/databases.ts | 4 + .../bin/migrationCanonicalStructures.test.js | 2 +- unitTests/bin/migrationMetadataPrefix.test.js | 6 +- .../bin/migrationStagingRecovery.test.js | 103 ++++++++++++++++++ .../primaryRocksMetadataRepair.test.js | 14 +-- utility/hdbTerms.ts | 3 + 7 files changed, 166 insertions(+), 18 deletions(-) create mode 100644 unitTests/bin/migrationStagingRecovery.test.js diff --git a/bin/copyDb.ts b/bin/copyDb.ts index 53f2651bd..c5a91703a 100644 --- a/bin/copyDb.ts +++ b/bin/copyDb.ts @@ -3,11 +3,12 @@ import { open, asBinary } from 'lmdb'; import { join } from 'path'; import { move, remove } from 'fs-extra'; import { existsSync, mkdirSync } from 'node:fs'; +import { rename } from 'node:fs/promises'; import { get } from '../utility/environment/environmentManager.ts'; import OpenEnvironmentObject from '../utility/lmdb/OpenEnvironmentObject.ts'; import { OpenDBIObject } from '../utility/lmdb/OpenDBIObject.ts'; import { INTERNAL_DBIS_NAME, AUDIT_STORE_NAME } from '../utility/lmdb/terms.ts'; -import { CONFIG_PARAMS, DATABASES_DIR_NAME } from '../utility/hdbTerms.ts'; +import { CONFIG_PARAMS, DATABASES_DIR_NAME, MIGRATING_DIR_SUFFIX } from '../utility/hdbTerms.ts'; import { AUDIT_STORE_OPTIONS } from '../resources/auditStore.ts'; import { describeSchema } from '../dataLayer/schemaDescribe.ts'; import { updateConfigValue } from '../config/configUtils.ts'; @@ -385,7 +386,31 @@ export async function migrateOnStart() { console.log('Migrating', databaseName, 'from LMDB to RocksDB at', targetPath); - await copyDbToRocks(rootStore, databaseName, targetPath); + // Migrate into a staging directory and only rename it into place once the copy fully + // verifies. A failure part-way must never leave a partial RocksDB at targetPath: on the + // next boot both .mdb and / would be discovered, and whichever binds last wins — + // if the partial RocksDB wins, migrateOnStart sees "already RocksDB", clears the flag, + // and abandons the intact LMDB. Staging dirs are excluded from discovery and any stale + // one is removed here before retrying, so an interrupted migration (throw or SIGKILL) + // always recovers by re-migrating from the untouched LMDB source. + const stagingPath = targetPath + MIGRATING_DIR_SUFFIX; + await remove(stagingPath); + try { + await copyDbToRocks(rootStore, databaseName, stagingPath); + } catch (error) { + try { + await remove(stagingPath); + } catch { + // discovery ignores the staging dir and the next attempt removes it + } + throw error; + } + // A directory already at targetPath can only be a stale partial from a pre-staging + // build's failed attempt (or a complete copy from a crash after rename but before the + // flag cleared) — discovery bound the LMDB source this boot, and we just produced a + // fresh verified copy from it, so replace. + await remove(targetPath); + await rename(stagingPath, targetPath); // Back up the original LMDB file console.log('Backing up LMDB', databaseName, 'to', backupDest); @@ -423,6 +448,11 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar const sourceDbisDb = sourceRootStore.dbisDb; const targetRootStore = openRocksDb(targetPath, { disableWAL: false }); + // Every handle opened on targetPath. All must be closed before returning so the caller can + // atomically rename a staging directory into place — rocksdb-js registers descriptors by + // path string, so a handle left open on the staging path would hold the DB lock against the + // runtime's open of the renamed directory (harper#2012 restart-safety). + const targetHandles: RocksDatabase[] = [targetRootStore]; // sharedStructuresKey wires the rocksdb-js getStructures/saveStructures closures // so that the plain msgpackr.Encoder used here persists structures within the // __dbis__ CF at Symbol.for('structures'). The runtime attributesDbi RecordEncoder @@ -436,6 +466,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar name: INTERNAL_DBIS_NAME, sharedStructuresKey: Symbol.for('structures'), }); + targetHandles.push(targetDbisDb); const STRUCTURES_KEY = Symbol.for('structures'); const copyStructures = (sourceDbi, storeName: string, extraTarget?: RocksDatabase) => { @@ -490,8 +521,10 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar let canonicalStructures: any; if (!isPrimary) { targetDbi = openRocksDb(targetPath, { dupSort: true, name: key }); + targetHandles.push(targetDbi); } else { targetDbi = openRocksDb(targetPath, { name: key }); + targetHandles.push(targetDbi); // Patch the existing encoder (encoder is a getter-only property on RocksDatabase, cannot be replaced) // to install RecordEncoder's encode method so metadata headers (timestamps, HAS_BLOBS flag) are written const existingEncoder = targetDbi.encoder as any; @@ -544,10 +577,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar await firstVersioned.written; await written; const roundTrip = targetDbi.getBinarySync(firstVersioned.key); - const roundTripVersion = - roundTrip && roundTrip.length >= 8 - ? new DataView(roundTrip.buffer, roundTrip.byteOffset, roundTrip.byteLength).getFloat64(0) - : undefined; + const roundTripVersion = roundTrip && roundTrip.length >= 8 ? roundTrip.readDoubleBE(0) : undefined; if (roundTripVersion !== firstVersioned.version) { throw new Error( `Migration of ${sourceDatabase} wrote record ${JSON.stringify(firstVersioned.key)} without its ` + @@ -620,7 +650,15 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // Node's unhandledRejection. for (const saving of pendingBlobSaves) saving.catch(() => {}); transaction.done(); - targetRootStore.close(); + // Close every target handle (dbi CFs before the root) so no descriptor holds the DB lock + // on this path — required for the caller's staging-directory rename (harper#2012). + for (const handle of targetHandles.reverse()) { + try { + handle.close(); + } catch (error) { + console.error('Error closing migration target store', error); + } + } } async function copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder?) { diff --git a/resources/databases.ts b/resources/databases.ts index 0164dd5c1..3c5d47759 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -15,6 +15,7 @@ import { CONFIG_PARAMS, LEGACY_DATABASES_DIR_NAME, DATABASES_DIR_NAME, + MIGRATING_DIR_SUFFIX, RESERVED_DATABASE_NAMES, } from '../utility/hdbTerms.ts'; import { getConfigPath } from '../config/configUtils.ts'; @@ -315,6 +316,8 @@ export function getDatabases(): Databases { // First load all the databases from our main database folder // TODO: Load any databases defined with explicit storage paths from the config for (const databaseEntry of readdirSync(databasePath, { withFileTypes: true })) { + // in-progress migration staging dirs are not databases until atomically renamed into place + if (databaseEntry.name.endsWith(MIGRATING_DIR_SUFFIX)) continue; const dbName = basename(databaseEntry.name, '.mdb'); const dbPath = join(databasePath, databaseEntry.name); @@ -374,6 +377,7 @@ export function getDatabases(): Databases { const databasePath = schemaConfig.path; if (existsSync(databasePath)) { for (const databaseEntry of readdirSync(databasePath, { withFileTypes: true })) { + if (databaseEntry.name.endsWith(MIGRATING_DIR_SUFFIX)) continue; if (databaseEntry.isFile() && extname(databaseEntry.name).toLowerCase() === '.mdb') { readMetaDb(join(databasePath, databaseEntry.name), basename(databaseEntry.name, '.mdb'), dbName); } else { diff --git a/unitTests/bin/migrationCanonicalStructures.test.js b/unitTests/bin/migrationCanonicalStructures.test.js index c1fad2f17..f7741dc2f 100644 --- a/unitTests/bin/migrationCanonicalStructures.test.js +++ b/unitTests/bin/migrationCanonicalStructures.test.js @@ -77,7 +77,7 @@ describe('migration: records still decode after the canonical-structures change console.log(`record ${id}:`, JSON.stringify(rec)); if (!rec || rec.__threw || rec.content === undefined || rec.headers === undefined) failures.push(id); } - assert.equal( + assert.strictEqual( failures.length, 0, `migrated records ${failures.join(',')} did not decode after reopen (structures did not resolve)` diff --git a/unitTests/bin/migrationMetadataPrefix.test.js b/unitTests/bin/migrationMetadataPrefix.test.js index f87523255..1c8e903b5 100644 --- a/unitTests/bin/migrationMetadataPrefix.test.js +++ b/unitTests/bin/migrationMetadataPrefix.test.js @@ -53,7 +53,7 @@ describe('migration: records carry the version/metadata prefix (#2012)', functio const raw = cf.getBinarySync(id); assert(raw, `record ${id} missing from migrated CF`); // float64 of a current-era ms timestamp starts with 0x42 - assert.equal(raw[0], 0x42, `record ${id} first byte ${raw[0]} — metadata prefix missing`); + assert.strictEqual(raw[0], 0x42, `record ${id} first byte ${raw[0]} — metadata prefix missing`); } } finally { cf.close(); @@ -72,9 +72,9 @@ describe('migration: records carry the version/metadata prefix (#2012)', functio for (const id of ['a', 'b', 'c']) { const entry = cf.getEntry(id); assert(entry?.value, `record ${id} missing via getEntry`); - assert.equal(entry.value.name, { a: 'alpha', b: 'beta', c: 'gamma' }[id]); + assert.strictEqual(entry.value.name, { a: 'alpha', b: 'beta', c: 'gamma' }[id]); assert(entry.value instanceof RecordObject, `record ${id} decoded without the record prototype`); - assert.equal(entry.version, sourceVersions.get(id), `record ${id} lost its source version`); + assert.strictEqual(entry.version, sourceVersions.get(id), `record ${id} lost its source version`); } } finally { cf.close(); diff --git a/unitTests/bin/migrationStagingRecovery.test.js b/unitTests/bin/migrationStagingRecovery.test.js new file mode 100644 index 000000000..52595918b --- /dev/null +++ b/unitTests/bin/migrationStagingRecovery.test.js @@ -0,0 +1,103 @@ +// Restart-safety guards for the staged LMDB→RocksDB migration (harper#2012 review): +// a failed migration must never leave a partial RocksDB at the database's real path, because on +// the next boot both .mdb and / are discovered and whichever binds last wins — a partial +// RocksDB that wins gets promoted (flag cleared) while the intact LMDB is abandoned. The +// migration therefore stages into .migrating (excluded from discovery) and atomically +// renames after verification, replacing any stale artifact at the real path. +const fs = require('fs-extra'); +const assert = require('node:assert'); +const path = require('path'); +const { table } = require('#src/resources/databases'); +const { setMainIsWorker } = require('#js/server/threads/manageThreads'); +const { get: envGet } = require('#src/utility/environment/environmentManager'); +const { CONFIG_PARAMS, MIGRATING_DIR_SUFFIX } = require('#src/utility/hdbTerms'); + +describe('migration: staging directory recovery (#2012)', function () { + if ((process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb') return; + const { setupTestDBPath } = require('../testUtils'); + const copyDB = require('#src/bin/copyDb'); + const { RocksDatabase } = require('@harperfast/rocksdb-js'); + const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); + const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); + + let rootPath, baseDir, targetPath, stagingPath, Tbl; + + before(async function () { + rootPath = setupTestDBPath(); + setMainIsWorker(true); + Tbl = table({ + table: 'StagedRec', + database: 'stagetest', + attributes: [{ name: 'id', isPrimaryKey: true }, { name: 'name' }], + }); + await Tbl.put({ id: 'a', name: 'alpha' }); + await Tbl.put({ id: 'b', name: 'beta' }); + + baseDir = path.join(rootPath, 'rocks-staging-recovery'); + targetPath = path.join(baseDir, 'stagetest'); + stagingPath = targetPath + MIGRATING_DIR_SUFFIX; + await fs.remove(baseDir); + }); + + after(async function () { + await fs.remove(baseDir); + }); + + it('promotion replaces stale staging and a stale partial RocksDB at the target path', async function () { + // Simulate the "both artifacts present after an interrupted run" restart state: a junk + // leftover staging dir AND a partial (unreadable) RocksDB already at the real path. + await fs.outputFile(path.join(stagingPath, 'junk'), 'stale staging from a failed attempt'); + await fs.outputFile(path.join(targetPath, 'CURRENT'), 'MANIFEST-000001\n'); + await fs.outputFile(path.join(targetPath, 'MANIFEST-000001'), 'partial'); + + // The promotion sequence migrateOnStart runs per database. + await fs.remove(stagingPath); + await copyDB.copyDbToRocks(Tbl.primaryStore.rootStore, 'stagetest', stagingPath); + await fs.remove(targetPath); + await fs.rename(stagingPath, targetPath); + + assert.strictEqual(fs.existsSync(stagingPath), false, 'staging dir must not survive promotion'); + + // Opening the renamed directory also guards the close-all-handles contract: a handle + // leaked on the staging path would still hold the RocksDB LOCK for this store. + const root = RocksDatabase.open(targetPath, {}); + const cf = new PrimaryRocksDatabase(targetPath, { + name: 'StagedRec/', + encoder: { Encoder: RecordEncoder }, + sharedStructuresKey: Symbol.for('structures'), + }).open(); + cf.initStore(root); + try { + for (const id of ['a', 'b']) { + const entry = cf.getEntry(id); + assert(entry?.value, `record ${id} missing after promotion`); + assert(entry.value instanceof RecordObject, `record ${id} lost its record prototype`); + assert(entry.version > 0, `record ${id} lost its version`); + } + } finally { + cf.close(); + root.close(); + } + }); + + it('database discovery ignores *.migrating staging directories', async function () { + const { getDatabases, resetDatabases } = require('#src/resources/databases'); + const databasesDir = path.join(rootPath, 'database'); + const ghostDir = path.join(databasesDir, 'ghost' + MIGRATING_DIR_SUFFIX); + // Rocks markers that would otherwise make discovery open it as a database + await fs.outputFile(path.join(ghostDir, 'CURRENT'), 'MANIFEST-000001\n'); + await fs.outputFile(path.join(ghostDir, 'MANIFEST-000001'), 'partial'); + try { + resetDatabases(); + const databases = getDatabases(); + assert.strictEqual( + databases['ghost' + MIGRATING_DIR_SUFFIX], + undefined, + 'staging dir must not be discovered as a database' + ); + } finally { + await fs.remove(ghostDir); + resetDatabases(); + } + }); +}); diff --git a/unitTests/resources/primaryRocksMetadataRepair.test.js b/unitTests/resources/primaryRocksMetadataRepair.test.js index d22e8fe2d..13d64b65d 100644 --- a/unitTests/resources/primaryRocksMetadataRepair.test.js +++ b/unitTests/resources/primaryRocksMetadataRepair.test.js @@ -43,14 +43,14 @@ describe('PrimaryRocksDatabase metadata repair (#2012)', function () { await store.put('noflags', { alpha: 1, beta: 2 }, version); const entry = store.getEntry('noflags'); - assert.equal(entry.version, version); - assert.deepEqual(Object.keys(entry.value), ['alpha', 'beta']); - assert.equal(entry.value.value, undefined, 'decode wrapper leaked as the record value'); + assert.strictEqual(entry.version, version); + assert.deepStrictEqual(Object.keys(entry.value), ['alpha', 'beta']); + assert.strictEqual(entry.value.value, undefined, 'decode wrapper leaked as the record value'); assert(entry.value instanceof RecordObject); for (const rangeEntry of store.getRange({ start: 'noflags', end: 'noflags~' })) { - assert.equal(rangeEntry.version, version); - assert.deepEqual(Object.keys(rangeEntry.value), ['alpha', 'beta']); + assert.strictEqual(rangeEntry.version, version); + assert.deepStrictEqual(Object.keys(rangeEntry.value), ['alpha', 'beta']); assert(rangeEntry.value instanceof RecordObject); } }); @@ -61,9 +61,9 @@ describe('PrimaryRocksDatabase metadata repair (#2012)', function () { await rawStore.put('legacy', packr.encode({ gamma: 3, delta: 4 })); const entry = store.getEntry('legacy'); - assert.deepEqual({ ...entry.value }, { gamma: 3, delta: 4 }); + assert.deepStrictEqual({ ...entry.value }, { gamma: 3, delta: 4 }); assert(entry.value instanceof RecordObject, 'point read returned a prototype-less plain object'); - assert.equal(entry.version, undefined); + assert.strictEqual(entry.version, undefined); const viaGet = store.getSync('legacy'); assert(viaGet instanceof RecordObject); diff --git a/utility/hdbTerms.ts b/utility/hdbTerms.ts index 95681c7ef..67fba203c 100644 --- a/utility/hdbTerms.ts +++ b/utility/hdbTerms.ts @@ -157,6 +157,9 @@ export const HDB_FILE_PERMISSIONS = 0o700; /** Database directory */ export const DATABASES_DIR_NAME = 'database'; +/** Suffix for in-progress LMDB→RocksDB migration staging directories: excluded from database + * discovery, atomically renamed into place only after the migration fully verifies (harper#2012) */ +export const MIGRATING_DIR_SUFFIX = '.migrating'; /** Legacy Database directory */ export const LEGACY_DATABASES_DIR_NAME = 'schema'; /** Transaction directory */ From 496a756827d54f8428e8597b81f676f562b35cc6 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Thu, 30 Jul 2026 17:54:32 -0600 Subject: [PATCH 3/4] fix(migration): full prefix-verification sweep, verifyMigratedDatabase export, visible test gates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - every primary dbi is raw-scanned after copy: unversioned records must exactly match the (counted) version-less source records, else the migration fails closed — full delivery of the #2012 verification ask - verifyMigratedDatabase(databasePath) reports per-table unversioned counts for instances migrated before the fix (pre-prod check / dev CM) - LMDB-gated migration tests now this.skip() visibly (pending) instead of a bare describe-body return that registers nothing; failure-path test covers migrateDatabaseToRocks staging cleanup via an injected error - copyDbToRocks acquires its read transaction inside try so an early throw still closes all staging handles (found by the failure-path test) Co-Authored-By: Claude Fable 5 --- bin/copyDb.ts | 152 ++++++++++++++---- .../bin/migrationCanonicalStructures.test.js | 15 +- unitTests/bin/migrationMetadataPrefix.test.js | 19 ++- .../bin/migrationStagingRecovery.test.js | 49 ++++-- 4 files changed, 175 insertions(+), 60 deletions(-) diff --git a/bin/copyDb.ts b/bin/copyDb.ts index c5a91703a..e967fb4e1 100644 --- a/bin/copyDb.ts +++ b/bin/copyDb.ts @@ -337,7 +337,7 @@ function openRocksDb(path: string, options: RocksDatabaseOptions & { dupSort?: b db = new (RocksIndexStore as any)(path, options).open(); } else { db = RocksDatabase.open(path, options); - db.encoder.name = options.name; + if (db.encoder) db.encoder.name = options.name; } return db; } @@ -386,31 +386,7 @@ export async function migrateOnStart() { console.log('Migrating', databaseName, 'from LMDB to RocksDB at', targetPath); - // Migrate into a staging directory and only rename it into place once the copy fully - // verifies. A failure part-way must never leave a partial RocksDB at targetPath: on the - // next boot both .mdb and / would be discovered, and whichever binds last wins — - // if the partial RocksDB wins, migrateOnStart sees "already RocksDB", clears the flag, - // and abandons the intact LMDB. Staging dirs are excluded from discovery and any stale - // one is removed here before retrying, so an interrupted migration (throw or SIGKILL) - // always recovers by re-migrating from the untouched LMDB source. - const stagingPath = targetPath + MIGRATING_DIR_SUFFIX; - await remove(stagingPath); - try { - await copyDbToRocks(rootStore, databaseName, stagingPath); - } catch (error) { - try { - await remove(stagingPath); - } catch { - // discovery ignores the staging dir and the next attempt removes it - } - throw error; - } - // A directory already at targetPath can only be a stale partial from a pre-staging - // build's failed attempt (or a complete copy from a crash after rename but before the - // flag cleared) — discovery bound the LMDB source this boot, and we just produced a - // fresh verified copy from it, so replace. - await remove(targetPath); - await rename(stagingPath, targetPath); + await migrateDatabaseToRocks(rootStore, databaseName, targetPath); // Back up the original LMDB file console.log('Backing up LMDB', databaseName, 'to', backupDest); @@ -443,6 +419,92 @@ export async function migrateOnStart() { } } +/** + * Count records in a raw (encoding: false) store whose value lacks the 8-byte version prefix + * (0x42 = first byte of a ms-epoch float64). Symbol keys are internal and skipped. + */ +function countRecords(rawDbi): { records: number; unversioned: number } { + let records = 0; + let unversioned = 0; + for (const { key, value } of rawDbi.getRange({})) { + if (typeof key === 'symbol') continue; + records++; + if (!value || value.length < 8 || value[0] !== 0x42) unversioned++; + } + return { records, unversioned }; +} + +/** + * Verification sweep for an already-migrated RocksDB database (harper#2012): reports, per + * primary-key dbi, how many records lack the version/metadata prefix. Such records decode + * without their record prototype on point reads and carry no version; a nonzero count beyond a + * table's known version-less records means it needs the no-op rewrite pass. Read-only, but takes + * the RocksDB lock — run in-process (inspector) on a live instance, or offline. + */ +export function verifyMigratedDatabase(databasePath: string): Record { + const rootStore = RocksDatabase.open(databasePath, {}); + const dbisDb = RocksDatabase.open(databasePath, { + name: INTERNAL_DBIS_NAME, + sharedStructuresKey: Symbol.for('structures'), + }); + const report: Record = {}; + try { + for (const { key, value: attribute } of dbisDb.getRange({})) { + if (typeof key === 'symbol' || !attribute?.isPrimaryKey) continue; + const rawDbi = RocksDatabase.open(databasePath, { name: key, encoding: false }); + try { + report[key] = countRecords(rawDbi); + } finally { + rawDbi.close(); + } + } + } finally { + dbisDb.close(); + rootStore.close(); + } + return report; +} + +/** + * Migrate one database LMDB→RocksDB with restart-safe promotion: copy into a staging directory + * (excluded from database discovery) and atomically rename it to targetPath only after the copy + * fully verifies. A failure part-way must never leave a partial RocksDB at targetPath — on the + * next boot both .mdb and / would be discovered, and whichever binds last wins; if the + * partial RocksDB won, migrateOnStart would see "already RocksDB", clear the flag, and abandon + * the intact LMDB. Any stale staging dir is removed before retrying, so an interrupted migration + * (throw or SIGKILL) always recovers by re-migrating from the untouched LMDB source. + */ +export async function migrateDatabaseToRocks(sourceRootStore, databaseName: string, targetPath: string) { + const stagingPath = targetPath + MIGRATING_DIR_SUFFIX; + await remove(stagingPath); + try { + await copyDbToRocks(sourceRootStore, databaseName, stagingPath); + } catch (error) { + try { + await remove(stagingPath); + } catch { + // discovery ignores the staging dir and the next attempt removes it + } + throw error; + } + // A directory already at targetPath can only be a stale partial from a pre-staging build's + // failed attempt (or a complete copy from a crash after rename but before the flag cleared) — + // discovery bound the LMDB source this boot, and we just produced a fresh verified copy from + // it, so replace. + await remove(targetPath); + // On Windows a directory rename can transiently fail while RocksDB background threads release + // their last file handles after close(); retry briefly before giving up. + for (let attempt = 0; ; attempt++) { + try { + await rename(stagingPath, targetPath); + break; + } catch (error: any) { + if (attempt >= 4 || !(error.code === 'EPERM' || error.code === 'EBUSY' || error.code === 'EACCES')) throw error; + await new Promise((resolve) => setTimeout(resolve, 250)); + } + } +} + export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, targetPath: string) { console.log(`Migrating database ${sourceDatabase} to RocksDB at ${targetPath}`); const sourceDbisDb = sourceRootStore.dbisDb; @@ -491,8 +553,11 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // referencing fileIds whose files were never durably written — exactly the missing-blob-file // state that triggers the base-copy resync wedge in harper#1337. const pendingBlobSaves = beginPendingMigrationBlobSaves(); - const transaction = sourceDbisDb.useReadTransaction(); + // Acquired inside the try: if it throws, the finally must still close the target handles so a + // staging directory can be removed and its path reopened cleanly (harper#2012). + let transaction; try { + transaction = sourceDbisDb.useReadTransaction(); for (const { key, value: attribute } of sourceDbisDb.getRange({ transaction })) { const isPrimary = attribute.isPrimaryKey; targetDbisDb.put(key, attribute); @@ -566,8 +631,9 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar copyStructures(sourceDbi, key); console.log('migrating', key, 'from', sourceDatabase, 'to RocksDB'); - const firstVersioned = await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); - if (firstVersioned !== undefined) { + const copied = await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); + if (copied?.firstVersioned !== undefined) { + const { firstVersioned, versionlessCount } = copied; // Invariant tripwire (#2012): a versioned record must round-trip with its 8-byte // version prefix. The full big-endian float64 is compared to the source version — // a first-byte check alone is ambiguous, since a prefix-less classic record can @@ -585,6 +651,20 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar `records would lose versions and prototypes` ); } + // Full verification sweep (harper#2012 ask #3): every migrated record must carry the + // version prefix except the (counted) version-less source records that were + // deliberately encoded plain. A raw first-byte scan needs no decode, so it is a + // cheap sequential pass over the freshly written CF. + const rawDbi = openRocksDb(targetPath, { name: key, encoding: false }); + targetHandles.push(rawDbi); + const { unversioned } = countRecords(rawDbi); + if (unversioned !== versionlessCount) { + throw new Error( + `Migration of ${sourceDatabase}/${key} left ${unversioned} record(s) without the version/metadata ` + + `prefix (expected ${versionlessCount} version-less source records) — records would lose ` + + `versions and prototypes` + ); + } } // Persist the canonical v5 classic structures the observer built, so every v5 runtime worker @@ -649,7 +729,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // later background failure is silently observed instead of crashing the process via // Node's unhandledRejection. for (const saving of pendingBlobSaves) saving.catch(() => {}); - transaction.done(); + transaction?.done(); // Close every target handle (dbi CFs before the root) so no descriptor holds the DB lock // on this path — required for the caller's staging-directory rename (harper#2012). for (const handle of targetHandles.reverse()) { @@ -665,6 +745,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar let recordsCopied = 0; let skippedRecord = 0; let firstVersioned; + let versionlessCount = 0; const MAX_RETRIES = 1000; let retries = MAX_RETRIES; let start = null; @@ -708,6 +789,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // Cleared even though the plain-encode hook would not consume stale globals: // they may have leaked from a prior record whose encode was skipped or threw. clearNextEncoding(); + versionlessCount++; } written = encodeBlobsWithFilePath( () => targetDbi.put(key, value, version), @@ -773,7 +855,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar } } console.log('finish migrating, copied', recordsCopied, 'entries, skipped', skippedRecord, 'delete records'); - return firstVersioned; + return { firstVersioned, versionlessCount }; } catch (err) { console.error( `Error iterating dbi for ${sourceDatabase} near key ${JSON.stringify(start)}, retrying (${retries} retries left):`, @@ -781,11 +863,15 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar ); if (typeof start === 'string') { if (start === 'z') { - return console.error('Reached end of dbi', start, 'for', sourceDatabase); + console.error('Reached end of dbi', start, 'for', sourceDatabase); + return; } start = start.slice(0, -2) + 'z'; } else if (typeof start === 'number') start++; - else return console.error('Unknown key type', start, 'for', sourceDatabase); + else { + console.error('Unknown key type', start, 'for', sourceDatabase); + return; + } } } // Fail loudly so migrateOnStart's try/catch preserves the migrateOnStart flag and diff --git a/unitTests/bin/migrationCanonicalStructures.test.js b/unitTests/bin/migrationCanonicalStructures.test.js index f7741dc2f..8c3b60189 100644 --- a/unitTests/bin/migrationCanonicalStructures.test.js +++ b/unitTests/bin/migrationCanonicalStructures.test.js @@ -20,15 +20,18 @@ const { CONFIG_PARAMS } = require('#src/utility/hdbTerms'); // concern: it requires the runtime's multi-column-family open + per-worker encoder wiring, which this // single-handle unit harness can't replicate. The observer's captured dictionary is verified in-process // during the migration; cross-process adoption is validated by the cluster repro. -describe('migration: records still decode after the canonical-structures change (#1453)', function () { - if ((process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb') return; - const { setupTestDBPath } = require('../testUtils'); - const copyDB = require('#src/bin/copyDb'); - const { RocksDatabase } = require('@harperfast/rocksdb-js'); +const { setupTestDBPath } = require('../testUtils'); +const copyDB = require('#src/bin/copyDb'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); + +const isLMDB = (process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) === 'lmdb'; +describe('migration: records still decode after the canonical-structures change (#1453)', function () { let rootPath, targetPath, Tbl; before(async function () { + // this.skip() (not a bare return in the describe body) so the gate is visible as pending + if (!isLMDB) return this.skip(); rootPath = setupTestDBPath(); setMainIsWorker(true); Tbl = table({ @@ -47,7 +50,7 @@ describe('migration: records still decode after the canonical-structures change }); after(async function () { - await fs.remove(path.join(rootPath, 'rocks-migrated-cstruct')); + if (rootPath) await fs.remove(path.join(rootPath, 'rocks-migrated-cstruct')); }); it('migrated records decode after reopen (structures resolve)', function () { diff --git a/unitTests/bin/migrationMetadataPrefix.test.js b/unitTests/bin/migrationMetadataPrefix.test.js index 1c8e903b5..31349ae29 100644 --- a/unitTests/bin/migrationMetadataPrefix.test.js +++ b/unitTests/bin/migrationMetadataPrefix.test.js @@ -11,18 +11,21 @@ const { setMainIsWorker } = require('#js/server/threads/manageThreads'); const { get: envGet } = require('#src/utility/environment/environmentManager'); const { CONFIG_PARAMS } = require('#src/utility/hdbTerms'); -describe('migration: records carry the version/metadata prefix (#2012)', function () { - if ((process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb') return; - const { setupTestDBPath } = require('../testUtils'); - const copyDB = require('#src/bin/copyDb'); - const { RocksDatabase } = require('@harperfast/rocksdb-js'); - const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); - const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); +const { setupTestDBPath } = require('../testUtils'); +const copyDB = require('#src/bin/copyDb'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); +const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); +const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); + +const isLMDB = (process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) === 'lmdb'; +describe('migration: records carry the version/metadata prefix (#2012)', function () { let rootPath, targetPath, Tbl; const sourceVersions = new Map(); before(async function () { + // this.skip() (not a bare return in the describe body) so the gate is visible as pending + if (!isLMDB) return this.skip(); rootPath = setupTestDBPath(); setMainIsWorker(true); Tbl = table({ @@ -43,7 +46,7 @@ describe('migration: records carry the version/metadata prefix (#2012)', functio }); after(async function () { - await fs.remove(path.join(rootPath, 'rocks-migrated-prefix')); + if (rootPath) await fs.remove(path.join(rootPath, 'rocks-migrated-prefix')); }); it('migrated record bytes start with the 8-byte version prefix', function () { diff --git a/unitTests/bin/migrationStagingRecovery.test.js b/unitTests/bin/migrationStagingRecovery.test.js index 52595918b..600a29d43 100644 --- a/unitTests/bin/migrationStagingRecovery.test.js +++ b/unitTests/bin/migrationStagingRecovery.test.js @@ -11,18 +11,20 @@ const { table } = require('#src/resources/databases'); const { setMainIsWorker } = require('#js/server/threads/manageThreads'); const { get: envGet } = require('#src/utility/environment/environmentManager'); const { CONFIG_PARAMS, MIGRATING_DIR_SUFFIX } = require('#src/utility/hdbTerms'); +const { setupTestDBPath } = require('../testUtils'); +const copyDB = require('#src/bin/copyDb'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); +const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); +const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); -describe('migration: staging directory recovery (#2012)', function () { - if ((process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb') return; - const { setupTestDBPath } = require('../testUtils'); - const copyDB = require('#src/bin/copyDb'); - const { RocksDatabase } = require('@harperfast/rocksdb-js'); - const { RecordEncoder, RecordObject } = require('#src/resources/RecordEncoder'); - const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); +const isLMDB = (process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) === 'lmdb'; +describe('migration: staging directory recovery (#2012)', function () { let rootPath, baseDir, targetPath, stagingPath, Tbl; before(async function () { + // this.skip() (not a bare return in the describe body) so the gate is visible as pending + if (!isLMDB) return this.skip(); rootPath = setupTestDBPath(); setMainIsWorker(true); Tbl = table({ @@ -40,7 +42,27 @@ describe('migration: staging directory recovery (#2012)', function () { }); after(async function () { - await fs.remove(baseDir); + if (baseDir) await fs.remove(baseDir); + }); + + it('a failed migration removes its staging dir and leaves no artifact at the target path', async function () { + // A source whose dbi iteration throws immediately — the staging dir is created before the + // copy loop starts, so the catch path must clean it up and leave targetPath absent. + const brokenSource = { + dbisDb: { + useReadTransaction() { + throw new Error('injected migration failure'); + }, + }, + }; + await assert.rejects( + () => copyDB.migrateDatabaseToRocks(brokenSource, 'stagetest', targetPath), + /injected migration failure/ + ); + assert.strictEqual(fs.existsSync(stagingPath), false, 'staging dir must be removed on failure'); + assert.strictEqual(fs.existsSync(targetPath), false, 'no artifact may appear at the target path on failure'); + // the LMDB source is untouched and still serves reads + assert.strictEqual((await Tbl.get('a')).name, 'alpha'); }); it('promotion replaces stale staging and a stale partial RocksDB at the target path', async function () { @@ -50,11 +72,7 @@ describe('migration: staging directory recovery (#2012)', function () { await fs.outputFile(path.join(targetPath, 'CURRENT'), 'MANIFEST-000001\n'); await fs.outputFile(path.join(targetPath, 'MANIFEST-000001'), 'partial'); - // The promotion sequence migrateOnStart runs per database. - await fs.remove(stagingPath); - await copyDB.copyDbToRocks(Tbl.primaryStore.rootStore, 'stagetest', stagingPath); - await fs.remove(targetPath); - await fs.rename(stagingPath, targetPath); + await copyDB.migrateDatabaseToRocks(Tbl.primaryStore.rootStore, 'stagetest', targetPath); assert.strictEqual(fs.existsSync(stagingPath), false, 'staging dir must not survive promotion'); @@ -80,6 +98,11 @@ describe('migration: staging directory recovery (#2012)', function () { } }); + it('verifyMigratedDatabase reports zero unversioned records for a clean migration', function () { + const report = copyDB.verifyMigratedDatabase(targetPath); + assert.deepStrictEqual(report['StagedRec/'], { records: 2, unversioned: 0 }); + }); + it('database discovery ignores *.migrating staging directories', async function () { const { getDatabases, resetDatabases } = require('#src/resources/databases'); const databasesDir = path.join(rootPath, 'database'); From 6ce7d1fb6ff2437bd7f6c70ae38aa70d60bd8760 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Thu, 30 Jul 2026 18:58:46 -0600 Subject: [PATCH 4/4] fix(migration): exempt version-less records from the sweep by key, not byte sniffing A version-less source record whose plain classic body leads with 0x42 (structure ref id 2) was counted as versioned by the sweep's first-byte heuristic, making unversioned < versionlessCount and throwing a spurious fail-closed invariant error. Track the version-less keys and exempt them exactly; versioned records keep the first-byte check (systemic regressions trip on the first record, and the tripwire validates one exact header). Co-Authored-By: Claude Fable 5 --- bin/copyDb.ts | 40 ++++++++++++++++++++++++++-------------- 1 file changed, 26 insertions(+), 14 deletions(-) diff --git a/bin/copyDb.ts b/bin/copyDb.ts index e967fb4e1..ed701ddbe 100644 --- a/bin/copyDb.ts +++ b/bin/copyDb.ts @@ -421,7 +421,9 @@ export async function migrateOnStart() { /** * Count records in a raw (encoding: false) store whose value lacks the 8-byte version prefix - * (0x42 = first byte of a ms-epoch float64). Symbol keys are internal and skipped. + * (0x42 = first byte of a ms-epoch float64). Symbol keys are internal and skipped. First-byte + * classification is a heuristic: a prefix-less classic record can also lead with 0x42 (structure + * ref id 2) and would be undercounted — fine for the whole-table detection this report serves. */ function countRecords(rawDbi): { records: number; unversioned: number } { let records = 0; @@ -633,7 +635,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar console.log('migrating', key, 'from', sourceDatabase, 'to RocksDB'); const copied = await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); if (copied?.firstVersioned !== undefined) { - const { firstVersioned, versionlessCount } = copied; + const { firstVersioned, versionlessKeys } = copied; // Invariant tripwire (#2012): a versioned record must round-trip with its 8-byte // version prefix. The full big-endian float64 is compared to the source version — // a first-byte check alone is ambiguous, since a prefix-less classic record can @@ -651,18 +653,26 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar `records would lose versions and prototypes` ); } - // Full verification sweep (harper#2012 ask #3): every migrated record must carry the - // version prefix except the (counted) version-less source records that were - // deliberately encoded plain. A raw first-byte scan needs no decode, so it is a - // cheap sequential pass over the freshly written CF. + // Full verification sweep (harper#2012 ask #3): every migrated record that had a source + // version must carry the prefix. Version-less source records are exempted by KEY, not + // by byte inspection — a plain classic body can also lead with 0x42 (structure ref + // id 2), so byte sniffing could misclassify them and throw spuriously. For versioned + // records the first-byte check suffices: a systemic prefix regression trips on the + // very first record (and the exact-header tripwire above already validated one). const rawDbi = openRocksDb(targetPath, { name: key, encoding: false }); targetHandles.push(rawDbi); - const { unversioned } = countRecords(rawDbi); - if (unversioned !== versionlessCount) { + let unprefixed = 0; + let scanned = 0; + for (const { key: recordKey, value } of rawDbi.getRange({})) { + if (typeof recordKey === 'symbol') continue; + if (versionlessKeys.has(typeof recordKey === 'object' ? JSON.stringify(recordKey) : recordKey)) continue; + scanned++; + if (!value || value.length < 8 || value[0] !== 0x42) unprefixed++; + } + if (unprefixed > 0) { throw new Error( - `Migration of ${sourceDatabase}/${key} left ${unversioned} record(s) without the version/metadata ` + - `prefix (expected ${versionlessCount} version-less source records) — records would lose ` + - `versions and prototypes` + `Migration of ${sourceDatabase}/${key} left ${unprefixed} of ${scanned} versioned record(s) without ` + + `the version/metadata prefix — records would lose versions and prototypes` ); } } @@ -745,7 +755,9 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar let recordsCopied = 0; let skippedRecord = 0; let firstVersioned; - let versionlessCount = 0; + // keys (normalized for compound keys) of source records with no version, deliberately + // encoded plain — the verification sweep exempts them by key, never by byte sniffing + const versionlessKeys = new Set(); const MAX_RETRIES = 1000; let retries = MAX_RETRIES; let start = null; @@ -789,7 +801,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar // Cleared even though the plain-encode hook would not consume stale globals: // they may have leaked from a prior record whose encode was skipped or threw. clearNextEncoding(); - versionlessCount++; + versionlessKeys.add(typeof key === 'object' ? JSON.stringify(key) : key); } written = encodeBlobsWithFilePath( () => targetDbi.put(key, value, version), @@ -855,7 +867,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar } } console.log('finish migrating, copied', recordsCopied, 'entries, skipped', skippedRecord, 'delete records'); - return { firstVersioned, versionlessCount }; + return { firstVersioned, versionlessKeys }; } catch (err) { console.error( `Error iterating dbi for ${sourceDatabase} near key ${JSON.stringify(start)}, retrying (${retries} retries left):`,