Record caching #410
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,178 @@ | ||
| import { RocksDatabase, type RocksDatabaseOptions, constants, type Store } from '@harperfast/rocksdb-js'; | ||
|
|
||
| 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'; | ||
|
|
||
| /** | ||
| * RocksDatabase subclass that owns all primary-store behaviour for Harper tables: | ||
| * - RecordEncoder metadata extraction (version, metadataFlags, etc.) | ||
| * - Optional WeakLRUCache keyed on entry version, verified via the process-wide VerificationTable | ||
| * | ||
| * Replaces both the old CachingRocksDatabase and the RocksDB-specific patches applied by | ||
| * handleLocalTimeForGets. Call initStore(rootStore) after open() — or let | ||
| * handleLocalTimeForGets delegate to it automatically via the isPrimaryRocksDatabase marker. | ||
| * | ||
| * Caching is enabled by default; pass { cache: false } to disable (useful for benchmarking | ||
| * or stores where version tracking is not desired). | ||
| * | ||
| * Cache freshness pattern (using the rocksdb-js VT API): | ||
| * 1. verifyVersion(key, cached.version) → true → return cached entry (no disk I/O) | ||
| * 2. verifyVersion(key, cached.version) → false → read from DB, populateVersion, update cache | ||
| */ | ||
| export class PrimaryRocksDatabase extends RocksDatabase { | ||
| readonly isPrimaryRocksDatabase = true; | ||
| #cache?: WeakLRUCache; | ||
| readCount = 0; | ||
| cachePuts = false; | ||
| declare rootStore: any; | ||
| declare decoder: any; | ||
|
|
||
| get #enc(): any { | ||
| return (this as any).encoder; | ||
| } | ||
|
|
||
| constructor(pathOrStore: string | Store, options?: RocksDatabaseOptions & { cache?: boolean }) { | ||
| const enableCache = (options as any)?.cache !== false; | ||
| super(pathOrStore, enableCache ? { ...options, verificationTable: true } : options); | ||
| if (enableCache) { | ||
| this.#cache = new WeakLRUCache(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Initialises encoder/decoder state. Must be called once after open() with the root | ||
| * RocksDatabase. Equivalent to the RocksDB branch of handleLocalTimeForGets, but as | ||
| * a real method rather than instance-level monkey-patching. | ||
| */ | ||
| initStore(rootStore: RocksDatabase) { | ||
| this.readCount = 0; | ||
| this.cachePuts = false; | ||
| this.rootStore = rootStore; | ||
| this.#enc.rootStore = rootStore; | ||
| this.#enc.isRocksDB = true; | ||
| this.decoder = this.#enc; | ||
| } | ||
|
|
||
| #withEntry(entry: Entry, id: any): Entry { | ||
| if (entry.value) { | ||
| if (entry.value.constructor === Object && this.#enc.structPrototype) { | ||
| const originalValue = entry.value; | ||
| entry.value = new this.#enc.structPrototype.constructor(); | ||
| Object.assign(entry.value, originalValue); | ||
| } | ||
| if (typeof entry.value === 'object' && entry.value !== null) { | ||
| entryMap.set(entry.value, entry); | ||
| } | ||
| } | ||
| entry.key = id; | ||
| return entry; | ||
| } | ||
|
|
||
| #processEntry(raw: any, id: any): Entry | undefined { | ||
| if (raw == null) return undefined; | ||
| if (raw[METADATA]) { | ||
| raw.metadataFlags = raw[METADATA]; | ||
| return this.#withEntry(raw, id); | ||
| } | ||
| return { value: raw, key: id } as Entry; | ||
| } | ||
|
|
||
| /** | ||
| * Core read method. Returns a full Entry (with version, metadataFlags, value, …) or | ||
| * undefined. When caching is enabled, passes `expectedVersion` to the native layer so | ||
| * a single call handles both verification (returns FRESH_VERSION_FLAG on hit) and VT | ||
| * population (auto-seeded on DB read). Only cold reads (no cached version) need a | ||
| * separate populateVersion call. | ||
| */ | ||
| getEntry(id: any, options?: any): any { | ||
| this.readCount++; | ||
| const cache = this.#cache; | ||
| const cached = cache?.get(id) as Entry | undefined; | ||
| const expectedVersion = cached?.version; | ||
|
|
||
| // Build get options, always merging with caller options to preserve | ||
| // transaction snapshot. Pass expectedVersion when cached: | ||
| // VT hit → native returns FRESH_VERSION_FLAG, no DB read | ||
| // VT miss → native reads DB and auto-populates VT slot | ||
| // For cold reads (no cached version), use populateVersion flag so the | ||
| // native layer seeds the VT slot in the same call. | ||
| let getOptions: any; | ||
| if (expectedVersion != null) { | ||
| getOptions = options ? { ...options, expectedVersion } : { expectedVersion }; | ||
| } else if (cache) { | ||
| getOptions = options ? { ...options, populateVersion: true } : { populateVersion: true }; | ||
| } else { | ||
| getOptions = options; | ||
| } | ||
| const raw = options?.async ? super.get(id, getOptions) : super.getSync(id, getOptions); | ||
|
|
||
| return when(raw, (result) => { | ||
| if (result === FRESH_VERSION_FLAG) return cached; | ||
| const entry = this.#processEntry(result, id); | ||
| if (entry == null) { | ||
| if (cache && cached !== undefined) cache.delete(id); | ||
| return undefined; | ||
| } | ||
| if (entry.version != null && cache) { | ||
| cache.set(id, entry, (entry.size ?? 0) >> 10); | ||
| } | ||
| return entry; | ||
| }); | ||
| } | ||
|
|
||
| getSync(id: any, options?: any): any { | ||
| const entry = this.getEntry(id, options) as Entry; | ||
| const value = entry?.value; | ||
| if (value != null && typeof value === 'object') entryMap.set(value, entry); | ||
| return value; | ||
| } | ||
|
|
||
| get(id: any, options?: any): any { | ||
| return when(this.getEntry(id, { ...options, async: true }), (entry: Entry) => { | ||
| const value = entry?.value; | ||
| if (value != null && typeof value === 'object') entryMap.set(value, entry); | ||
| return value; | ||
| }); | ||
| } | ||
|
|
||
| getRange(options?: any): any { | ||
| const iterable = super.getRange(options); | ||
| if (options?.valuesForKey) return iterable.map((v: any) => v?.value); | ||
| if (options?.values === false || options?.onlyCount) return iterable; | ||
| const hasRecordEncoder = !!this.#enc.isRocksDB; | ||
| return iterable.map((entry: any) => { | ||
| if (hasRecordEncoder) { | ||
| if (entry.value?.[METADATA]) { | ||
| entry.metadataFlags = entry.value[METADATA]; | ||
| Object.assign(entry, entry.value); | ||
| } | ||
| if (entry.value?.constructor === Object && this.#enc.structPrototype) { | ||
| const originalValue = entry.value; | ||
| entry.value = new this.#enc.structPrototype.constructor(); | ||
| for (const key in originalValue) entry.value[key] = originalValue[key]; | ||
| } | ||
| } | ||
| return entry; | ||
| }); | ||
| } | ||
|
|
||
| putSync(id: any, value: any, options?: any): any { | ||
| this.#cache?.delete(id); | ||
| return super.putSync(id, value, options); | ||
| } | ||
|
|
||
| removeSync(id: any, options?: any): any { | ||
| this.#cache?.delete(id); | ||
| return super.removeSync(id, options); | ||
| } | ||
|
|
||
| open(): PrimaryRocksDatabase { | ||
| return super.open() as PrimaryRocksDatabase; | ||
| } | ||
|
|
||
| static open(pathOrStore: string | Store, options?: RocksDatabaseOptions): PrimaryRocksDatabase { | ||
| return new PrimaryRocksDatabase(pathOrStore, options).open(); | ||
| } | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
weak-lru-cacheis not documented independencies.md.Per this repo's own policy, every new runtime dependency requires an entry in
dependencies.mdanswering the standard checklist (size, security track record, overlap, memory cost, removal plan, etc.). Please add the entry before merging.