diff --git a/DESIGN.md b/DESIGN.md index 8b8a19dcbd..7a1cb1a242 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 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 `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/agent/session.ts b/agent/session.ts index 68dddeb0db..899b5cbab2 100644 --- a/agent/session.ts +++ b/agent/session.ts @@ -60,12 +60,7 @@ export async function createSession(opts: CreateSessionOpts): Promise entry?.value); } + put(id: any, value: any, options?: 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); @@ -172,7 +182,12 @@ export class PrimaryRocksDatabase extends RocksDatabase { putSync(id: any, value: any, options?: 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..38179b06fe 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,22 @@ export function setNextEncoding(timestamp: number, metadata: number, expiresAt = nodeIdAtNextEncoding = nodeId; residencyIdAtNextEncoding = residencyId; } +export function stageRawPrimaryEncoding(encoder: any, version?: number) { + if ( + timestampNextEncoding !== 0 || + metadataInNextEncoding >= 0 || + encoder?.isRocksDB !== true || + encoder.useVersions === false || + encoder.autoVersion + ) + return false; + setNextEncoding( + typeof version === 'number' && version > 0 && Number.isFinite(version) ? 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..89b02b6d2a 100644 --- a/unitTests/resources/primaryRocksMetadataRepair.test.js +++ b/unitTests/resources/primaryRocksMetadataRepair.test.js @@ -10,11 +10,22 @@ 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 { ACTION_32_BIT } = require('#src/resources/auditStore'); +const { createBlob, encodeBlobsWithFilePath } = require('#src/resources/blob'); +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; @@ -69,3 +80,126 @@ 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('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++) { + store.putSync(`failure-${copy}`, { + backend: 'unknown', + method: 'generate', + model: 'missing', + success: false, + error_code: 'backend_not_found', + }); + } + for (let copy = 1; copy <= 2; copy++) { + store.putSync(`generate-${copy}`, { + backend: 'deterministic', + method: 'generate', + model: 'probe', + success: true, + prompt_tokens: 1, + completion_tokens: 1, + }); + } + for (let copy = 1; copy <= 2; copy++) { + store.putSync(`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); + 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 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('does not inherit a stale blob flag on a wrapper-staged raw write', function () { + const blob = createBlob(Buffer.from('prior write')); + blob.saveInRecord = true; + encodeBlobsWithFilePath(() => 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); + assert.deepStrictEqual(rawStore.getBinarySync('binary'), buffer); + 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); + }); +});