// src/mini-db.ts // // MiniDb: the public embedded API. Ties together the in-memory Store (with its // ordered key index), the WAL, recovery, compaction, dt-column indexes, value // secondary indexes, and full-text indexes. // // Document model: // { key: string(<=128), value: , dt1..dtN: } import fs from 'node:fs/promises'; import fsSync from 'node:fs'; import path from 'node:path'; import { randomUUID } from 'node:crypto'; import { Store } from './store.js'; import type { StoreRecord } from './store.js'; import { WAL } from './wal.js'; import { ValueReader } from './value-reader.js'; import { catchUpWalAsync } from './recovery.js'; import { compact, shouldCompact } from './compaction.js'; import { OpTracker } from './op-tracker.js'; import { TEXT_INDEXES_FILE, SNAPSHOT_FILE, isPersistentFile, rootPostingsFile, textDictionaryFile, textDocsFile } from './generation.js'; import { IndexManager } from './index-manager.js'; import { DtIndex } from './dt-index.js'; import { TextIndex } from './text-index/index.js'; import { CompoundIndexManager } from './compound-index.js'; import { toKStr, writeFileAtomic } from './value-codec.js'; import { LockFile } from './lockfile.js'; import { createSerializer } from './serialize.js'; import { MaintenanceScheduler, defaultWorkerSlots, TEXT_BUILD_SLOT_WAIT_MS } from './maintenance.js'; import { MemoryGuard } from './memory-guard.js'; import { backup as runBackup, backupInProgressError as newBackupInProgressError } from './backup.js'; import type { BackupDeps } from './backup.js'; import { QueryEngine } from './query-engine.js'; import { GenerationBuilder, TEXT_BUILD_WORKER_MIN_DOCS, GEN_BUILD_WAL_DELTA_BYTES } from './generation-builder.js'; import type { GenBuildOp } from './generation-builder.js'; import { WalGroupTracker } from './wal-group.js'; import { WritePath } from './write-path.js'; import { withWindowsEpermRetry } from './rename-replace.js'; import { openMiniDb, closeMiniDb, renewMiniDbLock, openOrRebuildMiniDb } from './lifecycle.js'; import { IndexAdmin } from './index-admin.js'; import { ReadPath } from './read-path.js'; import { createMiniDbStats } from './stats.js'; import { LifecycleTracker } from './lifecycle-status.js'; import type { MiniDbLifecycleStatus } from './lifecycle-status.js'; import type { LifecycleHooks } from './lifecycle.js'; import { TextRegistry } from './text-registry.js'; import type { TextIndexDef } from './text-registry.js'; import type { MaintenanceContext, MaintenanceTaskInfo } from './maintenance.js'; import type { FrameRef } from './codec.js'; import type { FsyncPolicy } from './wal.js'; import type { RecoveryMode, RecoveryInfo, ValueMode, RecoveredOp } from './recovery.js'; import type { IndexDef, IndexInfo } from './index-manager.js'; import type { CompoundIndexDef, CompoundIndexInfo } from './compound-index.js'; import type { RangeOptions } from './skiplist.js'; import type { TextIndexTokenizerName } from './trigram.js'; import type { PostingEntry } from './text-postings.js'; import { startWorkerTextBuild, textBuildWorkerAvailable, verifyFileCrcAsync, WorkerTextBuildError } from './worker/text-build.js'; import type { TextBuildCheckpoint, WorkerTextBuildFallbackReason } from './worker/text-build.js'; import { readBaseDocsImageAsync, BASE_DOCS_MAGIC, BASE_DOCS_VERSION } from './worker/text-build-core.js'; import type { TextBuildCoreResult } from './worker/text-build-core.js'; import { readGenerationFileCheckedAsync, readTextDictionaryImageAsync } from './gen-codec.js'; import type { ValueCodec, ValueCodecName, OpenOptions, RestoreOptions, SetOptions, BatchInputOp, DocRecord, ScanEntry, QueryOptions, } from './types.js'; // ClusterDb (the multi-process sharding layer) lives at the './cluster' // subpath export to keep this module free of import cycles. export class MiniDb { dir!: string; walPath!: string; /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ indexPath!: string; /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ compoundIndexPath!: string; store!: Store; wal!: WAL; valueReader?: ValueReader; valueMode: ValueMode = 'memory'; readonly indexes = new IndexManager(); readonly dt = new DtIndex(); readonly compound = new CompoundIndexManager(); /** Text-index registry state lives in the TextRegistry facet (declared * below, after stats); these views keep the generation / compaction / * write paths' call sites unchanged. */ private get text(): Map { return this.textRegistry.text; } private get textDefs(): TextIndexDef[] { return this.textRegistry.textDefs; } private get textDrops(): Set { return this.textRegistry.textDrops; } /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ codec!: ValueCodec; /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ codecName: ValueCodecName = 'buffer'; fsyncPolicy: FsyncPolicy = 'everysec'; syncIntervalMs = 1000; /** Lifecycle state machine. 'closing' is a real state (not just a flag on * the way down): a cleanup failure leaves the instance there so a later * close() call can retry the remaining cleanup, and ensureOpen rejects * 'closing' and 'closed' alike. */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ state: 'open' | 'closing' | 'closed' = 'open'; /** The in-flight close() cleanup pass, shared by concurrent close() calls. */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ closePromise: Promise | null = null; recoveryInfo: RecoveryInfo | null = null; /** Continuation watermark for catchUpFromWal: the WAL inode + applied * offset as advanced by the last successful catch-up (recoveryInfo's scan * endpoint anchors the first call). */ private walTail: { dev: number; ino: number; size: number } | null = null; /** Catch-up serializer: the sliced async apply yields to the event loop, so * overlapping catch-ups would interleave op-by-op (the pre-async apply was * atomic per call by virtue of being synchronous). Chaining restores that * per-call atomicity; a caller queued behind another catch-up observes the * advanced watermark and no-ops or falls back to a full reopen. */ private catchUpChain: Promise = Promise.resolve(); readOnly = false; /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ lock: LockFile | null = null; compactThresholdBytes = 64 * 1024 * 1024; autoCompact = true; compacting = false; _compactDone: Promise | null = null; /** Set only during compaction's short rotation critical section; writers park * on it (see the write-op gate). Null the rest of the time, so the snapshot * phase of compaction is fully non-blocking. */ _rotateLock: Promise | null = null; lastCompactError: unknown = null; maxMemoryBytes: number | null = null; maxMemoryPolicy: 'reject' | 'evict-lru' = 'reject'; /** The LRU access set lives in the MemoryGuard facet (declared below, after * stats); this view keeps the class's delete-only call sites unchanged. */ private get access(): Set { return this.memoryGuard.access; } /** Serializes write ops while any unique index exists (check-then-apply must * be atomic against other writers). Shared promise-chain pattern — see * serialize.ts. */ private readonly serializeUniqueWrites = createSerializer(); /** Per-sidecar mutation chains (one promise-chain mutex per index-definition * sidecar file, plan 10): a create/drop runs its whole staged → persist → * publish sequence under its sidecar's chain, so concurrent mutations of * the SAME definition file can never interleave (before this, two * concurrent creates shared one fixed .tmp — one renamed the other's tmp * away — and a persist failure diverged the live registry from disk). * Different sidecar types do NOT block each other, and the data write path * (set/batch/del) never touches these chains. The in-chain rebuild is a * full Store walk: index changes are rare admin operations, so holding the * chain across the walk is the accepted trade-off. */ private readonly secondaryDefChain = createSerializer(); private readonly compoundDefChain = createSerializer(); /** The WAL group-commit + in-place recovery gating facet (wal-group.ts): * owns pendingGroups/lastGroup and the recovery chain state * (walRecoveryChain/walRecoveryIdle/walRecoveryCovers); these views keep * the close / catch-up / backup call sites unchanged. */ private readonly walGroups = new WalGroupTracker({ restoreGroupKey: (pk, prev) => this.restoreGroupKey(pk, prev), writeDisabled: () => this.writeDisabled, recoverWalInPlace: (wal) => this.recoverWalInPlace(wal), }); private get walRecoveryChain(): Promise { return this.walGroups.walRecoveryChain; } private get walRecoveryIdle(): boolean { return this.walGroups.walRecoveryIdle; } /** Write-op gate + in-flight counter (plan 12's OpTracker): set/del/batch/ * expire run inside enter/leave, and backup() pauses the gate — the drain * completion is backup's linearization point (every write acknowledged * before it is in the backup; writes submitted meanwhile reject with * BACKUP_IN_PROGRESS). close() does NOT drain it: an op in flight at close * keeps its stage-7/8 semantics (its frame rejects as the WAL closes and * the op rolls back). */ private readonly writeOps = new OpTracker(); /** Serializes whole backup() runs: two backups to the same destination would * otherwise swap each other's freshly-renamed result aside and delete it, * and even to different destinations they would duplicate the compaction + * copy work. The write-gate pause itself is reference-counted and safe to * overlap (see op-tracker.ts). Same promise-chain pattern as * serializeUniqueWrites. */ private readonly serializeBackups = createSerializer(); /** Persistent index generations enabled (OpenOptions.indexGenerations, * default true). When false the instance behaves exactly as before stage * 5: full open-time rebuild + root postings rebuilds after compaction. */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ indexGenerationsEnabled = true; /** Stage 6: the unified maintenance scheduler (compaction + generation * builds: one heavy task at a time, backpressure, preflight, cancellation, * drain on close). Created in open() once the directory is known. */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ maintenance!: MaintenanceScheduler; /** Stage 6: workerized text generation build enabled (rollback switch). */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ textBuildWorkerEnabled = true; /** Sticky fallback: a worker that failed to produce ANY result once (spawn * failure) disables the worker path for the rest of this instance — the * next build uses the in-thread staged path. */ private textWorkerDisabled = false; /** Stage 6: worker aggregation memory budget. */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ textBuildMemoryBytes = 128 * 1024 * 1024; /** TUI-safe worker-slot policy: how long a worker-eligible text build * queues for a process-wide slot before the bounded inline core is * allowed as the last resort (never the unbounded staged aggregation). */ /* Non-private (package-internal): read through the GenerationBuilderDeps view. */ textBuildSlotWaitMs = TEXT_BUILD_SLOT_WAIT_MS; /** Stage 6: maintenance I/O concurrency (snapshot grouped reads). */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ maintenanceIoConcurrency = 8; /** Defer the open-time full text rebuild (no-generation fallback) to a background * bounded build (OpenOptions.deferOpenTextBuilds, default true). */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ deferTextBuildsEnabled = true; /** A read-only deferred build's private scratch dir (outside the db dir); * created on first use, dropped on close (lifecycle.ts closeResources). */ /* Non-private (package-internal): lifecycle.ts reads/writes this through its LifecycleHost view. */ roScratchDir: string | null = null; /** Runtime generation-build trigger (see maybeAutoGenerationBuild): * throttles — minimum interval between kicks, and the longer backoff * after a build failed/aborted (public test knobs, same pattern as the * search service's). */ genBuildKickMinIntervalMs = 60_000; genBuildKickFailureBackoffMs = 300_000; private lastGenBuildKickAt = 0; private lastGenBuildFailureAt = 0; lastGenBuildError: unknown = null; /** Abort handle / mutation queue / single-flight guard / status of the * generation build all live in the GenerationBuilder facet (declared * below); these views keep the open / write / close paths' call sites * unchanged (the write path feeds the SHARED genBuild object by * reference). */ private get genBuildAbort(): AbortController | null { return this.generationBuilder.genBuildAbort; } private get genBuild(): { queue: GenBuildOp[]; bytes: number; wal: WAL; aborted: boolean } | null { return this.generationBuilder.genBuild; } private get genBuildPromise(): Promise | null { return this.generationBuilder.genBuildPromise; } private get generationInfo(): { id: string; createdAt: number; walCheckpoint: number; records: number } | null { return this.generationBuilder.generationInfo; } /** Set when in-place WAL recovery's truncate fails (persistent I/O error): * from then on every write op throws a WAL_WRITE_DISABLED error * immediately; reads and close() keep working. The value is the truncate * error (the cause). DESIGNED CONSEQUENCE: the WAL stays poisoned, so * close() skips its final flush and the un-acked tail is LEFT in db.wal — * a later reopen replays it and the rejected writes resurface. That is * exactly why every commit-point failure is marked `ambiguous: true`: in * this state the caller cannot assume a rejected write had no effect. */ writeDisabled: unknown = null; readonly stats = createMiniDbStats(); /** Per-open lifecycle telemetry (the open() state machine + per-phase * wall-clock timings): driven by lifecycle.ts and the generation loader, * read through lifecycleStatus(). */ /* Non-private (package-internal): lifecycle.ts drives it through its LifecycleHost view. */ readonly lifecycle = new LifecycleTracker(); /** The text-index registry facet (text-registry.ts): owns the live TextIndex * map, the persisted definition list, and the staged-drop marks (declared * after stats because the injected deps reference it; everything else is * read lazily through the getters). */ private readonly textRegistry = new TextRegistry({ dir: () => this.dir, readOnly: () => this.readOnly, codecName: () => this.codecName, store: () => this.store, ensureOpen: () => this.ensureOpen(), ensureWritable: () => this.ensureWritable(), decode: (b) => this.decode(b), readValueAsync: (kstr) => this.readValueAsync(kstr), textRecords: () => this.textRecords(), persistTextIndexDefinitions: (defs) => this.persistTextIndexDefinitions(defs), boundedTextBuild: (name, ti, def, checkpoint) => this.boundedTextBuild(name, ti, def, checkpoint), }); /** The maxMemory guard facet: owns the LRU access set and the * projected-bytes enforce pass (declared after stats because the injected * deps reference it; store/config are read lazily through the getters). */ private readonly memoryGuard = new MemoryGuard({ store: () => this.store, valueMode: () => this.valueMode, maxMemoryBytes: () => this.maxMemoryBytes, maxMemoryPolicy: () => this.maxMemoryPolicy, stats: this.stats, evictKey: (pk) => this.evictKey(pk), }); /** The backup flow's injected deps (backup.ts); the callbacks read the * live fields at call time, so the facet never holds stale references. */ private readonly backupDeps: BackupDeps = { dir: () => this.dir, stats: this.stats, ensureOpen: () => this.ensureOpen(), compacting: () => this.compacting, compactDone: () => this._compactDone, readOnly: () => this.readOnly, compact: () => this.compact(), pauseWrites: () => this.writeOps.pause(), resumeWrites: () => this.writeOps.resume(), serializeBackups: (fn) => this.serializeBackups(fn), walRecoveryChain: () => this.walRecoveryChain, flushWal: () => this.wal.flush(), }; /** The unified query engine facet (query-engine.ts): read-only; the index * managers and the text map are stable references, store/codecName are * read lazily through the getters. */ private readonly queryEngine = new QueryEngine({ store: () => this.store, indexes: this.indexes, dt: this.dt, text: this.text, codecName: () => this.codecName, stats: this.stats, decode: (b) => this.decode(b), readValueAsync: (kstr) => this.readValueAsync(kstr), ensureOpen: () => this.ensureOpen(), }); /** The generation machinery facet (generation-builder.ts): the load/build * paths and the genBuild state (declared after queryEngine; every dep is * either a stable reference or a lazy getter/callback). */ private readonly generationBuilder = new GenerationBuilder({ dir: () => this.dir, walPath: () => this.walPath, codecName: () => this.codecName, valueMode: () => this.valueMode, readOnly: () => this.readOnly, state: () => this.state, indexGenerationsEnabled: () => this.indexGenerationsEnabled, store: () => this.store, wal: () => this.wal, getValueReader: () => this.valueReader, setValueReader: (reader) => { this.valueReader = reader; }, setRecoveryInfo: (info) => { this.recoveryInfo = info; }, maintenance: () => this.maintenance, maintenanceTaskCtx: () => this.maintenanceTaskCtx, textBuildWorkerEnabled: () => this.textBuildWorkerEnabled, textWorkerDisabled: () => this.textWorkerDisabled, disableTextWorker: () => { this.textWorkerDisabled = true; }, textBuildMemoryBytes: () => this.textBuildMemoryBytes, textBuildSlotWaitMs: () => this.textBuildSlotWaitMs, dt: this.dt, indexes: this.indexes, compound: this.compound, textRegistry: this.textRegistry, stats: this.stats, lifecycle: this.lifecycle, decode: (b) => this.decode(b), indexable: (v): v is Record => this.indexable(v), liveRecords: () => this.liveRecords(), liveRecordsRaw: () => this._liveRecordsRaw(), textRecords: () => this.textRecords(), seedAccessFromStore: () => this.seedAccessFromStore(), applyRecoveredOp: (op) => this.applyRecoveredOp(op), ensureOpen: () => this.ensureOpen(), ensureWritable: () => this.ensureWritable(), boundedTextBuild: (name, ti, def, checkpoint) => this.boundedTextBuild(name, ti, def, checkpoint), noteBuildFailure: (err) => { this.lastGenBuildFailureAt = Date.now(); if (err !== undefined) this.lastGenBuildError = err; }, noteBuildSuccess: () => { this.lastGenBuildError = null; }, }); /** The write path facet (write-path.ts): set/del/batch/expire and their * whole commit machinery (declared last — it wires the other facets as * collaborators; the wal-group / memory-guard / generation-builder * callbacks that point back here are lazy, so init order is safe). */ private readonly writePath = new WritePath({ store: () => this.store, wal: () => this.wal, valueMode: () => this.valueMode, codecName: () => this.codecName, rotateLock: () => this._rotateLock, dt: this.dt, indexes: this.indexes, compound: this.compound, textRegistry: this.textRegistry, walGroups: this.walGroups, memoryGuard: this.memoryGuard, generationBuilder: this.generationBuilder, writeOps: this.writeOps, serializeUniqueWrites: (fn) => this.serializeUniqueWrites(fn), stats: this.stats, encode: (v) => this.encode(v), decode: (b) => this.decode(b), indexable: (v): v is Record => this.indexable(v), ensureOpen: () => this.ensureOpen(), ensureWritable: () => this.ensureWritable(), maybeAutoCompact: () => this.maybeAutoCompact(), }); /** The secondary + compound index admin facet (index-admin.ts). The * persist seams below stay on MiniDb (tests stub them on the instance); * the facet drives them through the injected callbacks. */ private readonly indexAdmin = new IndexAdmin({ indexes: this.indexes, compound: this.compound, dt: this.dt, textRegistry: this.textRegistry, store: () => this.store, codecName: () => this.codecName, ensureOpen: () => this.ensureOpen(), ensureWritable: () => this.ensureWritable(), decode: (b) => this.decode(b), indexable: (v): v is Record => this.indexable(v), stats: this.stats, liveRecords: () => this.liveRecords(), liveRecordsRaw: () => this._liveRecordsRaw(), secondaryDefChain: (fn) => this.secondaryDefChain(fn), compoundDefChain: (fn) => this.compoundDefChain(fn), persistIndexDefinitions: (defs) => this.persistIndexDefinitions(defs), persistCompoundIndexDefinitions: (defs) => this.persistCompoundIndexDefinitions(defs), }); /** The read path facet (read-path.ts): KV reads, scans, dt reads, and the * live-record generators every (re)build feed goes through. */ private readonly readPath = new ReadPath({ store: () => this.store, getValueReader: () => this.valueReader, dt: this.dt, memoryGuard: this.memoryGuard, decode: (b) => this.decode(b), indexable: (v): v is Record => this.indexable(v), ensureOpen: () => this.ensureOpen(), }); /** The private methods the lifecycle functions (lifecycle.ts) call back * into; the fields they touch are read/written through the LifecycleHost * view directly (see the non-private markers above). */ private readonly lifecycleHooks: LifecycleHooks = { onStoreExpire: (k, rec) => this.onStoreExpire(k, rec), loadIndexDefinitions: () => this.loadIndexDefinitions(), loadCompoundIndexDefinitions: () => this.loadCompoundIndexDefinitions(), loadTextIndexDefinitions: () => this.loadTextIndexDefinitions(), tryLoadGeneration: (mode) => this.tryLoadGeneration(mode), rebuildAllIndexes: () => this.rebuildAllIndexes(), submitCompaction: () => this.submitCompaction(), buildGeneration: (trigger) => this.buildGeneration(trigger), seedAccessFromStore: () => this.seedAccessFromStore(), closeAllTextIndexes: () => { for (const ti of this.text.values()) ti.close(); }, generationInfo: () => this.generationInfo, generationStale: () => this.generationStale(), genBuildAbort: () => this.genBuildAbort, genBuildPromise: () => this.genBuildPromise, walRecoveryIdle: () => this.walRecoveryIdle, walRecoveryChain: () => this.walRecoveryChain, }; /** Hook called by compaction after the store snapshot + WAL are rotated, so * derived on-disk state can be rewritten against the new live set. * Structural part of the CompactionTarget interface; the compaction awaits * it, so it may be sync or async. * * Stage 5: with index generations enabled this is ONE publish transaction * — the snapshot rotation and the derived-state checkpoint (store image, * dt/secondary/compound images, text postings) land as a single new * generation, and the live text indexes rebase onto it. The synchronous * rebuildTextPostings() tail no longer runs. With generations disabled the * legacy behavior is kept exactly. */ onCompacted: () => void | Promise = async (): Promise => { const t0 = performance.now(); if (!this.indexGenerationsEnabled) { await this.rebuildTextPostings(); const ms = performance.now() - t0; this.stats.compactionPostingsDurationMs += ms; this.stats.textRebuildDurationMs += ms; return; } await this.buildGeneration('compact'); this.stats.compactionPostingsDurationMs += performance.now() - t0; }; /** The scheduler context of the in-flight maintenance task: compaction's * rotation reports its publishing phase through it (stage 6). */ private maintenanceTaskCtx: MaintenanceContext | null = null; /** CompactionTarget hook: the rotation critical section is the * compaction's publishing phase — a shutdown must wait it out instead of * cancelling mid-rotation (stage 6). The scheduler's markPublishing is * one-way, so 'running' needs no handling. */ onMaintenancePhase = (phase: 'running' | 'publishing'): void => { if (phase === 'publishing') this.maintenanceTaskCtx?.markPublishing(); }; /** Queue a compaction on the maintenance scheduler (stage 6): one heavy * task per database at a time; a duplicate submission dedupes onto the * in-flight/queued one. */ private submitCompaction(): Promise { return this.maintenance.submit('compact', async (ctx) => { this.maintenanceTaskCtx = ctx; try { await compact(this); } finally { this.maintenanceTaskCtx = null; } }); } static async open(opts: OpenOptions): Promise> { const db = new MiniDb(); await openMiniDb(db, opts, db.lifecycleHooks); return db; } /** * Open a database, and if opening fails due to corruption (not due to a live * lock), delete the directory and open a fresh empty database. Recommended for * a rebuildable cache. A live lock is rethrown. * * The destructive rebuild only ever runs for an open that could OWN the * directory: an error tagged `readOnlyOpen` (opts.readOnly, or a lock that * degraded via onLockFail:'readonly') is rethrown untouched — rebuilding * means deleting files, and a read-only bystander must never mutate a live * writer's directory (lock-review repro: the readonly fallback deleted the * writer's sidecar, and in the strict-recovery shape the whole directory). */ static async openOrRebuild( opts: OpenOptions, hooks: { onRebuild?: (err: unknown) => void } = {}, ): Promise> { return openOrRebuildMiniDb(opts, hooks, (o) => MiniDb.open(o)); } private encode(v: V): Buffer { return this.codec.encode(v); } private decode(b: Buffer | undefined): V | undefined { return b === undefined ? undefined : this.codec.decode(b); } private pk(key: string | Buffer): string { return toKStr(key); } private indexable(v: unknown): v is Record { return v !== null && typeof v === 'object'; } private *liveRecords(): Generator<{ key: Buffer; value: V | undefined; dt: Record | null }> { yield* this.readPath.liveRecords(); } /** Live indexable records with canonical keys, for (re)building text indexes. */ private *textRecords(): Generator<{ key: string; value: unknown }> { yield* this.readPath.textRecords(); } /** On-disk postings file path for a text index (root location — the legacy * pre-generation home; the name sanitization lives in generation.ts). */ private textPostingsPath(name: string): string { return this.textRegistry.textPostingsPath(name); } /** LEGACY postings maintenance (indexGenerations: false): rebuild every * dirty text index's on-disk postings from the live Store (see * TextRegistry.rebuildTextPostings). */ private async rebuildTextPostings(): Promise { return this.textRegistry.rebuildTextPostings(); } private async rebuildAllIndexes(): Promise { // Decide per text index whether its base build DEFERS to the bounded // background engine (deferOpenTextBuilds) or stays on the staged inline // walk below. Deferred indexes get neither a staged builder nor feeding // during the walk; open() returns with them "building" instead of // awaiting a corpus-scale rebuild on the caller's event loop. The gates // mirror boundedTextBuild's eligibility; a read-only (memory-base) index // IS deferrable — its build is redirected to the private scratch dir. const deferred = new Set(); if (this.deferTextBuildsEnabled) { for (const [name, ti] of this.text) { if (ti.hasCustomTokenizer) continue; if (!this.textBuildWorkerEnabled || this.textWorkerDisabled) continue; if (this.store.size < TEXT_BUILD_WORKER_MIN_DOCS) continue; deferred.add(name); } } const textRebuildMsBefore = this.stats.textRebuildDurationMs; await this.indexAdmin.rebuildAllIndexes({ skipTextIndex: (name) => deferred.has(name) }); this.lifecycle.time('textRebuildMs', this.stats.textRebuildDurationMs - textRebuildMsBefore); // Every non-deferred text index was just rebuilt by the staged in-thread // walk above (the full-recovery path leaves them empty). for (const name of this.text.keys()) { if (!deferred.has(name)) this.lifecycle.noteTextIndexSource(name, 'staged'); } if (deferred.size > 0) this.deferOpenTextBuilds([...deferred]); } /** Arm the rebase queue on each deferred text index NOW and build its base * in a background maintenance task with the bounded engine, pinned at the * recovery checkpoint. The fallback full recovery feeds only the Store, * so the write * buffers are empty at this point and arming here (before open() returns — * no write or catch-up can interleave) captures exactly the * post-checkpoint ops. Until a build commits, searches on the index raise * TextIndexBuildingError instead of silently serving partial results. A * read-only opener builds into a private scratch dir OUTSIDE the live * writer's db dir and adopts the disk base there (commitRebase's * memBase→disk flip). */ private deferOpenTextBuilds(names: readonly string[]): void { const rec = this.recoveryInfo; if (rec === null) return; const armed: { name: string; ti: TextIndex; def: TextIndexDef }[] = []; for (const name of names) { const ti = this.text.get(name); const def = this.textDefs.find((d) => d.name === name); if (ti === undefined || def === undefined) continue; // Mark the base unavailable from this exact point: the index has no // base at all until the deferred build commits (the fallback recovery // fed only the Store), so searches raise the typed building signal. ti.basePending = true; ti.beginRebase(); armed.push({ name, ti, def }); this.lifecycle.markTextIndexPending(name); } if (armed.length === 0) return; // The checkpoint the build covers: the recovery scan endpoint (see the // header — the queue captures every op after it). const pin = (): TextBuildCheckpoint => ({ walOffset: rec.walScanEnd, walDev: rec.walDev, walIno: rec.walIno, snapshotDev: rec.snapshotDev, snapshotIno: rec.snapshotIno, }); // Re-pinning after a failed attempt (a live writer's compaction can // rotate the WAL under a read-only build, invalidating the anchors): // clear the write buffer, re-read the anchors, and re-arm the queue — // one synchronous segment, so no op slips between the pin and the // capture. Ops the buffer held are all below the fresh checkpoint, hence // covered by the retried build. const repin = (ti: TextIndex): TextBuildCheckpoint => { ti.resetWriteBuffer(); ti.abortRebase(); const walAnchor = fsSync.statSync(this.walPath); let snapshotDev = 0; let snapshotIno = 0; try { const snapAnchor = fsSync.statSync(path.join(this.dir, SNAPSHOT_FILE)); snapshotDev = snapAnchor.dev; snapshotIno = snapAnchor.ino; } catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; } const checkpoint: TextBuildCheckpoint = { walOffset: walAnchor.size, walDev: walAnchor.dev, walIno: walAnchor.ino, snapshotDev, snapshotIno, }; ti.beginRebase(); return checkpoint; }; const scratchDir = this.readOnly ? (this.roScratchDir ??= path.join( this.dir, '..', `${path.basename(this.dir)}.ro-scratch`, `${process.pid}-${randomUUID().slice(0, 8)}`, )) : null; void this.maintenance .submit('text-build', async (ctx) => { for (const { name, ti, def } of armed) { const output = scratchDir !== null ? { dir: scratchDir, postingsPath: path.join(scratchDir, rootPostingsFile(name)) } : null; let checkpoint = pin(); let committed = false; let hostedMode: 'worker' | 'inline' | null = null; const tBuild = performance.now(); for (let attempt = 1; attempt <= 3 && !committed; attempt++) { if (attempt > 1) checkpoint = repin(ti); try { const hosted = await this.boundedTextBuild(name, ti, def, checkpoint, output, ctx.signal); // Ineligible after all (e.g. the sticky worker-disable landed // mid-task): no build will come — stop retrying. if (hosted === null) break; committed = true; hostedMode = hosted; } catch { // Shutdown cancels the task: leave close() to finish teardown. if (ctx.signal.aborted) return; } } this.lifecycle.time('textRebuildMs', performance.now() - tBuild); if (committed && hostedMode !== null) { this.stats.textDeferredBuilds++; this.lifecycle.clearTextIndexPending(name, hostedMode); } else { // Every attempt failed: disarm and mark the base pending — // searches keep the typed building signal instead of silently // degrading to delta-only. A later generation build (its text // commits swap live bases) or the next open attaches a base. ti.abortRebase(); ti.basePending = true; this.stats.textDeferredBuildErrors++; } } }) .catch(() => { // Task-level failure (preflight ENOSPC, backpressure, cancellation): // mark every still-armed index pending (see above). for (const { ti } of armed) { if (ti.rebasing) { ti.abortRebase(); ti.basePending = true; this.stats.textDeferredBuildErrors++; } } }); } /** Whether a text index's base is currently unavailable — a deferred * open-time build is in flight or finally failed — so searches on it * raise TextIndexBuildingError instead of serving partial results. */ textIndexBuilding(name: string): boolean { const ti = this.text.get(name); return ti !== undefined && ti.basePending; } private *_liveRecordsRaw(): Generator<{ key: Buffer; value: unknown }> { yield* this.readPath.liveRecordsRaw(); } // ---- persistent index generations (stage 5) ------------------------------ // // The generation machinery (load path, build path, genBuild state) lives in // the GenerationBuilder facet (generation-builder.ts); MiniDb keeps the // private views and delegates below so the open / compaction / write / close // paths' call sites stay unchanged. private async tryLoadGeneration(mode: RecoveryMode): Promise { return this.generationBuilder.tryLoadGeneration(mode); } private async buildGeneration(trigger: 'open' | 'compact' | 'manual' | 'wal-growth' | 'close'): Promise { return this.generationBuilder.buildGeneration(trigger); } /** Explicit maintenance (stage 5): build + publish a fresh index generation * now. Writer only. The load path is automatic; this exists for operators * who want to force a checkpoint after a large burst of writes instead of * waiting for the next compaction. */ async rebuildGeneration(): Promise { return this.generationBuilder.rebuildGeneration(); } /** Build ONE text index's base with the bounded engine (byte-budget * aggregation + segmented external merge — hosted in a worker thread when * its entry file exists, the same core inline on the main thread * otherwise) and swap it in through the rebase machinery. This is the * memory-bounded alternative to `ti.build(...)`'s staged O(corpus) * aggregation for the two full-corpus build entries: createTextIndex and * the open-time generation-loader rebuild. Returns 'worker' | 'inline' * when the build ran; null when the index is ineligible and the caller * must use the staged path (tiny corpus, custom tokenizer, a memory-base * index without an external `output`, or the explicit rollback switches). * On throw the rebase is aborted and the live index is untouched — the * caller decides whether a staged fallback is safe to attempt. * * `output` redirects the build's artifacts to a caller-owned directory * OUTSIDE the db dir and skips the rename into the db root — the shape a * read-only opener needs: it must not write into the live writer's * directory, so its base postings are built into a private scratch dir * and adopted there (commitRebase flips memBase→disk, same as a * generation attach). The postings file at `output.postingsPath` stays * live after the commit and is owned by the caller from then on. */ private async boundedTextBuild( name: string, ti: TextIndex, def: TextIndexDef, checkpoint: TextBuildCheckpoint | null, output: { dir: string; postingsPath: string } | null = null, signal?: AbortSignal, ): Promise<'worker' | 'inline' | null> { if ( !this.textBuildWorkerEnabled || this.textWorkerDisabled || this.state !== 'open' || // A memory-base index (read-only open) must not write into the live // writer's db dir: it is only eligible when the build's artifacts are // redirected to a caller-owned scratch location. (ti.memBase !== null && output === null) || ti.hasCustomTokenizer || this.store.size < TEXT_BUILD_WORKER_MIN_DOCS ) { return null; } // Pin the checkpoint. createTextIndex (checkpoint === null) gets a seal // identical to the generation build's: beginRebase starts the mutation // capture and the appendOffset read is one synchronous segment with it, // so the build covers every frame below the offset and the rebase queue // every op after it. A caller-pinned checkpoint (the generation loader's // manifest anchors, already verified against the live WAL) skips the // seal — the queue stays empty because writes cannot interleave with the // open-time load. let sealedOffset: number; let walDev: number; let walIno: number; let snapDev = 0; let snapIno = 0; const sizeAtCheckpoint = this.store.size; // The seal needs the live WAL's appendOffset; a memory-base index is a // read-only open whose WAL was never opened, so its caller MUST pin the // checkpoint (recoveryInfo anchors) instead. if (checkpoint === null && ti.memBase !== null) { throw new Error('bounded text build on a memory-base index requires a pinned checkpoint'); } // A caller may have pre-armed the capture (the deferred open-time build // arms at the recovery checkpoint, so ops queue from that exact point // while the build waits on the maintenance queue); arm only otherwise. if (!ti.rebasing) ti.beginRebase(); try { if (checkpoint === null) { sealedOffset = this.wal.appendOffset; await this.wal.flush(); const walAnchor = fsSync.statSync(this.walPath); walDev = walAnchor.dev; walIno = walAnchor.ino; try { const snapAnchor = fsSync.statSync(path.join(this.dir, SNAPSHOT_FILE)); snapDev = snapAnchor.dev; snapIno = snapAnchor.ino; } catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; } } else { sealedOffset = checkpoint.walOffset; walDev = checkpoint.walDev; walIno = checkpoint.walIno; snapDev = checkpoint.snapshotDev; snapIno = checkpoint.snapshotIno; } } catch (e) { ti.abortRebase(); throw e; } // In-place builds land artifacts in a per-index tmp dir inside the db // dir (fixed name, removed before use and in the finally — a crashed // attempt is cleaned by the next one), and the postings file is renamed // to its final root location only after verification, so a crash never // strands a partial file at the live path. An `output`-redirected build // (read-only scratch) writes straight into the caller's dir: crash // residue there is harmless — the next attempt overwrites it and the // whole dir is dropped on close. const tmpDir = output === null ? path.join(this.dir, `${rootPostingsFile(name)}.tmpbuild`) : null; let slotRelease: (() => void) | null = null; try { const artifactsDir = tmpDir ?? output!.dir; if (tmpDir !== null) await fs.rm(tmpDir, { recursive: true, force: true }); await fs.mkdir(artifactsDir, { recursive: true }); const postingsPath = tmpDir !== null ? path.join(tmpDir, rootPostingsFile(name)) : output!.postingsPath; const dictionaryPath = path.join(artifactsDir, textDictionaryFile(name)); const baseDocsPath = path.join(artifactsDir, `${textDocsFile(name)}.base`); // Worker thread when its entry file exists; the inline bounded core is // the last resort — never the unbounded staged aggregation. TUI-safe // slot policy: queue for a process-wide worker slot first (bounded by // textBuildSlotWaitMs and the caller's abort signal) instead of // dropping a large build onto the main thread the moment every slot is // busy; only a persisted slot drought (or a deployment without the // worker file) hosts the SAME bounded core inline. const workerAvailable = textBuildWorkerAvailable(); let inlineReason: WorkerTextBuildFallbackReason | undefined; if (workerAvailable) { slotRelease = await defaultWorkerSlots.acquireBounded(this.textBuildSlotWaitMs, signal); if (slotRelease === null) inlineReason = 'slot-pressure'; } else { inlineReason = 'runtime-unavailable'; } const inline = slotRelease === null; const handle = startWorkerTextBuild( { snapshotPath: snapIno !== 0 ? path.join(this.dir, SNAPSHOT_FILE) : null, walPath: this.walPath, walOffset: sealedOffset, walDev, walIno, snapshotDev: snapDev, snapshotIno: snapIno, indexes: [ { name, fields: def.fields, tokenizer: def.tokenizer === 'ngram' ? ('ngram' as const) : ('default' as const), postingsPath, dictionaryPath, baseDocsPath, }, ], memoryBudgetBytes: this.textBuildMemoryBytes, }, { shouldAbort: () => this.state !== 'open' || signal?.aborted === true, inline, inlineReason, signal, onFallback: (reason) => { this.stats.textWorkerFallbacks++; this.stats.lastTextWorkerFallback = reason; }, }, ); let result: TextBuildCoreResult; try { result = await handle.promise; } catch (error) { if (!handle.inline && !(error instanceof WorkerTextBuildError && error.aborted)) { this.stats.textWorkerErrors++; } throw error; } if (result.scannedLiveKeys > sizeAtCheckpoint) { throw new Error( `text build scanned ${result.scannedLiveKeys} live keys > checkpoint store ${sizeAtCheckpoint} (pinning protocol violation)`, ); } // Verify before trusting (the worker's output is discardable evidence): // streaming crc on the raw postings file, envelope+crc on the images. const r = result.indexes[0]!; await verifyFileCrcAsync(postingsPath, r.postingsInfo); const dictPayload = await readGenerationFileCheckedAsync(dictionaryPath, 'MDTD', 1, r.dictionaryInfo); const dictEntries = (await readTextDictionaryImageAsync(dictPayload)).map( (e) => [e.term, { off: e.off, len: e.len, df: e.df }] as [string, PostingEntry], ); const baseDocsPayload = await readGenerationFileCheckedAsync(baseDocsPath, BASE_DOCS_MAGIC, BASE_DOCS_VERSION, r.baseDocsInfo); const baseDocs = await readBaseDocsImageAsync(baseDocsPayload); const containers = await TextIndex.prepareRebaseContainers(dictEntries, baseDocs.keys, baseDocs.docLens); const finalPostingsPath = tmpDir !== null ? this.textPostingsPath(name) : postingsPath; if (tmpDir !== null) await fs.rename(postingsPath, finalPostingsPath); ti.commitRebase({ postingsPath: finalPostingsPath, containers, liveCount: r.liveCount, postingsFileInfo: r.postingsInfo, }); if (!handle.inline) this.stats.textWorkerBuilds++; return handle.inline ? 'inline' : 'worker'; } catch (e) { ti.abortRebase(); throw e; } finally { slotRelease?.(); if (tmpDir !== null) { await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => {}); } else { // Scratch artifacts: the dictionary/base-docs images were commit-time // evidence only — the committed base is the postings file alone, // which stays live and is dropped with the whole scratch dir on // close. (A failed build's partial postings file is overwritten by // the next attempt and removed with the dir as well.) await fs.rm(path.join(output!.dir, textDictionaryFile(name)), { force: true }).catch(() => {}); await fs.rm(path.join(output!.dir, `${textDocsFile(name)}.base`), { force: true }).catch(() => {}); } } } /** Stable generation status: the generation this instance loaded at open or * last published (null when running on the legacy recovery path). */ getIndexGeneration(): { id: string; createdAt: number; walCheckpoint: number; records: number } | null { return this.generationBuilder.getIndexGeneration(); } private async loadIndexDefinitions(): Promise { return this.indexAdmin.loadIndexDefinitions(this.indexPath); } /** Persist the given secondary-index definition list. The CONTENT is the * caller's transaction decision (live list ± the mutation), never an * implicit snapshot of the registry — a create persists live+staged BEFORE * publishing, a drop persists live-minus BEFORE removing. */ private async persistIndexDefinitions(defs: IndexInfo[]): Promise { await writeFileAtomic(this.indexPath, JSON.stringify(defs), { stats: this.stats }); } private async loadTextIndexDefinitions(): Promise { return this.textRegistry.loadTextIndexDefinitions(); } /** Persist the given text-index definition list (same transaction-content * rule as persistIndexDefinitions). This is the registry's persistence * SEAM (injected into TextRegistry): kept on MiniDb so tests can stub it * on the instance. */ private async persistTextIndexDefinitions(defs: TextIndexDef[]): Promise { await writeFileAtomic(path.join(this.dir, TEXT_INDEXES_FILE), JSON.stringify(defs), { stats: this.stats }); } private async loadCompoundIndexDefinitions(): Promise { return this.indexAdmin.loadCompoundIndexDefinitions(this.compoundIndexPath); } /** Persist the given compound-index definition list (same * transaction-content rule as persistIndexDefinitions). */ private async persistCompoundIndexDefinitions(defs: CompoundIndexInfo[]): Promise { await writeFileAtomic(this.compoundIndexPath, JSON.stringify(defs), { stats: this.stats }); } /** Drop every derived index entry for a key that just expired in the Store. */ private onStoreExpire(k: string, _rec: StoreRecord): void { this.access.delete(k); this.dt.del(k); this.compound.remove(k); if (this.indexes.size) this.indexes.remove(k, undefined); for (const ti of this.text.values()) ti.remove(k); } private maybeAutoCompact(): void { if (this.autoCompact && !this.compacting && shouldCompact(this)) this.submitCompaction().catch(() => {}); this.maybeAutoGenerationBuild(); } /** The staleness rule shared by the runtime wal-growth trigger and the * close-time best-effort publish: no valid current generation while there * is data worth checkpointing (a writer that started from an empty db and * grew its WAL past the threshold without ever reaching the compaction * threshold — the window the open-time kick cannot cover), or the WAL * grew/rotated past the threshold since the current generation's * checkpoint (a rotated-away anchor reads as a negative drift). */ private generationStale(): boolean { if (this.readOnly || !this.indexGenerationsEnabled || this.state !== 'open') return false; const gen = this.generationInfo; const walOffset = this.wal.appendOffset; return gen === null ? this.store.size > 0 && walOffset >= GEN_BUILD_WAL_DELTA_BYTES : walOffset - gen.walCheckpoint >= GEN_BUILD_WAL_DELTA_BYTES || walOffset < gen.walCheckpoint; } /** Runtime generation-availability trigger, riding the per-write * maintenance hook (maybeAutoCompact): kick a background generation build * when the current generation is missing/stale. Throttled per instance, * with a longer backoff after a failed/aborted build so a hopelessly * churning writer is not re-kicked every interval; the scheduler's * same-kind dedupe makes a redundant kick cheap. */ private maybeAutoGenerationBuild(): void { const now = Date.now(); if (now - this.lastGenBuildKickAt < this.genBuildKickMinIntervalMs) return; if (now - this.lastGenBuildFailureAt < this.genBuildKickFailureBackoffMs) return; if (!this.generationStale()) return; this.lastGenBuildKickAt = now; void this.buildGeneration('wal-growth').catch(() => {}); } // ---- WAL poison: in-place recovery + flush-group rollback ---------------- // // Commit point semantics: an op is committed when its frame's `done` // resolves. A WAL write/fsync failure poisons the WAL (see wal.ts) and // rejects every un-acked frame; each rejected op rolls its flush group back // to the pre-group records, then the instance recovers the WAL in place — // truncate db.wal to the failed batch's first predicted offset (removing // exactly the un-acked bytes), refreshSize, clearPoison — so the on-disk // tail a reopen would replay and the in-memory state agree again. private writeDisabledError(): Error { return this.walGroups.writeDisabledError(); } /** Recover a poisoned WAL back to a known-safe point: truncate db.wal to * the failed batch's first predicted offset — exactly the un-acked bytes; * every acknowledged write sits in earlier, successful batches — then * re-sync the live WAL's size bookkeeping and clear the poison. * * Mutual exclusion with a compaction rotation (which has its own recovery: * swapping in a fresh WAL at the real EOF): the truncate targets the PATH, * so it is correct whether or not the rotation's recovery swapped the WAL * meanwhile, and the bookkeeping refresh hits the CURRENT WAL. Both sides * only ever truncate to the same poison offset, so the composition never * double-executes. * * A truncate failure (the I/O error persists) parks the instance in * writeDisabled: the poison is kept, so appends keep rejecting, reads keep * working and close() skips its final flush. */ private async recoverWalInPlace(wal: WAL): Promise { let poison = wal.poison; if (!poison) return; // Settle any in-flight flush first: a poisonPending truncation point is // predicted against the in-flight batch fully landing, so truncating past // the real EOF would zero-extend the file (a corrupt gap on reopen). A // failed in-flight batch widens the point via poisonWith meanwhile. await wal.whenIdle(); poison = wal.poison; if (!poison) return; try { // Stale-coordinate guard: if db.wal was REPLACED since the poison was // recorded (a compaction rotation committed a new, shorter file at the // path — the commit-body guards make this unreachable for poisons // recorded during/after the seal, so this is defense-in-depth), the // offset belongs to the old file's coordinate system and truncating to // it would zero-extend the new file. The new file never carried the // un-acked tail, so skipping the truncate is the correct recovery. await withWindowsEpermRetry(async () => { const st = await fs.stat(this.walPath); if (poison.failedAtOffset <= st.size) await fs.truncate(this.walPath, poison.failedAtOffset); }); } catch (err) { this.writeDisabled = err; return; } await this.wal.refreshSize(); wal.clearPoison(); } private touchAccess(pk: string): void { this.memoryGuard.touchAccess(pk); } private seedAccessFromStore(): void { this.memoryGuard.seedAccessFromStore(); } private async evictKey(pk: string): Promise { return this.writePath.evictKey(pk); } // ---- KV API ------------------------------------------------------------- get(key: string | Buffer): V | undefined { return this.readPath.get(key); } /** Async value resolution for a canonical key (see ReadPath.readValueAsync). */ private async readValueAsync(kstr: string): Promise { return this.readPath.readValueAsync(kstr); } /** Async twin of get() (stage 6, additive): identical result; only the * disk-mode value read moves off the event loop. */ async getAsync(key: string | Buffer): Promise { return this.readPath.getAsync(key); } getRecord(key: string | Buffer): DocRecord | undefined { return this.readPath.getRecord(key); } async set(key: string | Buffer, value: V, { ttl, dt }: SetOptions = {}): Promise { return this.writePath.set(key, value, { ttl, dt }); } async del(key: string | Buffer): Promise { return this.writePath.del(key); } /** Atomically apply a batch of operations (all-or-nothing). */ async batch(ops: readonly BatchInputOp[]): Promise { return this.writePath.batch(ops); } /** The unguarded restore core behind restoreKey and the flush-group * rollback: put the key back to `prev` across the store and every derived * index (TTL/access/dt/secondary/compound/text). */ private restoreGroupKey(pk: string, prev: StoreRecord | undefined): void { this.writePath.restoreGroupKey(pk, prev); } /** Apply one recovered WAL frame during catchUpFromWal: the same ops * open-time recovery derives from it (frameToOps), plus the incremental * derived-index maintenance applyOp performs on the write path — minus * unique checks: the writer already validated, and intermediate frame * states must apply literally (LWW). */ private applyRecoveredFrameAsync(f: FrameRef, fd: number, slice: () => boolean): Promise { return this.writePath.applyRecoveredFrameAsync(f, fd, slice); } private applyRecoveredOp(op: RecoveredOp): void { this.writePath.applyRecoveredOp(op); } has(key: string | Buffer): boolean { return this.readPath.has(key); } get size(): number { return this.store.size; } async mset(entries: readonly (readonly [string, V])[]): Promise { if (!entries.length) return; await this.batch(entries.map(([key, value]) => ({ op: 'set' as const, key, value }))); } mget(keys: readonly string[]): (V | undefined)[] { return this.readPath.mget(keys); } async expire(key: string | Buffer, ttlMs: number): Promise { return this.writePath.expire(key, ttlMs); } ttl(key: string | Buffer): number { return this.readPath.ttl(key); } // ---- key-ordered scans -------------------------------------------------- scan(opts: RangeOptions = {}): ScanEntry[] { return this.readPath.scan(opts); } prefix(p: string, limit = Infinity): ScanEntry[] { return this.readPath.prefix(p, limit); } // ---- dt column queries -------------------------------------------------- dtColumns(): string[] { return this.readPath.dtColumns(); } dtRange(col: string, opts: RangeOptions & { limit?: number } = {}): (ScanEntry & { dtValue: number })[] { return this.readPath.dtRange(col, opts); } // ---- value secondary indexes ------------------------------------------- async createIndex(name: string, opts: IndexDef): Promise { return this.indexAdmin.createIndex(name, opts); } async dropIndex(name: string): Promise { return this.indexAdmin.dropIndex(name); } listIndexes(): IndexInfo[] { return this.indexAdmin.listIndexes(); } findEq(name: string, value: unknown): { key: string; value: V | undefined }[] { return this.indexAdmin.findEq(name, value); } findRange(name: string, opts: Parameters[1]): { key: string; value: V | undefined; field: number }[] { return this.indexAdmin.findRange(name, opts); } // ---- compound indexes (groupBy + orderBy) ------------------------------- async createCompoundIndex(name: string, def: CompoundIndexDef): Promise { return this.indexAdmin.createCompoundIndex(name, def); } async dropCompoundIndex(name: string): Promise { return this.indexAdmin.dropCompoundIndex(name); } listCompoundIndexes(): CompoundIndexInfo[] { return this.indexAdmin.listCompoundIndexes(); } /** * Ordered range within a group, e.g. "sessions in workspace X ordered by * updatedAt". O(log N + limit) — no full sort. */ compoundRange( name: string, groupValue: unknown, opts: RangeOptions & { limit?: number } = {}, ): { key: string; value: V | undefined; orderValue: unknown }[] { return this.indexAdmin.compoundRange(name, groupValue, opts); } // ---- full-text search --------------------------------------------------- async createTextIndex( name: string, { fields, tokenizer }: { fields?: readonly string[]; tokenizer?: TextIndexTokenizerName } = {}, ): Promise { return this.textRegistry.createTextIndex(name, { fields, tokenizer }); } async dropTextIndex(name: string): Promise { return this.textRegistry.dropTextIndex(name); } search(name: string, q: string, opts: { op?: 'AND' | 'OR'; limit?: number; maxVisits?: number } = {}): { key: string; value: V | undefined; score: number }[] { return this.textRegistry.search(name, q, opts); } /** * `search` with work accounting: `opts.maxVisits` bounds how many posting * entries the index visits (see TextIndex.searchBounded); the result * reports the visits and whether the budget truncated the candidate set * (hits are then a subset of the full matches, never false hits). */ searchBounded( name: string, q: string, opts: { op?: 'AND' | 'OR'; limit?: number; maxVisits?: number } = {}, ): { hits: { key: string; value: V; score: number }[]; visits: number; truncated: boolean } { return this.textRegistry.searchBounded(name, q, opts); } /** Async twin of searchBounded (stage 6, additive): identical hits and * budget accounting; the postings reads and the disk-mode value reads run * off the event loop. The server's search path prefers this variant. */ async searchBoundedAsync( name: string, q: string, opts: { op?: 'AND' | 'OR'; limit?: number; maxVisits?: number } = {}, ): Promise<{ hits: { key: string; value: V; score: number }[]; visits: number; truncated: boolean }> { return this.textRegistry.searchBoundedAsync(name, q, opts); } /** Async convenience: the async counterpart of search(). */ async searchAsync( name: string, q: string, opts: { op?: 'AND' | 'OR'; limit?: number; maxVisits?: number } = {}, ): Promise<{ key: string; value: V | undefined; score: number }[]> { return this.textRegistry.searchAsync(name, q, opts); } // ---- unified query ------------------------------------------------------ query(q: QueryOptions = {}): ScanEntry[] { return this.queryEngine.query(q); } /** Async twin of query() (stage 6, additive): identical results and * ordering; the disk-mode value reads (and the text branch's postings * reads) run off the event loop. The candidate-collection logic mirrors * query() exactly — keep both in sync when the query planner changes. */ async queryAsync(q: QueryOptions = {}): Promise[]> { return this.queryEngine.queryAsync(q); } /** The rejection a write op gets while a backup holds the write gate: the * fence is short (file copies) and retryable, so callers can simply * re-issue the write afterwards. */ private backupInProgressError(): Error { return newBackupInProgressError(); } /** Write a consistent online backup of this database directory. * * Semantics (plan 12): backup pauses the write gate — new writes reject * with BACKUP_IN_PROGRESS — and waits for every in-flight write to settle. * That drain completion IS the linearization point: every write * acknowledged before it is included in the backup, every write submitted * after it is not. The copy itself is an atomic commit: persistent files * go to a sibling temp dir, every copied file is fsync'd, the manifest is * written LAST (the commit marker — a manifest on disk implies every file * it lists is fully copied and durable), then the temp dir is renamed over * the destination (an existing previous backup is swapped aside first and * restored if the rename fails). A failure anywhere before the rename * leaves the destination untouched and the temp dir removed — never a half * backup. Concurrent backups serialize on serializeBackups. */ async backup(destDir: string, opts: { compact?: boolean } = {}): Promise { return runBackup(this.backupDeps, destDir, opts); } /** Restore a backup directory into destDir and open it. */ static async restore(srcDir: string, destDir: string, opts: RestoreOptions = {}): Promise> { if (!srcDir) throw new TypeError('restore: srcDir is required'); if (!destDir) throw new TypeError('restore: destDir is required'); const { force, ...openOpts } = opts; if (force) { await fs.rm(destDir, { recursive: true, force: true }); } else { try { const existing = await fs.readdir(destDir); if (existing.length) throw new Error(`restore destination is not empty: ${destDir}`); } catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; } } await fs.mkdir(destDir, { recursive: true }); const names = await fs.readdir(srcDir); for (const name of names) { if (isPersistentFile(name) || name === 'backup.manifest.json') { const src = path.join(srcDir, name); const st = await fs.stat(src); if (st.isDirectory()) await fs.cp(src, path.join(destDir, name), { recursive: true }); else await fs.copyFile(src, path.join(destDir, name)); } } return MiniDb.open({ ...openOpts, dir: destDir }); } // ---- maintenance -------------------------------------------------------- /** Refresh the write lock's timestamp (see {@link LockFile.renew}). No-op * for a read-only instance. Exposed for lease-style holders such as the * cluster shard pool, which renew on a timer to prove liveness. */ async renewLock(): Promise { return renewMiniDbLock(this); } /** Advanced/internal (read-replica owners such as the cluster shard pool): * incrementally apply WAL frames appended to db.wal after `offset` — the * same frames open-time recovery would replay, interpreted identically * (frameToOps: valueMode memory/disk refs, expired-SET drop with LWW, * TYPE_BATCH unrolling, dt meta) — plus incremental maintenance of every * derived index (dt, compound, secondary, text). Unique constraints are * NOT checked: the writer already validated, and intermediate frame states * must apply literally, last-writer-wins. * * The instance tracks its own continuation: the first call must pass * recoveryInfo.walScanEnd, every later call the previous call's returned * offset, and the fs identity of the WAL opened for reading must match the * inode recovery (or the last catch-up) scanned — an offset too old/new, a * rotated file and a shrunken one all return null, meaning: reopen from * scratch. A partial/torn tail left by a writer mid-writev is NOT an * error: the scan stops at the last fully-valid frame; call again later * and its CRC validates once the writev landed. * * Cooperative: the scan runs through the windowed async scanner and the * apply yields between primitive ops on the walApplySlicer budgets, so a * replica that fell far behind does not block the host's event loop in * one synchronous scan+apply. Calls are serialized per instance (see * catchUpChain). */ async catchUpFromWal(offset: number): Promise<{ offset: number; appliedFrames: number } | null> { this.ensureOpen(); const run = this.catchUpChain.then(() => this.doCatchUpFromWal(offset)); // The chain itself must survive a caller's rejection (treated as "reopen // from scratch" by every caller): swallow for the chain, not for the caller. this.catchUpChain = run.then( () => undefined, () => undefined, ); return run; } private async doCatchUpFromWal(offset: number): Promise<{ offset: number; appliedFrames: number } | null> { // Re-check on dequeue: the instance may have been closed while this call // waited on the chain behind another catch-up (both production callers // gate close externally; this is the library-contract backstop). this.ensureOpen(); const ri = this.recoveryInfo; const anchor = this.walTail ?? (ri && ri.walIno ? { dev: ri.walDev, ino: ri.walIno, size: ri.walScanEnd } : null); if (!anchor || offset !== anchor.size) return null; const res = await catchUpWalAsync(this.walPath, offset, anchor, (f, fd, slice) => this.applyRecoveredFrameAsync(f, fd, slice)); if (res) this.walTail = { dev: anchor.dev, ino: anchor.ino, size: res.offset }; return res; } async compact(): Promise { this.ensureOpen(); this.ensureWritable(); await this.submitCompaction(); } /** Stage 6: observable maintenance state (queued/running/publishing plus * the recent failed/complete history). The public read model of the * internal scheduler — callers never see workers or file details. */ maintenanceStatus(): MaintenanceTaskInfo[] { return this.maintenance.status(); } /** The last open()'s lifecycle read model: which path served the open * ('generation-load' + 'wal-catch-up' vs 'full-rebuild'), whether the * instance is 'ready' or still 'degraded' (a deferred text-index base * build in flight), the per-phase wall-clock timings, and how every text * index's base was served (generation image vs worker/inline/staged * rebuild). Diagnostics only — the cumulative counters stay in `stats`. */ lifecycleStatus(): MiniDbLifecycleStatus { return this.lifecycle.snapshot(); } async close(): Promise { return closeMiniDb(this, this.lifecycleHooks); } private ensureOpen(): void { if (this.state !== 'open') throw new Error('MiniDb is closed'); } private ensureWritable(): void { if (this.readOnly) throw new Error('MiniDb is open in read-only mode'); if (this.writeDisabled) throw this.writeDisabledError(); } }