Write the version/metadata prefix on LMDB→RocksDB migrated records so they keep versions and record prototypes

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

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

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 {

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();
 		}
 	});
 });

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();
+		}
+	});
+});

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 <db>.mdb and <db>/ 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 <db>.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();
+		}
+	});
+});