diff --git a/bin/copyDb.ts b/bin/copyDb.ts index 4d8fa640c..ed701ddbe 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'; @@ -20,7 +21,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'); @@ -330,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; } @@ -379,7 +386,7 @@ export async function migrateOnStart() { console.log('Migrating', databaseName, 'from LMDB to RocksDB at', targetPath); - await copyDbToRocks(rootStore, databaseName, targetPath); + await migrateDatabaseToRocks(rootStore, databaseName, targetPath); // Back up the original LMDB file console.log('Backing up LMDB', databaseName, 'to', backupDest); @@ -412,11 +419,104 @@ 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. 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; + 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; 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 @@ -430,6 +530,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) => { @@ -454,8 +555,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); @@ -484,8 +588,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; @@ -527,7 +633,49 @@ 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 copied = await copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder); + if (copied?.firstVersioned !== undefined) { + 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 + // 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 ? roundTrip.readDoubleBE(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` + ); + } + // 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); + 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 ${unprefixed} of ${scanned} versioned record(s) without ` + + `the version/metadata prefix — 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 @@ -591,13 +739,25 @@ 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(); - targetRootStore.close(); + 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()) { + try { + handle.close(); + } catch (error) { + console.error('Error closing migration target store', error); + } + } } async function copyDbiToRocks(sourceDbi, targetDbi, isPrimary, transaction, observerEncoder?) { let recordsCopied = 0; let skippedRecord = 0; + let firstVersioned; + // 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; @@ -626,18 +786,31 @@ 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(); + versionlessKeys.add(typeof key === 'object' ? JSON.stringify(key) : key); + } 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 +867,7 @@ export async function copyDbToRocks(sourceRootStore, sourceDatabase: string, tar } } console.log('finish migrating, copied', recordsCopied, 'entries, skipped', skippedRecord, 'delete records'); - return; + return { firstVersioned, versionlessKeys }; } catch (err) { console.error( `Error iterating dbi for ${sourceDatabase} near key ${JSON.stringify(start)}, retrying (${retries} retries left):`, @@ -702,11 +875,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/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/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 75b2aa347..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,33 +50,44 @@ 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 () { // 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 }; } 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)` ); } 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..31349ae29 --- /dev/null +++ b/unitTests/bin/migrationMetadataPrefix.test.js @@ -0,0 +1,87 @@ +// 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'); + +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({ + 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 () { + if (rootPath) 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.strictEqual(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.strictEqual(entry.value.name, { a: 'alpha', b: 'beta', c: 'gamma' }[id]); + assert(entry.value instanceof RecordObject, `record ${id} decoded without the record prototype`); + assert.strictEqual(entry.version, sourceVersions.get(id), `record ${id} lost its source version`); + } + } finally { + cf.close(); + root.close(); + } + }); +}); diff --git a/unitTests/bin/migrationStagingRecovery.test.js b/unitTests/bin/migrationStagingRecovery.test.js new file mode 100644 index 000000000..600a29d43 --- /dev/null +++ b/unitTests/bin/migrationStagingRecovery.test.js @@ -0,0 +1,126 @@ +// 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'); +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({ + 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 () { + 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 () { + // 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'); + + await copyDB.migrateDatabaseToRocks(Tbl.primaryStore.rootStore, 'stagetest', 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('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'); + 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 new file mode 100644 index 000000000..13d64b65d --- /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.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.strictEqual(rangeEntry.version, version); + assert.deepStrictEqual(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.deepStrictEqual({ ...entry.value }, { gamma: 3, delta: 4 }); + assert(entry.value instanceof RecordObject, 'point read returned a prototype-less plain object'); + 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 */