From 825aabd7cd4acbf97734a13366e3279c55cdbcf6 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 14 Aug 2026 09:20:53 -0600 Subject: [PATCH 1/3] fix(storage): version raw primary writes --- DESIGN.md | 6 ++ resources/PrimaryRocksDatabase.ts | 21 +++- resources/RecordEncoder.ts | 13 +++ resources/blob.ts | 3 + .../primaryRocksMetadataRepair.test.js | 99 +++++++++++++++++++ 5 files changed, 139 insertions(+), 3 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 8b8a19dcbd..5c24a14d08 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -24,6 +24,12 @@ To match v4: `OpenDBIObject` sets `randomAccessStructure = isPrimary`, and `Reco - The struct **read** hook is left intact, so records a prior v5 already wrote in struct mode still decode (read-compat); only new writes switch to records mode. - Companion change in `structon`: `prepareStructures` saves the shared structures in the legacy plain-array form when there are no typed structs (instead of the `{named, typed}` Map), so the `Symbol.for('structures')` buffer for a records-mode `__dbis__` is also v4-decodable. +## Raw RocksDB primary writes are versioned at the store boundary + +Classic msgpackr shared-record ids occupy `0x40–0x5f`; record id 2 is therefore `0x42`. RocksDB primary records also begin with `0x42` in normal operation because a current-era millisecond timestamp encoded as a big-endian float64 has that first byte. `RecordEncoder.decode()` cannot distinguish a prefix-less classic record whose structure id is `0x42` from a versioned record using only the persisted bytes. + +`PrimaryRocksDatabase.put()` and `putSync()` enforce the write-side invariant: if a versioned primary write reaches the store without metadata already staged by `recordUpdater`, the wrapper stages a monotonic version and a zero-flags metadata word. The full 12-byte header is required; a timestamp alone would make the decoder consume the first four payload bytes as metadata. A positional numeric version from legacy raw-put callers is retained verbatim. Metadata is cleared in `finally`, including binary writes that bypass `RecordEncoder`, so staged state cannot leak into the next store operation. Normal table writes arrive with metadata already staged and stay on their existing path without an additional allocation. + ## Version gate at startup: downgrades prompt, and only the minor direction is confirmable `getVersionUpdateInfo()` (`dataLayer/hdbInfoController.ts`) compares the store's `data_version_num` (latest `system.hdb_info` record) against the binary's `packageJson.version` on every start. Data newer than binary by a **major** version → hard refusal. Newer by a **minor** version → `forceDowngradePrompt()` asks for confirmation; answering yes records the data version back down to the binary's version and boots (upgrade directives are deliberately additive/downgrade-compatible — see the struct-mode section above and `patchHdbSecretIsHashAttribute` in `upgrade/directives/5-2-0.ts`). diff --git a/resources/PrimaryRocksDatabase.ts b/resources/PrimaryRocksDatabase.ts index cd8b4c25a3..54f3f40668 100644 --- a/resources/PrimaryRocksDatabase.ts +++ b/resources/PrimaryRocksDatabase.ts @@ -3,7 +3,7 @@ import { RocksDatabase, type RocksDatabaseOptions, constants, type Store } from const FRESH_VERSION_FLAG = constants.FRESH_VERSION_FLAG; import { WeakLRUCache } from 'weak-lru-cache'; import { when } from '../utility/when.ts'; -import { entryMap, METADATA, type Entry } from './RecordEncoder.ts'; +import { clearNextEncoding, entryMap, METADATA, stageRawPrimaryEncoding, type Entry } from './RecordEncoder.ts'; /** * RocksDatabase subclass that owns all primary-store behaviour for Harper tables: @@ -150,6 +150,16 @@ export class PrimaryRocksDatabase extends RocksDatabase { return when(this.getEntry(id, { ...options, async: true }), (entry: Entry) => entry?.value); } + put(id: any, value: any, options?: any, _ifVersion?: any): any { + this.#cache?.delete(id); + const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version); + try { + return super.put(id, value, options); + } finally { + if (staged) clearNextEncoding(); + } + } + getRange(options?: any): any { const iterable = super.getRange(options); if (options?.valuesForKey) return iterable.map((v: any) => v?.value); @@ -170,9 +180,14 @@ export class PrimaryRocksDatabase extends RocksDatabase { }); } - putSync(id: any, value: any, options?: any): any { + putSync(id: any, value: any, options?: any, _ifVersion?: any): any { this.#cache?.delete(id); - return super.putSync(id, value, options); + const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version); + try { + return super.putSync(id, value, options); + } finally { + if (staged) clearNextEncoding(); + } } removeSync(id: any, options?: any): any { diff --git a/resources/RecordEncoder.ts b/resources/RecordEncoder.ts index a229f83cb9..fd192d8fc4 100644 --- a/resources/RecordEncoder.ts +++ b/resources/RecordEncoder.ts @@ -27,6 +27,7 @@ import { encodeBlobsWithFilePath, findBlobsInObject, getFileId, + resetBlobsWereEncoded, } from './blob.ts'; import { getThisNodeId } from './nodeIdMapping.ts'; import { recordAction } from './analytics/write.ts'; @@ -677,6 +678,18 @@ export function setNextEncoding(timestamp: number, metadata: number, expiresAt = nodeIdAtNextEncoding = nodeId; residencyIdAtNextEncoding = residencyId; } +export function stageRawPrimaryEncoding(encoder: any, version?: number) { + if ( + encoder.useVersions === false || + encoder.autoVersion || + timestampNextEncoding !== 0 || + metadataInNextEncoding >= 0 + ) + return false; + setNextEncoding(version ?? getNextMonotonicTime(), 0); + resetBlobsWereEncoded(); + return true; +} /** * Reset the module-level "next encoding" metadata to its no-metadata defaults. These globals are * set just before a versioned encode and consumed (and reset) by the encode hook. If the consuming diff --git a/resources/blob.ts b/resources/blob.ts index 81981b9fac..fdef49a4f4 100644 --- a/resources/blob.ts +++ b/resources/blob.ts @@ -133,6 +133,9 @@ let encodeForStorageForRecordId: number = undefined; // only enable encoding of let promisedWrites: Array>; let currentStore: any; // the root store of the database we are currently encoding for export let blobsWereEncoded = false; // keep track of whether blobs were encoded with file paths +export function resetBlobsWereEncoded() { + blobsWereEncoded = false; +} // the header is 8 bytes const DEFAULT_BLOB_READ_TIMEOUT = 20000; /** diff --git a/unitTests/resources/primaryRocksMetadataRepair.test.js b/unitTests/resources/primaryRocksMetadataRepair.test.js index 13d64b65d1..02c9cf7e04 100644 --- a/unitTests/resources/primaryRocksMetadataRepair.test.js +++ b/unitTests/resources/primaryRocksMetadataRepair.test.js @@ -10,6 +10,7 @@ const fs = require('fs-extra'); const { setupTestDBPath } = require('../testUtils'); const { RocksDatabase } = require('@harperfast/rocksdb-js'); const { Packr } = require('msgpackr'); +const { ACTION_32_BIT } = require('#src/resources/auditStore'); const { RecordEncoder, RecordObject, setNextEncoding } = require('#src/resources/RecordEncoder'); const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); @@ -69,3 +70,101 @@ describe('PrimaryRocksDatabase metadata repair (#2012)', function () { assert(viaGet instanceof RecordObject); }); }); + +describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () { + let dbPath, root, store, rawStore; + + before(function () { + if (isLMDB) return this.skip(); + dbPath = path.join(setupTestDBPath(), 'rocks-raw-write-versioning'); + fs.removeSync(dbPath); + root = RocksDatabase.open(dbPath, {}); + store = new PrimaryRocksDatabase(dbPath, { + name: 'RawWriteTest/', + encoder: { Encoder: RecordEncoder }, + useVersions: true, + sharedStructuresKey: Symbol.for('structures'), + }).open(); + store.initStore(root); + rawStore = RocksDatabase.open(dbPath, { name: 'RawWriteTest/', encoding: false }); + }); + + after(function () { + rawStore?.close(); + store?.close(); + root?.close(); + }); + + it('versions mixed-shape async raw puts before classic structure id 0x42 can collide', async function () { + for (let copy = 1; copy <= 2; copy++) { + await store.put(`failure-${copy}`, { + backend: 'unknown', + method: 'generate', + model: 'missing', + success: false, + error_code: 'backend_not_found', + }); + } + for (let copy = 1; copy <= 2; copy++) { + await store.put(`generate-${copy}`, { + backend: 'deterministic', + method: 'generate', + model: 'probe', + success: true, + prompt_tokens: 1, + completion_tokens: 1, + }); + } + for (let copy = 1; copy <= 2; copy++) { + await store.put(`stream-${copy}`, { + backend: 'deterministic', + method: 'generateStream', + model: 'probe', + success: true, + }); + } + + const bytes = rawStore.getBinarySync('stream-2'); + assert.strictEqual(bytes.readUint32BE(8), ACTION_32_BIT << 24, 'raw write must include a zero flags word'); + assert.strictEqual( + bytes[12], + 0x42, + `precondition: stream shape must use classic structure id 0x42; bytes=${bytes.toString('hex')}` + ); + assert.deepStrictEqual( + { ...store.getSync('stream-2') }, + { backend: 'deterministic', method: 'generateStream', model: 'probe', success: true } + ); + assert(store.getEntry('stream-2').version > 0); + }); + + it('uses a positional version verbatim for sync raw puts', function () { + const version = 1_800_000_000_000.25; + store.putSync('explicit-version', { source: 'legacy-call-shape' }, version, 123); + assert.strictEqual(store.getEntry('explicit-version').version, version); + }); + + it('does not replace metadata already staged by recordUpdater', function () { + const version = 1_800_000_000_001.5; + setNextEncoding(version, 0); + store.putSync('staged-version', { source: 'record-updater' }, { version }); + assert.strictEqual(store.getEntry('staged-version').version, version); + }); + + it('clears wrapper metadata when a raw binary write bypasses the encoder', function () { + store.putSync('binary', { ['\x10binary-data\x02']: Buffer.from([1, 2, 3]) }, 123); + store.putSync('after-binary', { source: 'next-write' }); + assert.notStrictEqual(store.getEntry('after-binary').version, 123); + }); + + it('clears wrapper metadata when the store rejects before encoding', function () { + const closed = new PrimaryRocksDatabase(dbPath, { + name: 'ClosedRawWriteTest/', + encoder: { Encoder: RecordEncoder }, + }); + closed.initStore(root); + assert.throws(() => closed.putSync('closed', { source: 'failure' }, 456), /Database not open/); + store.putSync('after-failure', { source: 'next-write' }); + assert.notStrictEqual(store.getEntry('after-failure').version, 456); + }); +}); From 2d726807bb154394a3b3297e0040f2ad42b58239 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 14 Aug 2026 09:41:58 -0600 Subject: [PATCH 2/3] fix(storage): harden raw write metadata --- DESIGN.md | 2 +- agent/session.ts | 8 +--- resources/PrimaryRocksDatabase.ts | 4 +- resources/RecordEncoder.ts | 12 +++-- .../primaryRocksMetadataRepair.test.js | 48 ++++++++++++++----- 5 files changed, 50 insertions(+), 24 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 5c24a14d08..7a1cb1a242 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -28,7 +28,7 @@ To match v4: `OpenDBIObject` sets `randomAccessStructure = isPrimary`, and `Reco Classic msgpackr shared-record ids occupy `0x40–0x5f`; record id 2 is therefore `0x42`. RocksDB primary records also begin with `0x42` in normal operation because a current-era millisecond timestamp encoded as a big-endian float64 has that first byte. `RecordEncoder.decode()` cannot distinguish a prefix-less classic record whose structure id is `0x42` from a versioned record using only the persisted bytes. -`PrimaryRocksDatabase.put()` and `putSync()` enforce the write-side invariant: if a versioned primary write reaches the store without metadata already staged by `recordUpdater`, the wrapper stages a monotonic version and a zero-flags metadata word. The full 12-byte header is required; a timestamp alone would make the decoder consume the first four payload bytes as metadata. A positional numeric version from legacy raw-put callers is retained verbatim. Metadata is cleared in `finally`, including binary writes that bypass `RecordEncoder`, so staged state cannot leak into the next store operation. Normal table writes arrive with metadata already staged and stay on their existing path without an additional allocation. +`PrimaryRocksDatabase.put()` and `putSync()` enforce the write-side invariant: if an initialized RocksDB encoder for a versioned primary write reaches the store without metadata already staged by `recordUpdater`, the wrapper stages a monotonic version and a zero-flags metadata word. The full 12-byte header is required; a timestamp alone would make the decoder consume the first four payload bytes as metadata. A positive, finite positional numeric version from legacy raw-put callers is retained verbatim. Metadata is cleared in `finally`, including binary writes that bypass `RecordEncoder`, so staged state cannot leak into the next store operation. The wrapper also resets the blob encoder's prior-write signal before a raw record encode so the row cannot inherit a stale `HAS_BLOBS` flag. Normal table writes arrive with metadata already staged and stay on their existing path without an additional allocation. ## Version gate at startup: downgrades prompt, and only the minor direction is confirmable diff --git a/agent/session.ts b/agent/session.ts index 68dddeb0db..531c92cf17 100644 --- a/agent/session.ts +++ b/agent/session.ts @@ -60,12 +60,8 @@ export async function createSession(opts: CreateSessionOpts): Promise entry?.value); } - put(id: any, value: any, options?: any, _ifVersion?: any): any { + put(id: any, value: any, options?: any): any { this.#cache?.delete(id); const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version); try { @@ -180,7 +180,7 @@ export class PrimaryRocksDatabase extends RocksDatabase { }); } - putSync(id: any, value: any, options?: any, _ifVersion?: any): any { + putSync(id: any, value: any, options?: any): any { this.#cache?.delete(id); const staged = stageRawPrimaryEncoding(this.#enc, typeof options === 'number' ? options : options?.version); try { diff --git a/resources/RecordEncoder.ts b/resources/RecordEncoder.ts index fd192d8fc4..38179b06fe 100644 --- a/resources/RecordEncoder.ts +++ b/resources/RecordEncoder.ts @@ -680,13 +680,17 @@ export function setNextEncoding(timestamp: number, metadata: number, expiresAt = } export function stageRawPrimaryEncoding(encoder: any, version?: number) { if ( - encoder.useVersions === false || - encoder.autoVersion || timestampNextEncoding !== 0 || - metadataInNextEncoding >= 0 + metadataInNextEncoding >= 0 || + encoder?.isRocksDB !== true || + encoder.useVersions === false || + encoder.autoVersion ) return false; - setNextEncoding(version ?? getNextMonotonicTime(), 0); + setNextEncoding( + typeof version === 'number' && version > 0 && Number.isFinite(version) ? version : getNextMonotonicTime(), + 0 + ); resetBlobsWereEncoded(); return true; } diff --git a/unitTests/resources/primaryRocksMetadataRepair.test.js b/unitTests/resources/primaryRocksMetadataRepair.test.js index 02c9cf7e04..b93b0f7056 100644 --- a/unitTests/resources/primaryRocksMetadataRepair.test.js +++ b/unitTests/resources/primaryRocksMetadataRepair.test.js @@ -11,11 +11,20 @@ const { setupTestDBPath } = require('../testUtils'); const { RocksDatabase } = require('@harperfast/rocksdb-js'); const { Packr } = require('msgpackr'); const { ACTION_32_BIT } = require('#src/resources/auditStore'); -const { RecordEncoder, RecordObject, setNextEncoding } = require('#src/resources/RecordEncoder'); +const { + RecordEncoder, + RecordObject, + setNextEncoding, + stageRawPrimaryEncoding, +} = require('#src/resources/RecordEncoder'); const { PrimaryRocksDatabase } = require('#src/resources/PrimaryRocksDatabase'); const isLMDB = process.env.HARPER_STORAGE_ENGINE === 'lmdb'; +function asBinary(buffer) { + return { ['\x10binary-data\x02']: buffer }; +} + describe('PrimaryRocksDatabase metadata repair (#2012)', function () { let dbPath, root, store, rawStore; @@ -95,9 +104,13 @@ describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () { root?.close(); }); - it('versions mixed-shape async raw puts before classic structure id 0x42 can collide', async function () { + it('does not stage metadata for an encoder that is not initialized for RocksDB', function () { + assert.strictEqual(stageRawPrimaryEncoding({ useVersions: true, isRocksDB: false }, 123), false); + }); + + it('versions mixed-shape raw puts before classic structure id 0x42 can collide', function () { for (let copy = 1; copy <= 2; copy++) { - await store.put(`failure-${copy}`, { + store.putSync(`failure-${copy}`, { backend: 'unknown', method: 'generate', model: 'missing', @@ -106,7 +119,7 @@ describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () { }); } for (let copy = 1; copy <= 2; copy++) { - await store.put(`generate-${copy}`, { + store.putSync(`generate-${copy}`, { backend: 'deterministic', method: 'generate', model: 'probe', @@ -116,7 +129,7 @@ describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () { }); } for (let copy = 1; copy <= 2; copy++) { - await store.put(`stream-${copy}`, { + store.putSync(`stream-${copy}`, { backend: 'deterministic', method: 'generateStream', model: 'probe', @@ -140,19 +153,32 @@ describe('PrimaryRocksDatabase raw-write versioning (#1762)', function () { it('uses a positional version verbatim for sync raw puts', function () { const version = 1_800_000_000_000.25; - store.putSync('explicit-version', { source: 'legacy-call-shape' }, version, 123); + store.putSync('explicit-version', { source: 'legacy-call-shape' }, version); assert.strictEqual(store.getEntry('explicit-version').version, version); }); + it('uses a positional version verbatim for async raw puts', async function () { + const version = 1_800_000_000_000.5; + await store.put('explicit-async-version', { source: 'async-call-shape' }, version); + assert.strictEqual(store.getEntry('explicit-async-version').version, version); + }); + + it('replaces an unusable explicit version with a monotonic version', function () { + store.putSync('zero-version', { source: 'legacy-zero' }, 0); + assert(store.getEntry('zero-version').version > 0); + }); + it('does not replace metadata already staged by recordUpdater', function () { - const version = 1_800_000_000_001.5; - setNextEncoding(version, 0); - store.putSync('staged-version', { source: 'record-updater' }, { version }); - assert.strictEqual(store.getEntry('staged-version').version, version); + const stagedVersion = 1_800_000_000_001.5; + setNextEncoding(stagedVersion, 0); + store.putSync('staged-version', { source: 'record-updater' }, { version: stagedVersion + 1 }); + assert.strictEqual(store.getEntry('staged-version').version, stagedVersion); }); it('clears wrapper metadata when a raw binary write bypasses the encoder', function () { - store.putSync('binary', { ['\x10binary-data\x02']: Buffer.from([1, 2, 3]) }, 123); + const buffer = Buffer.from([1, 2, 3]); + store.putSync('binary', asBinary(buffer), 123); + assert.deepStrictEqual(rawStore.getBinarySync('binary'), buffer); store.putSync('after-binary', { source: 'next-write' }); assert.notStrictEqual(store.getEntry('after-binary').version, 123); }); From b078b205305ab821e0dc9a16bb9456e7780054d2 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 14 Aug 2026 09:47:47 -0600 Subject: [PATCH 3/3] test(storage): cover stale blob metadata --- agent/session.ts | 3 +-- unitTests/resources/primaryRocksMetadataRepair.test.js | 9 +++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/agent/session.ts b/agent/session.ts index 531c92cf17..899b5cbab2 100644 --- a/agent/session.ts +++ b/agent/session.ts @@ -60,8 +60,7 @@ export async function createSession(opts: CreateSessionOpts): Promise store.encoder.encode({ blob }), 99, root); + store.putSync('after-blob', { source: 'raw-write' }); + assert.strictEqual(rawStore.getBinarySync('after-blob').readUint32BE(8), ACTION_32_BIT << 24); + }); + it('clears wrapper metadata when a raw binary write bypasses the encoder', function () { const buffer = Buffer.from([1, 2, 3]); store.putSync('binary', asBinary(buffer), 123);