From df6c0201796c399d2b919513be242985080f63d6 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Sun, 19 Jul 2026 15:58:28 +0800 Subject: [PATCH] feat(minidb): switch agent-core-v2 query-store to ClusterDb with 16 shards - ClusterDb: add query() (per-shard fan-out with skip=0 and limit=skip+limit, global re-sort, then skip/limit) and compound index management (create/drop/list through the cluster registry, fanned out to every shard and caught up on shard open) - MiniDbQueryStore: replace the single MiniDb with a 16-shard ClusterDb, so multiple kimi processes share the read model instead of failing with storage.locked; per-shard LockError now propagates as a transient error rather than permanently disabling the read model - corruption: lift openOrRebuild's predicate (SyntaxError / CorruptFrameError) to one process-lifetime wipe-and-reopen rebuild - lower lockAcquireTimeoutMs to 1s for the cache read model - tests: cluster query merge and compound fan-out; store coexistence with a peer instance, corrupt-registry rebuild, 16-shard topology; sessionIndex locked-fallback test now drives a stub IQueryStore --- packages/agent-core-v2/docs/errors.md | 2 +- .../app/sessionIndex/sessionIndexService.ts | 9 +- .../backends/minidb/miniDbQueryStore.ts | 209 +++++++++++------- .../app/sessionIndex/sessionIndex.test.ts | 59 ++--- .../backends/minidb/miniDbQueryStore.test.ts | 71 +++--- packages/minidb/src/cluster/index.ts | 120 +++++++++- packages/minidb/src/cluster/types.ts | 2 + packages/minidb/test/cluster/basic.test.ts | 82 ++++++- 8 files changed, 403 insertions(+), 151 deletions(-) diff --git a/packages/agent-core-v2/docs/errors.md b/packages/agent-core-v2/docs/errors.md index 35e403fef9..e488d07ffe 100644 --- a/packages/agent-core-v2/docs/errors.md +++ b/packages/agent-core-v2/docs/errors.md @@ -70,7 +70,7 @@ The os / persistence / wire domains show the standard shapes: - **`os.fs` (`HostFsError`, `os/interface/hostFsErrors.ts`)** — every `IHostFileSystem` backend translates raw errnos at its boundary via the pure `toHostFsError(err, { path, op })`: `ENOENT→os.fs.not_found`, `EISDIR→os.fs.is_directory`, `ENOTDIR→os.fs.not_directory`, `EEXIST→os.fs.already_exists`, `EACCES/EPERM→os.fs.permission_denied`, `ENOTEMPTY→os.fs.not_empty`, everything else `os.fs.unknown`. `details` carries `{ path, op, errno?, syscall? }`. Documented boolean semantics (e.g. `createExclusive` returning `false` on `EEXIST`) stay booleans, not errors. - **`os.process` (`HostProcessError`, `os/interface/hostProcess.ts`)** — `os.process.spawn_failed` (details `{ command, args?, cwd?, errno? }`) and `os.process.kill_failed`; both carry the raw error as `cause`. Kill keeps its deliberate tolerances: `ESRCH` is a silent no-op, `EPERM` degrades to `child.kill()`. -- **`storage` (`StorageError`, `persistence/interface/storage.ts`)** — `storage.not_found` / `decode_failed` / `corrupted` / `io_failed` / `locked`. ENOENT keeps its established absence semantics (`read → undefined`, `list → []`) and is *not* an error; other I/O failures become `storage.io_failed` (`retryable`). Codec parse failures become `storage.decode_failed` with `{ scope, key, format }`; append-log corruption is `AppendLogCorruptedError` (`storage.corrupted`). A query-store open failure (writer lock held by another process) throws `storage.locked` — consumers (e.g. `FileSessionIndex`) catch it explicitly and fall back to their non-read-model path with a one-time warning; there is no silent no-op degradation. +- **`storage` (`StorageError`, `persistence/interface/storage.ts`)** — `storage.not_found` / `decode_failed` / `corrupted` / `io_failed` / `locked`. ENOENT keeps its established absence semantics (`read → undefined`, `list → []`) and is *not* an error; other I/O failures become `storage.io_failed` (`retryable`). Codec parse failures become `storage.decode_failed` with `{ scope, key, format }`; append-log corruption is `AppendLogCorruptedError` (`storage.corrupted`). `storage.locked` is reserved for a store exclusively held by another process — consumers (e.g. `FileSessionIndex`) catch it explicitly and fall back to their non-read-model path with a one-time warning; there is no silent no-op degradation. (The minidb query-store backend is a multi-process `ClusterDb` and no longer throws it: peers share the store, and per-shard lock contention surfaces as a transient `LockError` instead.) - **`wire` (`WireError`, `wire/errors.ts`)** — `DuplicateOpError` (`wire.duplicate_op`, a build-time bug), `CycleError` (`wire.cycle`, details carry the drain depth and a capped op-type sample), and `wire.unknown_record`: replay skips records whose Op type is absent from `OP_REGISTRY` (compatibility), reports each skip through `onUnexpectedError`, and returns `{ unknownRecords }` so the caller knows the restore was lossy. ## Serialization & boundary translation diff --git a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts index c17047dba3..a6a9b490dd 100644 --- a/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts +++ b/packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts @@ -30,10 +30,11 @@ * backfill on a cold miss. Writes (create / archive / metadata update) keep the * read model warm via `SessionMetadata`; new sessions that have not been * mirrored yet are simply a cold miss and backfilled on first read. The legacy - * N+1 path remains as the flag-off fallback — and as the runtime fallback when - * the query store reports `storage.locked` (another process holds the writer - * lock): the first lock warns once and disables the read model for the rest of - * the process lifetime. + * N+1 path remains as the flag-off fallback — and as the runtime fallback if + * the query store ever reports `storage.locked`: the first lock warns once and + * disables the read model for the rest of the process lifetime. (The minidb + * backend is a multi-process `ClusterDb` and no longer produces that error; + * the wiring stays as defense in depth.) * * This is the local-deployment backend of `ISessionIndex`; a server deployment * would substitute a database-backed `DbSessionIndex`. Bound at App scope. diff --git a/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts b/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts index 26573ecc49..56b70b45ea 100644 --- a/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts +++ b/packages/agent-core-v2/src/persistence/backends/minidb/miniDbQueryStore.ts @@ -1,40 +1,53 @@ /** - * `minidb` backend — `IQueryStore` implementation over `MiniDb`. + * `minidb` backend — `IQueryStore` implementation over `ClusterDb`. + * + * A rebuildable, in-process derived read-model. The store is a `ClusterDb` + * of 16 shards rooted at `/query-store`: keys are hash-routed over + * ordinary `MiniDb` directories, so multiple kimi processes can read and + * write the same read model concurrently (a single writer per shard, readers + * that never take write locks) instead of failing against a database-wide + * single-writer lock. Authoritative data lives in `IAppendLogStore` / + * `IAtomicDocumentStore`, never here, so losing the read model is always + * safe. * - * A rebuildable, in-process derived read-model. `MiniDb` is opened with - * `openOrRebuild`, so on-disk corruption becomes a clean rebuild rather than a - * hard failure: authoritative data lives in `IAppendLogStore` / - * `IAtomicDocumentStore`, never here, so losing the read model is always safe. * Values are JSON (`valueCodec: 'json'`, required by secondary indexes and - * `query`) and held in memory (`valueMode: 'memory'`); durability is `everysec`, - * which is acceptable for a cache. The store is rooted at - * `/query-store`. + * `query`) and held in memory (`valueMode: 'memory'`); durability is + * `everysec`, which is acceptable for a cache. Writes are atomic per shard; + * a `batch` spanning shards is best-effort across them — a projector can + * always replay from its checkpoint. `lockAcquireTimeoutMs` is lowered from + * the 30s default: a cache read must not hang behind a contended shard, and + * with `lockHoldMs` yields one second is ample for a live writer. * - * The database is opened **lazily** on the first actual IO, not at construction. - * Construction therefore does no filesystem work and never touches the single - * writer lock — important because `MiniDbQueryStore` is resolved transitively - * whenever a consumer (e.g. `SessionMetadata`) is constructed, including in - * tests that share a home dir and never read or write the read model. Only a - * real `put`/`get`/`query`/... opens the database. + * The database is opened **lazily** on the first actual IO, not at + * construction. Construction therefore does no filesystem work — important + * because `MiniDbQueryStore` is resolved transitively whenever a consumer + * (e.g. `SessionMetadata`) is constructed, including in tests that share a + * home dir and never read or write the read model. * - * An open failure — typically another kimi process holding the single-writer - * lock on `/query-store` — throws `StorageError(storage.locked)` - * instead of silently degrading to a no-op. The failure is memoized (the - * rejected open promise is cached), so the error is stable for the process - * lifetime and consumers can catch it once and fall back to their - * non-read-model paths. + * Corruption handling lifts `MiniDb.openOrRebuild`'s predicate + * (`SyntaxError` / `CorruptFrameError`) to the cluster: the first + * rebuildable failure triggers one process-lifetime rebuild — close, delete + * the directory, reopen empty, retry the operation once — and consumers' + * checkpoint-based reprojection repopulates the model. Every other error + * propagates as-is; in particular a per-shard `LockError` (a live process + * holding a shard beyond the acquire timeout) is transient and must NOT + * become `storage.locked`, which consumers would treat as a permanent + * read-model outage. * - * A `collection` is encoded as a key prefix (`\u0000`); indexes - * are global to the `MiniDb` instance, so index names are prefixed with the - * collection to keep them isolated, and value indexes are created `sparse` so - * documents from other collections (which lack the indexed field) are skipped. + * A `collection` is encoded as a key prefix (`` + NUL + ``); index + * names are prefixed with the collection to keep them isolated in the + * cluster-wide registry, and value indexes are created `sparse` so documents + * from other collections (which lack the indexed field) are skipped. * * Bound at App scope as a peer of the other access-pattern stores. */ +import { promises as fsp } from 'node:fs'; + import { join } from 'pathe'; -import { MiniDb, type QueryOptions } from '@moonshot-ai/minidb'; +import { type QueryOptions } from '@moonshot-ai/minidb'; +import { ClusterDb } from '@moonshot-ai/minidb/cluster'; import { InstantiationType } from '#/_base/di/extensions'; import { Disposable, toDisposable } from '#/_base/di/lifecycle'; @@ -51,11 +64,12 @@ import { type SortDir, type WriteOp, } from '#/persistence/interface/queryStore'; -import { StorageError, StorageErrors } from '#/persistence/interface/storage'; const SEP = String.fromCodePoint(0); const CHECKPOINT_COLLECTION = '__checkpoint__'; const STORE_SUBDIR = 'query-store'; +const SHARD_COUNT = 16; +const LOCK_ACQUIRE_TIMEOUT_MS = 1000; function physicalKey(collection: string, key: string): string { return `${collection}${SEP}${key}`; @@ -65,11 +79,18 @@ function indexName(collection: string, name: string): string { return `${collection}:${name}`; } +/** The `MiniDb.openOrRebuild` rebuildable predicate: only unrecoverable + * on-disk corruption justifies wiping the read model. */ +function isRebuildable(error: unknown): boolean { + return error instanceof SyntaxError || (error as { name?: string }).name === 'CorruptFrameError'; +} + export class MiniDbQueryStore extends Disposable implements IQueryStore { declare readonly _serviceBrand: undefined; private readonly dir: string; - private dbPromise: Promise | undefined; + private dbPromise: Promise | undefined; + private rebuildPromise: Promise | undefined; private readonly ensuredIndexes = new Set(); constructor( @@ -83,84 +104,112 @@ export class MiniDbQueryStore extends Disposable implements IQueryStore { })); } - private openDb(): Promise { - if (this.dbPromise !== undefined) return this.dbPromise; - this.dbPromise = MiniDb.openOrRebuild( - { - dir: this.dir, - valueCodec: 'json', - valueMode: 'memory', - fsyncPolicy: 'everysec', - }, - { - onRebuild: (err) => { - this.log.warn('minidb query-store rebuilt after corruption', { - dir: this.dir, - error: String(err), - }); - }, - }, - ).catch((error) => { - throw new StorageError( - StorageErrors.codes.STORAGE_LOCKED, - 'minidb query-store is locked by another process', - { details: { dir: this.dir }, cause: error }, - ); - }); + private openDb(): Promise { + // A rebuild wipes and recreates the directory; opens started while one is + // in flight must wait for it instead of racing the rm. + if (this.rebuildPromise !== undefined) return this.openDbAfterRebuild(); + this.dbPromise ??= this.openFresh(); return this.dbPromise; } + private async openDbAfterRebuild(): Promise { + await this.rebuildPromise; + this.dbPromise ??= this.openFresh(); + return this.dbPromise; + } + + private openFresh(): Promise { + return ClusterDb.open({ + dir: this.dir, + shardCount: SHARD_COUNT, + valueCodec: 'json', + valueMode: 'memory', + fsyncPolicy: 'everysec', + lockAcquireTimeoutMs: LOCK_ACQUIRE_TIMEOUT_MS, + }); + } + + /** One process-lifetime rebuild: wipe the corrupt store and let the next + * open start empty. Concurrent callers share the same in-flight rebuild. */ + private rebuild(cause: unknown): Promise { + this.rebuildPromise ??= (async () => { + this.log.warn('minidb query-store rebuilt after corruption', { + dir: this.dir, + error: String(cause), + }); + const previous = this.dbPromise; + // Reset before the first await so a concurrent openDb() waits on + // rebuildPromise instead of reusing the corrupt instance. + this.dbPromise = undefined; + this.ensuredIndexes.clear(); + if (previous !== undefined) { + const db = await previous.catch(() => undefined); + await db?.close().catch(() => {}); + } + await fsp.rm(this.dir, { recursive: true, force: true }); + })(); + return this.rebuildPromise; + } + + private async withDb(op: (db: ClusterDb) => Promise): Promise { + try { + return await op(await this.openDb()); + } catch (error) { + if (!isRebuildable(error)) throw error; + await this.rebuild(error); + // One retry on the fresh store; a second failure propagates as-is. + return op(await this.openDb()); + } + } + async put(collection: string, key: string, value: T): Promise { - const db = await this.openDb(); - await db.set(physicalKey(collection, key), value); + await this.withDb((db) => db.set(physicalKey(collection, key), value)); } async batch(ops: readonly WriteOp[]): Promise { if (ops.length === 0) return; - const db = await this.openDb(); - await db.batch( - ops.map((op) => - op.kind === 'put' - ? { op: 'set' as const, key: physicalKey(op.collection, op.key), value: op.value } - : { op: 'del' as const, key: physicalKey(op.collection, op.key) }, + await this.withDb((db) => + db.batch( + ops.map((op) => + op.kind === 'put' + ? { op: 'set' as const, key: physicalKey(op.collection, op.key), value: op.value } + : { op: 'del' as const, key: physicalKey(op.collection, op.key) }, + ), ), ); } async delete(collection: string, key: string): Promise { - const db = await this.openDb(); - await db.del(physicalKey(collection, key)); + await this.withDb((db) => db.del(physicalKey(collection, key))); } async get(collection: string, key: string): Promise { - const db = await this.openDb(); - return db.get(physicalKey(collection, key)) as T | undefined; + return this.withDb((db) => db.get(physicalKey(collection, key)) as Promise); } query(collection: string): IQuery { - return new MiniDbQuery(() => this.openDb(), collection); + return new MiniDbQuery((op) => this.withDb(op), collection); } async ensureIndex(collection: string, def: IndexDef): Promise { const guard = `${collection}:${def.kind}:${def.name}`; if (this.ensuredIndexes.has(guard)) return; - const db = await this.openDb(); const name = indexName(collection, def.name); - if (def.kind === 'value') { - if (!db.listIndexes().some((i) => i.name === name)) { - await db.createIndex(name, { field: def.field, sparse: true, unique: def.unique }); - } - } else if (def.kind === 'compound') { - if (!db.listCompoundIndexes().some((i) => i.name === name)) { - await db.createCompoundIndex(name, { groupBy: def.groupBy, orderBy: def.orderBy }); - } - } else { + await this.withDb(async (db) => { try { - await db.createTextIndex(name, { fields: def.fields }); + if (def.kind === 'value') { + await db.createIndex(name, { field: def.field, sparse: true, unique: def.unique }); + } else if (def.kind === 'compound') { + await db.createCompoundIndex(name, { groupBy: def.groupBy, orderBy: def.orderBy }); + } else { + await db.createTextIndex(name, { fields: def.fields }); + } } catch (error) { + // A raced ensure (a peer process created it first, or a rebuild + // replayed this call) is a no-op: the definition already exists. if (!(error instanceof Error) || !error.message.includes('already exists')) throw error; } - } + }); this.ensuredIndexes.add(guard); } @@ -173,8 +222,7 @@ export class MiniDbQueryStore extends Disposable implements IQueryStore { } async close(): Promise { - if (this.dbPromise === undefined) return; - const db = await this.dbPromise.catch(() => undefined); + const db = await this.dbPromise?.catch(() => undefined); await db?.close(); } } @@ -187,7 +235,7 @@ class MiniDbQuery implements IQuery { private skip = 0; constructor( - private readonly openDb: () => Promise, + private readonly withDb: (op: (db: ClusterDb) => Promise) => Promise, private readonly collection: string, ) {} @@ -213,7 +261,6 @@ class MiniDbQuery implements IQuery { } async execute(): Promise> { - const db = await this.openDb(); const prefix = `${this.collection}${SEP}`; const q: QueryOptions = { key: { prefix } }; if (Object.keys(this.filter).length > 0) q.filter = this.filter as Record; @@ -222,7 +269,7 @@ class MiniDbQuery implements IQuery { } q.skip = this.skip; if (this.lim !== undefined) q.limit = this.lim + 1; - const rows = db.query(q) as ReadonlyArray<{ key: string; value: T }>; + const rows = (await this.withDb((db) => db.query(q))) as ReadonlyArray<{ key: string; value: T }>; let items = rows.map((r) => r.value); let nextCursor: string | undefined; if (this.lim !== undefined && items.length > this.lim) { diff --git a/packages/agent-core-v2/test/app/sessionIndex/sessionIndex.test.ts b/packages/agent-core-v2/test/app/sessionIndex/sessionIndex.test.ts index 706152d051..540a979f34 100644 --- a/packages/agent-core-v2/test/app/sessionIndex/sessionIndex.test.ts +++ b/packages/agent-core-v2/test/app/sessionIndex/sessionIndex.test.ts @@ -4,8 +4,6 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { MiniDb } from '@moonshot-ai/minidb'; - import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, @@ -24,7 +22,7 @@ import { JsonAtomicDocumentStore } from '#/persistence/backends/node-fs/atomicDo import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore'; import { IQueryStore } from '#/persistence/interface/queryStore'; -import { IFileSystemStorageService } from '#/persistence/interface/storage'; +import { IFileSystemStorageService, StorageError, StorageErrors } from '#/persistence/interface/storage'; import { stubBootstrap } from '../bootstrap/stubs'; import { stubFlag } from '../flag/stubs'; @@ -398,33 +396,36 @@ describe('FileSessionIndex (read model)', () => { it('falls back to the legacy disk path when the query store is locked', async () => { await seedSession('active', { title: 'from disk', createdAt: 1, updatedAt: 2 }); - const lockHolder = await MiniDb.open({ - dir: join(homeDir, 'cache', 'query-store'), - valueCodec: 'json', - }); + // The minidb cluster backend shares the store across processes and no + // longer produces storage.locked itself; stub it here so the + // disable-and-fall-back wiring stays under test. + const locked = new StorageError(StorageErrors.codes.STORAGE_LOCKED, 'locked by test'); + const lockedStore: IQueryStore = { + ...stubQueryStore(), + ensureIndex: async () => { throw locked; }, + get: async () => { throw locked; }, + query: () => { throw locked; }, + }; const warnings: string[] = []; const log = { ...stubLog(), warn: (msg: string) => { warnings.push(msg); } }; - try { - const fileStorage = new FileStorageService(homeDir); - const host = createScopedTestHost([ - stubPair(IFileSystemStorageService, fileStorage), - stubPair(IAtomicDocumentStore, new JsonAtomicDocumentStore(fileStorage)), - stubPair(IBootstrapService, stubBootstrap(homeDir)), - stubPair(ILogService, log), - stubPair(IFlagService, stubFlag(true)), - ]); - disposeHost = () => { host.dispose(); }; - const store = host.app.accessor.get(ISessionIndex); - // The read model throws storage.locked; the index serves from disk. - const page = await store.list({ workspaceIds: [workspaceId] }); - expect(page.items.map((s) => s.id)).toEqual(['active']); - expect(page.items[0]?.title).toBe('from disk'); - expect(await store.get('active')).toMatchObject({ id: 'active', title: 'from disk' }); - expect(await store.countActive([workspaceId])).toBe(1); - // The lock is warned about once, then the read model stays disabled. - expect(warnings).toEqual(['query-store locked by another process; disabling read model']); - } finally { - await lockHolder.close(); - } + const fileStorage = new FileStorageService(homeDir); + const host = createScopedTestHost([ + stubPair(IFileSystemStorageService, fileStorage), + stubPair(IAtomicDocumentStore, new JsonAtomicDocumentStore(fileStorage)), + stubPair(IBootstrapService, stubBootstrap(homeDir)), + stubPair(IQueryStore, lockedStore), + stubPair(ILogService, log), + stubPair(IFlagService, stubFlag(true)), + ]); + disposeHost = () => { host.dispose(); }; + const store = host.app.accessor.get(ISessionIndex); + // The read model throws storage.locked; the index serves from disk. + const page = await store.list({ workspaceIds: [workspaceId] }); + expect(page.items.map((s) => s.id)).toEqual(['active']); + expect(page.items[0]?.title).toBe('from disk'); + expect(await store.get('active')).toMatchObject({ id: 'active', title: 'from disk' }); + expect(await store.countActive([workspaceId])).toBe(1); + // The lock is warned about once, then the read model stays disabled. + expect(warnings).toEqual(['query-store locked by another process; disabling read model']); }); }); diff --git a/packages/agent-core-v2/test/persistence/backends/minidb/miniDbQueryStore.test.ts b/packages/agent-core-v2/test/persistence/backends/minidb/miniDbQueryStore.test.ts index fe02ef2ee3..1f92a76e93 100644 --- a/packages/agent-core-v2/test/persistence/backends/minidb/miniDbQueryStore.test.ts +++ b/packages/agent-core-v2/test/persistence/backends/minidb/miniDbQueryStore.test.ts @@ -9,13 +9,14 @@ import { LifecycleScope, _clearScopedRegistryForTests, registerScopedService } f import { createScopedTestHost, stubPair } from '#/_base/di/test'; import { ILogService } from '#/_base/log/log'; import { IBootstrapService } from '#/app/bootstrap/bootstrap'; -import { MiniDb } from '@moonshot-ai/minidb'; +import { ClusterDb } from '@moonshot-ai/minidb/cluster'; import { MiniDbQueryStore } from '#/persistence/backends/minidb/miniDbQueryStore'; import { IQueryStore } from '#/persistence/interface/queryStore'; import { stubBootstrap } from '../../../app/bootstrap/stubs'; import { stubLog } from '../../../_base/log/stubs'; const COLLECTION = 'session'; +const SEP = String.fromCodePoint(0); describe('MiniDbQueryStore', () => { let homeDir: string; @@ -132,36 +133,26 @@ describe('MiniDbQueryStore', () => { expect(await store.getCheckpoint('wire:abc')).toEqual({ seq: 42 }); }); - it('throws storage.locked when the database lock is held by another process', async () => { + it('shares the store with a second cluster instance instead of locking it out', async () => { const storeDir = join(homeDir, 'cache', 'query-store'); - const lockHolder = await MiniDb.open({ dir: storeDir, valueCodec: 'json' }); + // A peer instance stands in for another kimi process: it has its own + // lock pool, so write locks are genuinely contended between the two. + const peer = await ClusterDb.open({ dir: storeDir, shardCount: 16, valueCodec: 'json' }); try { const store = build(); - await expect(store.put(COLLECTION, 'a', { id: 'a' })).rejects.toMatchObject({ - code: 'storage.locked', - }); - await expect( - store.batch([{ kind: 'put', collection: COLLECTION, key: 'b', value: { id: 'b' } }]), - ).rejects.toMatchObject({ code: 'storage.locked' }); - await expect( - store.ensureIndex(COLLECTION, { kind: 'value', name: 'byId', field: 'id' }), - ).rejects.toMatchObject({ code: 'storage.locked' }); - await expect(store.get(COLLECTION, 'a')).rejects.toMatchObject({ - code: 'storage.locked', - }); - await expect(store.getCheckpoint('wire:abc')).rejects.toMatchObject({ - code: 'storage.locked', - }); - await expect(store.query(COLLECTION).execute()).rejects.toMatchObject({ - code: 'storage.locked', - }); - await expect(store.close()).resolves.toBeUndefined(); + // Writes from the peer are visible here, and vice versa — the + // database-wide single-writer lockout (storage.locked) is gone. + await peer.set(`${COLLECTION}${SEP}peer`, { id: 'peer', v: 1 }); + expect(await store.get(COLLECTION, 'peer')).toEqual({ id: 'peer', v: 1 }); + await store.put(COLLECTION, 'mine', { id: 'mine', v: 2 }); + expect(await peer.get(`${COLLECTION}${SEP}mine`)).toEqual({ id: 'mine', v: 2 }); + await store.close(); } finally { - await lockHolder.close(); + await peer.close(); } }); - it('preserves data and drops a corrupt index sidecar on reopen', async () => { + it('wipes and rebuilds the store after the cluster registry is corrupted', async () => { const first = build(); await first.put(COLLECTION, 'a', { id: 'a', v: 1 }); await first.ensureIndex(COLLECTION, { kind: 'value', name: 'byV', field: 'v' }); @@ -169,16 +160,32 @@ describe('MiniDbQueryStore', () => { disposeHost?.(); disposeHost = undefined; - // A corrupt index-definition sidecar holds only derived metadata. The - // opener must not wipe the database over it: the sidecar is dropped, the - // data is preserved, and the caller can re-register the definition. - const indexFile = join(homeDir, 'cache', 'query-store', 'db.indexes.json'); - await fsp.writeFile(indexFile, '{ definitely not valid json'); + // A corrupt cluster registry surfaces as a SyntaxError on the next index + // op. The store answers with one process-lifetime rebuild: the directory + // is wiped (the read model is derivable, so data is NOT preserved) and + // the retried op succeeds against the fresh cluster. + const registryFile = join(homeDir, 'cache', 'query-store', 'cluster.indexes.json'); + await fsp.writeFile(registryFile, '{ definitely not valid json'); const second = build(); - expect(await second.get(COLLECTION, 'a')).toEqual({ id: 'a', v: 1 }); await second.ensureIndex(COLLECTION, { kind: 'value', name: 'byV', field: 'v' }); - const page = await second.query<{ id: string; v: number }>(COLLECTION).where({ v: 1 }).execute(); - expect(page.items).toEqual([{ id: 'a', v: 1 }]); + expect(await second.get(COLLECTION, 'a')).toBeUndefined(); + await second.put(COLLECTION, 'b', { id: 'b', v: 2 }); + const page = await second.query<{ id: string; v: number }>(COLLECTION).where({ v: 2 }).execute(); + expect(page.items).toEqual([{ id: 'b', v: 2 }]); + }); + + it('opens a 16-shard cluster under the cache dir', async () => { + const store = build(); + await store.put(COLLECTION, 'a', { id: 'a' }); + const storeDir = join(homeDir, 'cache', 'query-store'); + const meta = JSON.parse(await fsp.readFile(join(storeDir, 'cluster.meta.json'), 'utf8')) as { + shardCount: number; + }; + expect(meta.shardCount).toBe(16); + const entries = await fsp.readdir(storeDir); + for (let i = 0; i < 16; i++) { + expect(entries).toContain(`shard-${String(i).padStart(2, '0')}`); + } }); }); diff --git a/packages/minidb/src/cluster/index.ts b/packages/minidb/src/cluster/index.ts index c16b867dd4..9d0ff3f1fe 100644 --- a/packages/minidb/src/cluster/index.ts +++ b/packages/minidb/src/cluster/index.ts @@ -16,6 +16,9 @@ // - Index definitions live in a cluster-wide registry (cluster.indexes.json) // as the source of truth; every shard writer applies missing definitions // right after opening, and create/drop operations fan out to all shards. +// - query() fans MiniDb's unified query out to every shard (each shard runs +// it with skip=0 and limit=skip+limit, index pruning included), re-sorts +// the merged result globally, and applies skip/limit at the end. // // Consistency: single-key and same-shard batch ops are strongly consistent // (single writer per shard, atomic WAL frames). Cross-shard mset/mdel/batch @@ -25,8 +28,10 @@ import fs from 'node:fs/promises'; import path from 'node:path'; -import type { BatchInputOp, IndexDef, IndexInfo, MiniDb, ScanEntry, SetOptions } from '../index.js'; +import type { BatchInputOp, IndexDef, IndexInfo, MiniDb, QueryOptions, ScanEntry, SetOptions } from '../index.js'; +import type { CompoundIndexDef, CompoundIndexInfo } from '../compound-index.js'; import { LockError } from '../lockfile.js'; +import { getPath } from '../query.js'; import { Coordinator } from './coordinator.js'; import { ShardLockPool } from './lock-pool.js'; import { Router } from './router.js'; @@ -112,6 +117,9 @@ export class ClusterDb { for (const { name, def } of reg.indexes) { if (!db.listIndexes().some((i) => i.name === name)) await db.createIndex(name, def); } + for (const { name, def } of reg.compoundIndexes) { + if (!db.listCompoundIndexes().some((i) => i.name === name)) await db.createCompoundIndex(name, def); + } for (const { name, fields } of reg.textIndexes) { try { await db.createTextIndex(name, { fields: fields ?? undefined }); @@ -255,14 +263,50 @@ export class ClusterDb { return this.scan({ prefix: p, limit }); } + /** Merged query over all shards. Every shard runs the full query locally + * (index-assisted candidate pruning included) with skip=0 and a limit of + * skip+limit — the global top-(skip+limit) under any total order is + * contained in each shard's local top-(skip+limit). The merged result is + * then re-sorted globally (by the explicit sort, else by key bytes to + * match scan's global order) and skip/limit applies at the end. Text + * scores are computed per shard (per-shard idf), so a text query's + * global ranking is approximate; use search() when the score matters. */ + async query(q: QueryOptions = {}): Promise[]> { + this.ensureOpen(); + const skip = q.skip ?? 0; + const limit = q.limit === undefined ? Infinity : q.limit; + const needed = skip + limit; + const all: ScanEntry[] = []; + for (const id of this.router.shardIds()) { + const rows = await this.reader(id, (db) => db.query({ ...q, skip: 0, limit: needed })); + for (const r of rows) all.push(r); + } + if (q.sort) { + // The same comparator MiniDb.query applies, re-run on the merged set. + const entries = Object.entries(q.sort); + all.sort((a, b) => { + for (const [p, dir] of entries) { + const av = getPath(a.value, p) as number | string; + const bv = getPath(b.value, p) as number | string; + const c = av < bv ? -1 : av > bv ? 1 : 0; + if (c !== 0) return dir < 0 ? -c : c; + } + return 0; + }); + } else { + all.sort(compareEntries); + } + return skip > 0 || limit !== Infinity ? all.slice(skip, skip + limit) : all; + } + // ---- secondary indexes ------------------------------------------------------ private static async loadRegistry(file: string): Promise { try { const raw = JSON.parse(await fs.readFile(file, 'utf8')) as Partial; - return { indexes: raw.indexes ?? [], textIndexes: raw.textIndexes ?? [] }; + return { indexes: raw.indexes ?? [], compoundIndexes: raw.compoundIndexes ?? [], textIndexes: raw.textIndexes ?? [] }; } catch (e) { - if ((e as NodeJS.ErrnoException).code === 'ENOENT') return { indexes: [], textIndexes: [] }; + if ((e as NodeJS.ErrnoException).code === 'ENOENT') return { indexes: [], compoundIndexes: [], textIndexes: [] }; throw e; } } @@ -469,6 +513,76 @@ export class ClusterDb { if (!reg.indexes.some((i) => i.name === name)) throw new Error(`no such index: ${name}`); } + // ---- compound indexes (groupBy + orderBy) ------------------------------------ + + private static sameCompoundIndexDef(a: CompoundIndexDef, b: CompoundIndexDef): boolean { + return a.groupBy === b.groupBy && a.orderBy === b.orderBy && (a.orderType ?? 'number') === (b.orderType ?? 'number'); + } + + /** Create a compound index on every shard and record it in the cluster + * registry, with the same fan-out / rollback / catch-up model as + * createIndex. Definition management only: a merged compoundRange query + * is a follow-up for when a consumer needs one. */ + async createCompoundIndex(name: string, def: CompoundIndexDef): Promise { + this.ensureOpen(); + this.requireJsonCodec('compound indexes'); + const reg = await ClusterDb.loadRegistry(this.indexPath); + if (reg.compoundIndexes.some((i) => i.name === name)) throw new Error(`compound index "${name}" already exists`); + const createdOn: number[] = []; + try { + await this.forEachShardWriter(async (db, shardId) => { + if (!db.listCompoundIndexes().some((i) => i.name === name)) { + await db.createCompoundIndex(name, def); + createdOn.push(shardId); + } + }); + } catch (e) { + // Roll back the partial fan-out (see createIndex). + await this.rollbackShards(createdOn, async (db) => { + await db.dropCompoundIndex(name); + }); + throw e; + } + await this.mutateRegistry((current) => { + const existing = current.compoundIndexes.find((i) => i.name === name); + if (existing) { + // A raced create of the same definition already published. + if (ClusterDb.sameCompoundIndexDef(existing.def, def)) return false; + throw new Error(`compound index "${name}" already exists`); + } + current.compoundIndexes.push({ name, def }); + return true; + }); + } + + async dropCompoundIndex(name: string): Promise { + this.ensureOpen(); + const reg = await ClusterDb.loadRegistry(this.indexPath); + const existed = reg.compoundIndexes.some((i) => i.name === name); + await this.forEachShardWriter(async (db) => { + if (db.listCompoundIndexes().some((i) => i.name === name)) await db.dropCompoundIndex(name); + }); + if (!existed) return false; + await this.mutateRegistry((current) => { + if (!current.compoundIndexes.some((i) => i.name === name)) return false; + current.compoundIndexes = current.compoundIndexes.filter((i) => i.name !== name); + return true; + }); + return true; + } + + /** Cluster-wide compound index definitions from the registry (source of truth). */ + async listCompoundIndexes(): Promise { + this.ensureOpen(); + const reg = await ClusterDb.loadRegistry(this.indexPath); + return reg.compoundIndexes.map(({ name, def }) => ({ + name, + groupBy: def.groupBy, + orderBy: def.orderBy, + orderType: def.orderType ?? 'number', + })); + } + // ---- full-text search ------------------------------------------------------- async createTextIndex(name: string, opts: { fields?: readonly string[] } = {}): Promise { diff --git a/packages/minidb/src/cluster/types.ts b/packages/minidb/src/cluster/types.ts index 9897be641c..9c37716a00 100644 --- a/packages/minidb/src/cluster/types.ts +++ b/packages/minidb/src/cluster/types.ts @@ -5,6 +5,7 @@ import type { ValueCodecName } from '../index.js'; import type { IndexDef } from '../index-manager.js'; +import type { CompoundIndexDef } from '../compound-index.js'; import type { FsyncPolicy } from '../wal.js'; /** Cross-shard write semantics. @@ -89,6 +90,7 @@ export interface ScanOptions { * when an index was created catches up later. */ export interface ClusterIndexRegistry { indexes: { name: string; def: IndexDef }[]; + compoundIndexes: { name: string; def: CompoundIndexDef }[]; textIndexes: { name: string; fields: readonly string[] | null }[]; } diff --git a/packages/minidb/test/cluster/basic.test.ts b/packages/minidb/test/cluster/basic.test.ts index 2130b77294..46df602967 100644 --- a/packages/minidb/test/cluster/basic.test.ts +++ b/packages/minidb/test/cluster/basic.test.ts @@ -7,7 +7,8 @@ import { test } from 'vitest'; import assert from 'node:assert/strict'; import fs from 'node:fs/promises'; import path from 'node:path'; -import { ClusterDb } from '../../src/cluster/index.js'; +import { MiniDb } from '../../src/index.js'; +import { ClusterDb, shardDirName } from '../../src/cluster/index.js'; import { shardFor } from '../../src/cluster/utils.js'; import { tmpDir, rmrf } from '../e2e/helpers/tmp.js'; import { keyOnShard, keysByShard } from './helpers.js'; @@ -220,3 +221,82 @@ test('stats reflect writer usage', async () => { await rmrf(dir); } }); + +interface QueryDoc { + g: string; + n: number; +} + +test('query() merges filter + sort + skip/limit across shards', async () => { + const dir = await tmpDir('minidb-cluster-'); + try { + const db = await ClusterDb.open({ dir, shardCount: 4, valueCodec: 'json' }); + // The four globally-highest n values all sit on shard 0: a per-shard + // fetch must take skip+limit rows or a global page would lose them. + for (const [i, n] of [100, 101, 102, 103].entries()) { + const key = keyOnShard(`high${i}`, 0, 4); + await db.set(key, { g: 'x', n }); + } + // 40 filler docs with lower n, hash-scattered over all shards. + const filler: string[] = []; + for (let i = 0; i < 40; i++) { + filler.push(`q:${i}`); + await db.set(`q:${i}`, { g: 'x', n: i }); + } + + const sorted = (rows: { value: QueryDoc | undefined }[]) => rows.map((e) => e.value!.n); + const page1 = await db.query({ filter: { g: 'x' }, sort: { n: -1 }, limit: 2 }); + assert.deepEqual(sorted(page1), [103, 102]); + const page2 = await db.query({ filter: { g: 'x' }, sort: { n: -1 }, skip: 2, limit: 2 }); + assert.deepEqual(sorted(page2), [101, 100]); + const page3 = await db.query({ filter: { g: 'x' }, sort: { n: -1 }, skip: 4, limit: 2 }); + assert.deepEqual(sorted(page3), [39, 38]); + + // Without an explicit sort the global order is key bytes (as in scan). + const unordered = await db.query({ key: { prefix: 'q:' } }); + assert.deepEqual( + unordered.map((e) => e.key), + filler.toSorted(), + ); + + // An unfiltered, unbounded query sees every doc exactly once. + const all = await db.query(); + assert.equal(all.length, 44); + await db.close(); + } finally { + await rmrf(dir); + } +}); + +test('compound index definitions fan out to all shards and survive reopen', async () => { + const dir = await tmpDir('minidb-cluster-'); + try { + const db = await ClusterDb.open({ dir, shardCount: 4, valueCodec: 'json' }); + await db.set(keyOnShard('ci', 0, 4), { g: 'x', n: 1 }); + await db.createCompoundIndex('byGN', { groupBy: 'g', orderBy: 'n' }); + const info = [{ name: 'byGN', groupBy: 'g', orderBy: 'n', orderType: 'number' }]; + assert.deepEqual(await db.listCompoundIndexes(), info); + + // Every shard applied the definition (verified through its own sidecar). + for (let id = 0; id < 4; id++) { + const shard = await MiniDb.open({ dir: path.join(dir, shardDirName(id, 4)), valueCodec: 'json', readOnly: true }); + try { + assert.ok(shard.listCompoundIndexes().some((i) => i.name === 'byGN'), `shard ${id} has the index`); + } finally { + await shard.close(); + } + } + + // A duplicate create is rejected, and the registry round-trips a reopen. + await assert.rejects(() => db.createCompoundIndex('byGN', { groupBy: 'g', orderBy: 'n' }), /already exists/); + await db.close(); + const db2 = await ClusterDb.open({ dir, shardCount: 4, valueCodec: 'json' }); + assert.deepEqual(await db2.listCompoundIndexes(), info); + assert.equal(await db2.dropCompoundIndex('byGN'), true); + assert.deepEqual(await db2.listCompoundIndexes(), []); + assert.equal(await db2.dropCompoundIndex('byGN'), false); + await db2.close(); + } finally { + await rmrf(dir); + } +});