Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions config-root.schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -412,6 +412,10 @@
"description": "Disable fsync for faster writes (risk of data loss on crash). Default: false"
},
"caching": { "type": "boolean", "description": "Enable in-memory caching of records. Default: true" },
"randomAccessFields": {
"type": "boolean",
"description": "Encode records as typed random-access structures, optimizing for fast field access and smaller records. Best for tables with stable, homogeneous field types; wide or variably-typed schemas can set this to false to fall back to classic shared-structure encoding. Applies to each primary table on open: existing records still decode either way, only new writes change. Default: true"
},
"compression": {
"oneOf": [
{ "type": "boolean" },
Expand Down
8 changes: 4 additions & 4 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@
"minimist": "1.2.8",
"moment": "2.30.1",
"mqtt-packet": "~9.0.1",
"msgpackr": "1.11.13",
"msgpackr": "1.12.0",
"needle": "3.5.0",
"node-forge": "^1.3.1",
"node-stream-zip": "1.15.0",
Expand Down
36 changes: 35 additions & 1 deletion resources/RecordEncoder.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
*/

import { Encoder } from 'msgpackr';
import { get as envGet } from '../utility/environment/environmentManager.js';
import { CONFIG_PARAMS } from '../utility/hdbTerms.ts';
import {
HAS_PREVIOUS_RESIDENCY_ID,
HAS_CURRENT_RESIDENCY_ID,
Expand Down Expand Up @@ -90,6 +92,16 @@ export class RecordEncoder extends Encoder {
// long-lived primary store, so a wide/sparse schema (whose records vary by per-field value
// width) can grow it unbounded and exhaust memory. Caller-overridable; default caps it.
options.maxOwnStructures ??= 256;
// When random-access fields are disabled (storage.randomAccessFields=false), write records as
// classic shared structures instead of typed random-access structures. randomAccessStructure stays
// on so reads still decode either form — existing typed-struct data remains readable; only new
// writes change. lmdb-js does not forward non-whitelisted encoder options, so the flag is derived
// from the global config here rather than passed through the store options. Read at construction
// (DBI open, a cold path) so env/CLI config overrides are applied; an explicit option still wins
// (e.g. rocksdb-js's option spread, or tests).
if (options.readOnlyStructures === undefined && envGet(CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS) === false) {
options.readOnlyStructures = true;
}
/**
* The base class for records that provides the read-only methods for accessing
* metadata and will be assigned computed property getters. On its own, these instances
Expand Down Expand Up @@ -246,7 +258,14 @@ export class RecordEncoder extends Encoder {
let nextByte = buffer[start];
let metadataFlags = 0;
try {
if ((this.isRocksDB && nextByte === 66) || (nextByte < 32 && end > 2)) {
// The metadata/timestamp prefix is detected heuristically by the first byte. For rocksdb a
// local-timestamp prefix starts with 66 — but 66 (0x42) is also classic shared-structure
// record-id #2, so a timestamp-less classic record beginning with that id is misread as a
// timestamped record (8 bytes stripped → corrupt). Callers that pass a value known to have no
// prefix (e.g. the audit store's getValue) set options.noMetadata to skip the heuristic. Typed
// structs start at 0x20-0x3f and never hit this, which is why it only surfaces with classic
// structures (random-access fields off).
if (!options?.noMetadata && ((this.isRocksDB && nextByte === 66) || (nextByte < 32 && end > 2))) {
// record with metadata
// this means that the record starts with a local timestamp (that was assigned by lmdb-js).
// we copy it so we can decode it as float-64; we need to do it first because if structural data
Expand Down Expand Up @@ -343,6 +362,21 @@ export class RecordEncoder extends Encoder {
}
}
}

/**
* Encoder for custom-index object stores (e.g. HNSW vector graphs). These hold fixed-shape internal
* nodes — numeric-keyed per-level connection arrays and quantized bins — that depend on random-access
* struct encoding and are mutated in place during graph maintenance. Keep them writing typed structs
* even when storage.randomAccessFields disables structs for user tables: their node shapes are
* controlled, so the wide/heterogeneous OOM risk that motivates opt-out doesn't apply, and classic
* (frozen) decoding would break the in-place graph mutation.
*/
export class IndexRecordEncoder extends RecordEncoder {
constructor(options) {
options.readOnlyStructures = false;
super(options);
}
}
function getTimestamp() {
TIMESTAMP_HOLDER[0] = TIMESTAMP_HOLDER[0] ^ 0x40; // restore the first byte, we xor to differentiate the first byte from structures
return TIMESTAMP_VIEW.getFloat64(0);
Expand Down
23 changes: 23 additions & 0 deletions resources/Table.ts
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,19 @@ const RECORD_PRUNING_INTERVAL = 60000; // one minute
envMngr.initSync();
const LMDB_PREFETCH_WRITES = envMngr.get(CONFIG_PARAMS.STORAGE_PREFETCHWRITES);
const LOCK_TIMEOUT = 10000;

// True only for frozen plain/record objects (the immutable decoded records). Excludes Buffers,
// TypedArrays, ArrayBuffers and primitives/null, which must not be shallow-copied via spread (that
// would corrupt binary values). Used to decide when to copy-on-mutate before stamping a record.
function isFrozenRecordObject(value: any): boolean {
return (
value !== null &&
typeof value === 'object' &&
!ArrayBuffer.isView(value) &&
!(value instanceof ArrayBuffer) &&
Object.isFrozen(value)
);
}
export const INVALIDATED = 1;
export const EVICTED = 8; // note that 2 is reserved for timestamps
const TEST_WRITE_KEY_BUFFER = Buffer.allocUnsafeSlow(8192);
Expand Down Expand Up @@ -1639,6 +1652,12 @@ export function makeTable(options) {
if (fullUpdate || (recordUpdate && hasChanges(this.#changes === recordUpdate ? this : recordUpdate))) {
if (!context?.source) {
transaction.checkOverloaded();
// Records are intentionally immutable: decoded records are frozen (and 5.2 record
// caching relies on it), so mutating in place would corrupt cached/shared state.
// validate() coerces values and we stamp created/updated times + the primary key
// below, so copy-on-mutate when recordUpdate is frozen (e.g. a record decoded during
// log replay) instead of writing through the frozen object.
if (isFrozenRecordObject(recordUpdate)) recordUpdate = { ...recordUpdate };
this.validate(recordUpdate, !fullUpdate);
if (updatedTimeProperty) {
recordUpdate[updatedTimeProperty.name] =
Expand Down Expand Up @@ -4255,6 +4274,10 @@ export function makeTable(options) {
}
}
if (typeof updatedRecord.toJSON === 'function') updatedRecord = updatedRecord.toJSON();
// updatedRecord may still be a frozen record (e.g. a reused existingRecord); copy-on-mutate
// before stamping the primary key below (records are immutable — 5.2 record caching relies
// on it — so we must not write through the frozen object).
if (isFrozenRecordObject(updatedRecord)) updatedRecord = { ...updatedRecord };
if (primaryKey && updatedRecord[primaryKey] !== id) updatedRecord[primaryKey] = id;
}
resolved = true;
Expand Down
5 changes: 4 additions & 1 deletion resources/auditStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -552,7 +552,10 @@ export function readAuditEntry(buffer: Uint8Array, start = 0, end = undefined):
if (action & HAS_RECORD || (action & HAS_PARTIAL_RECORD && !fullRecord)) {
if (!value) {
value = decodeFromDatabase(
() => store.decoder.decode(buffer.subarray(decoder.position, end)),
// the audit value has no on-disk timestamp/metadata prefix (the audit entry carries
// its own time), so skip the prefix heuristic — otherwise a classic record whose
// structure-id byte is 66 (0x42) is misread as a rocksdb timestamp. See RecordEncoder.decode.
() => store.decoder.decode(buffer.subarray(decoder.position, end), { noMetadata: true }),
store.rootStore
);
}
Expand Down
8 changes: 7 additions & 1 deletion resources/databases.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import harperLogger from '../utility/logging/harper_logger.js';
const { forComponent } = harperLogger;
import * as manageThreads from '../server/threads/manageThreads.js';
import { openAuditStore, readAuditEntry, createAuditEntry, type AuditRecord } from './auditStore.ts';
import { handleLocalTimeForGets } from './RecordEncoder.ts';
import { handleLocalTimeForGets, IndexRecordEncoder } from './RecordEncoder.ts';
import { deleteRootBlobPathsForDB } from './blob.ts';
import { CUSTOM_INDEXES } from './indexes/customIndexes.ts';
import { OpenDBIObject } from '../utility/lmdb/OpenDBIObject.js';
Expand Down Expand Up @@ -824,6 +824,12 @@ function openIndex(dbiKey: string, rootStore: RootDatabaseKind, attribute: any)
const objectStorage =
attribute.isPrimaryKey || (attribute.indexed.type && CUSTOM_INDEXES[attribute.indexed.type]?.useObjectStore);
const dbiInit = createOpenDBIObject(!objectStorage, objectStorage);
// Custom-index object stores (e.g. HNSW vector graphs) must keep writing typed structs regardless
// of the storage.randomAccessFields opt-out — their internal nodes are mutated in place and depend
// on random-access struct encoding (see IndexRecordEncoder).
if (attribute.indexed?.type && CUSTOM_INDEXES[attribute.indexed.type]?.useObjectStore) {
dbiInit.encoder = { Encoder: IndexRecordEncoder };
}
let dbi:
| LMDBDatabase
| (RocksDatabase & {
Expand Down
1 change: 1 addition & 0 deletions static/defaultConfig.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ rootPath: null
storage:
writeAsync: false
caching: true
randomAccessFields: true
compression: true
noReadAhead: false
path: null
Expand Down
179 changes: 179 additions & 0 deletions unitTests/resources/recordEncoder.test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,179 @@
require('../testUtils');
const assert = require('assert');

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Small repo-standard nit: new tests should import Node's built-in assert module with the node: prefix. Either node:assert or node:assert/strict works; the important part is avoiding the bare assert package name.\n\nsuggestion\nconst assert = require('node:assert/strict');\n\n\nCodex review

const { RecordEncoder, IndexRecordEncoder } = require('#src/resources/RecordEncoder');
const env = require('#js/utility/environment/environmentManager');
const terms = require('#src/utility/hdbTerms');

// In-memory shared structures (mirrors how a DBI shares the structures array under
// Symbol.for('structures')) so struct/record structures cross between encoder instances. We keep the
// live array reference rather than round-tripping through encode/decode, since msgpackr attaches
// bookkeeping to the structures array that a re-decode would strip.
function sharedStore() {
let structures = [];
return {
save(s) {
structures = s;
return true;
},
get() {
return structures;
},
};
}

function makeEncoder(store, extra) {
return new RecordEncoder({
// randomAccessStructure stays on regardless of the opt-out — reads must keep decoding typed
// structs so existing data is still readable; only writes change.
randomAccessStructure: true,
getStructures: store.get,
saveStructures: store.save,
...extra,
});
}

const record = { name: 'price', type: 'Float', indexed: true };

describe('RecordEncoder random-access fields opt-out (readOnlyStructures)', () => {
it('writes typed random-access structs by default (random-access fields on)', () => {
const store = sharedStore();
const enc = makeEncoder(store);
const bytes = enc.encode(record);
assert.ok(bytes[0] >= 0x20 && bytes[0] < 0x40, `expected typed-struct header byte, got 0x${bytes[0].toString(16)}`);
});

it('writes classic shared structures when readOnlyStructures is set (opt-out)', () => {
const store = sharedStore();
const enc = makeEncoder(store, { readOnlyStructures: true });
const bytes = enc.encode(record);
assert.ok(
bytes[0] >= 0x40 && bytes[0] < 0x80,
`expected classic shared-structure byte, got 0x${bytes[0].toString(16)}`
);
});

it('opt-out encoder still reads typed-struct data written before opting out', () => {
// Existing data was written as typed structs; after the user opts out the same store must keep
// decoding it (randomAccessStructure stays on), only new writes switch to classic.
const store = sharedStore();
const typedWriter = makeEncoder(store);
const typedBytes = typedWriter.encode(record);
assert.ok(typedBytes[0] >= 0x20 && typedBytes[0] < 0x40, 'precondition: typed-struct bytes');

const optedOutReader = makeEncoder(store, { readOnlyStructures: true });
const decoded = optedOutReader.decode(Buffer.from(typedBytes));
assert.ok(decoded, 'typed-struct data should still decode (not swallowed to null)');
assert.strictEqual(decoded.name, record.name);
assert.strictEqual(decoded.type, record.type);
assert.strictEqual(decoded.indexed, record.indexed);
});

it('classic records and typed structs round-trip together on the same store', () => {
const store = sharedStore();
const typedWriter = makeEncoder(store);
const classicWriter = makeEncoder(store, { readOnlyStructures: true });
const reader = makeEncoder(store);

const a = { id: 1, kind: 'typed' };
const b = { id: 2, kind: 'classic', extra: true };
const typedBytes = typedWriter.encode(a);
const classicBytes = classicWriter.encode(b);
assert.ok(typedBytes[0] >= 0x20 && typedBytes[0] < 0x40, 'precondition: typed bytes');
assert.ok(classicBytes[0] >= 0x40 && classicBytes[0] < 0x80, 'precondition: classic bytes');

// Typed structs decode with the RecordObject prototype, classic records as plain objects; spread
// to compare field values regardless of prototype.
assert.deepStrictEqual({ ...reader.decode(Buffer.from(typedBytes)) }, a);
assert.deepStrictEqual({ ...reader.decode(Buffer.from(classicBytes)) }, b);
});

it('IndexRecordEncoder keeps writing typed structs even with readOnlyStructures requested', () => {
// Object-store indexes (HNSW) must stay in struct mode regardless of the table opt-out.
const store = sharedStore();
const enc = new IndexRecordEncoder({
randomAccessStructure: true,
readOnlyStructures: true,
getStructures: store.get,
saveStructures: store.save,
});
const bytes = enc.encode(record);
assert.ok(
bytes[0] >= 0x20 && bytes[0] < 0x40,
`index encoder should ignore readOnlyStructures and write typed structs, got 0x${bytes[0].toString(16)}`
);
});

it('decodes a classic record whose structure-id byte is 66 (0x42) when noMetadata is set', () => {
// Regression: 66 (0x42) is classic shared-structure record-id #2 and also a rocksdb local-timestamp
// marker. On a rocksdb store the prefix heuristic strips 8 bytes and corrupts the record, decoding
// to null (the MQTT "publish non-JSON" failure). The audit store passes { noMetadata: true } to skip
// the heuristic for values that have no on-disk timestamp prefix.
const store = sharedStore();
const writer = makeEncoder(store, { readOnlyStructures: true });
// The 3rd distinct classic structure starts with byte 0x42; encode three shapes to reach it.
writer.encode({ a: 1 });
writer.encode({ b: 2, c: 3 });
const target = { d: 4, e: 5, f: 6 };
const bytes = Buffer.from(writer.encode(target));
assert.strictEqual(bytes[0], 66, 'precondition: target record begins with structure-id byte 66 (0x42)');

const reader = makeEncoder(store);
// Warm up the structure cache with a normal (non-rocksdb) decode so getStructures isn't needed
// once we flip isRocksDB below (the rocksdb getStructures path needs a rootStore we don't mock).
assert.deepStrictEqual(reader.decode(bytes), target, 'baseline classic decode works');

// Simulate the rocksdb decode path, where byte 66 collides with the local-timestamp marker.
reader.isRocksDB = true;
assert.deepStrictEqual(
reader.decode(bytes, { noMetadata: true }),
target,
'with noMetadata the classic record still decodes correctly'
);
// Without noMetadata the rocksdb timestamp heuristic misreads byte 66 and corrupts the decode.
assert.notDeepStrictEqual(
reader.decode(bytes),
target,
'without noMetadata the 0x42 collision corrupts the decode (demonstrating why the flag is needed)'
);
});

describe('storage.randomAccessFields config drives the default', () => {
let previous;
beforeEach(() => {
previous = env.get(terms.CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS);
});
afterEach(() => {
env.setProperty(terms.CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS, previous);
});

it('writes classic structures when storage.randomAccessFields is false (no explicit option)', () => {
env.setProperty(terms.CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS, false);
const enc = makeEncoder(sharedStore()); // no explicit readOnlyStructures
const bytes = enc.encode(record);
assert.ok(
bytes[0] >= 0x40 && bytes[0] < 0x80,
`config off should write classic structures, got 0x${bytes[0].toString(16)}`
);
});

it('writes typed structs when storage.randomAccessFields is true', () => {
env.setProperty(terms.CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS, true);
const enc = makeEncoder(sharedStore());
const bytes = enc.encode(record);
assert.ok(
bytes[0] >= 0x20 && bytes[0] < 0x40,
`config on should write typed structs, got 0x${bytes[0].toString(16)}`
);
});

it('an explicit readOnlyStructures option overrides the config', () => {
env.setProperty(terms.CONFIG_PARAMS.STORAGE_RANDOMACCESSFIELDS, false);
const enc = makeEncoder(sharedStore(), { readOnlyStructures: false });
const bytes = enc.encode(record);
assert.ok(
bytes[0] >= 0x20 && bytes[0] < 0x40,
`explicit readOnlyStructures:false should keep typed structs despite config, got 0x${bytes[0].toString(16)}`
);
});
});
});
1 change: 1 addition & 0 deletions utility/hdbTerms.ts
Original file line number Diff line number Diff line change
Expand Up @@ -554,6 +554,7 @@ export const CONFIG_PARAMS = {
STORAGE_WRITEASYNC: 'storage_writeAsync',
STORAGE_OVERLAPPINGSYNC: 'storage_overlappingSync',
STORAGE_CACHING: 'storage_caching',
STORAGE_RANDOMACCESSFIELDS: 'storage_randomAccessFields',
STORAGE_COMPRESSION: 'storage_compression',
STORAGE_NOREADAHEAD: 'storage_noReadAhead',
STORAGE_PREFETCHWRITES: 'storage_prefetchWrites',
Expand Down