feat(minidb): ClusterDb sharding, incremental reader catch-up, and engine hardening

.changeset/minidb-durability-hardening.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/minidb": patch
+"@moonshot-ai/kimi-code": patch
+---
+
+Harden the embedded key-value engine's durability: WAL compaction now always terminates under sustained write storms instead of chasing the tail forever, a committed write can no longer slip through a compaction rotation undetected, torn WAL tails no longer misplace later disk-mode value pointers, read-only opens never create or modify database files or compact under a live writer, corrupt index-definition files no longer force a full rebuild, stale compaction temp files are cleaned on open, and the process lock can no longer be taken over by several processes at once.

.changeset/minidb-perf-hardening.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/minidb": patch
+"@moonshot-ai/kimi-code": patch
+---
+
+Speed up the embedded key-value engine under stress: queries with skip/limit now stream candidates instead of decoding every match first, LRU eviction picks victims in O(1) instead of scanning every key, bursts of simultaneously expired TTL keys are drained within seconds, existence checks and size counting no longer read values when they only need metadata, and one oversized token can no longer poison the full-text index.

.changeset/minidb-reader-catchup.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/minidb": minor
+"@moonshot-ai/kimi-code": patch
+---
+
+Cluster readers of the embedded key-value engine now catch up incrementally by replaying only newly appended WAL frames after another process writes, instead of fully reopening the shard on every read; cross-process read latency drops by orders of magnitude at larger shard sizes, and readers still fall back to a full reopen after WAL rotation or truncation.

.changeset/minidb-review-hardening.md

@@ -0,0 +1,6 @@
+---
+"@moonshot-ai/minidb": patch
+"@moonshot-ai/kimi-code": patch
+---
+
+Keep the embedded key-value engine writable when a WAL compaction rotation fails mid-way instead of wedging it until reopen, stop a rolled-back write from erasing a concurrently committed value for the same key, let the RESP server survive aborted connections, recover after oversized requests, and answer each pipelined command independently, and keep the previous full-text index intact when a postings rebuild fails.

packages/agent-core-v2/test/persistence/backends/minidb/miniDbQueryStore.test.ts

@@ -161,18 +161,24 @@ describe('MiniDbQueryStore', () => {
     }
   });
 
-  it('rebuilds a clean store after on-disk corruption', async () => {
+  it('preserves data and drops a corrupt index sidecar on reopen', async () => {
     const first = build();
-    await first.put(COLLECTION, 'a', { v: 1 });
+    await first.put(COLLECTION, 'a', { id: 'a', v: 1 });
     await first.ensureIndex(COLLECTION, { kind: 'value', name: 'byV', field: 'v' });
     await first.close();
     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');
 
     const second = build();
-    expect(await second.get(COLLECTION, 'a')).toBeUndefined();
+    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 }]);
   });
 });

packages/minidb/DESIGN_NOTES.md

@@ -111,7 +111,8 @@ See `src/wal.ts`.
 Borrowed from Redis RDB/`BGREWRITEAOF` and Bitcask's merge. Compaction is
 **non-blocking for writes**: the live WAL doubles as a Redis-style rewrite
 buffer (`aof_rewrite_buf`), so the (slow) snapshot is written while writers
-keep appending, and they pause only for a brief final rotation.
+keep appending, and they pause only for the final rotation — bounded, but the
+pause scales with the remaining tail when a write storm outruns the pre-copy.
 
 - When the WAL exceeds a size threshold, compact (rewrite state):
   1. **Fence.** Flush the WAL and record `baseOffset = wal.size`. Every write
@@ -124,28 +125,40 @@ keep appending, and they pause only for a brief final rotation.
      store while we iterate. The snapshot need not be point-in-time — the WAL
      tail below repairs any fuzziness on replay.
   3. **Pre-copy the tail.** Stream `WAL[baseOffset .. head]` into `db.wal.tmp`,
-     looping until the remaining delta is small. Also non-blocking: post-fence
+     while the copy is CONVERGING: each pass must shrink the remaining delta
+     meaningfully, bounded by a small pass cap. Also non-blocking: post-fence
      writes accumulate in the live WAL instead of a separate in-memory buffer.
-  4. **Rotate** (the only blocking phase, and brief): set `_rotateLock` so new
-     writers park, flush, copy the tiny remaining tail delta, close the old
-     WAL, `rename` the snapshot into place, `rename` the new WAL into place,
-     and reopen it. The two renames are ordered **snapshot-first** so a crash
-     between them pairs the new snapshot with the old full WAL — replaying the
-     whole old WAL on top of the new snapshot is idempotent for pre-fence
-     frames and correct for post-fence frames, so the state stays consistent.
-     Reversing the order would pair an old snapshot with a truncated new WAL
-     and lose pre-fence data.
+     Under sustained writes whose append rate approaches the copy rate the gap
+     stops shrinking, and looping until it was small enough would never
+     terminate — observed in stress testing as compactions stalling for as long
+     as a write storm lasted (WAL unbounded, disk amplification ~20x). Giving
+     up to the rotation phase is what keeps compaction live in that regime.
+  4. **Rotate** (the only blocking phase): set `_rotateLock` so new writers
+     park, then **seal** the old WAL — post-seal appends fail fast with a
+     retryable error and the op re-runs against the new WAL — flush, and copy
+     the remaining tail in a drain loop. With the WAL sealed the head no
+     longer moves, so the loop provably terminates; the pause scales with the
+     tail the pre-copy could not drain (the same bounded end-of-rewrite pause
+     Redis accepts for its AOF diff flush). Finish by `rename`ing the snapshot
+     into place, `rename`ing the new WAL into place, and reopening it. The two
+     renames are ordered **snapshot-first** so a crash between them pairs the
+     new snapshot with the old full WAL — replaying the whole old WAL on top
+     of the new snapshot is idempotent for pre-fence frames and correct for
+     post-fence frames, so the state stays consistent. Reversing the order
+     would pair an old snapshot with a truncated new WAL and lose pre-fence
+     data.
 - Reads are unaffected throughout. Writes pause only for the rotation critical
-  section (one flush, a small copy, two renames + two `fsyncDir`s); the bulk
-  snapshot + tail copy happen concurrently with writes. Recovery is always
-  `load snapshot + replay WAL`, last-writer-wins, which is exactly what makes
-  the non-point-in-time snapshot converge to the correct latest state.
-- A writer's gate check (`if (_rotateLock) await _rotateLock`) and its
-  `wal.append()` are in the same synchronous segment, and `_rotateLock` is set
-  synchronously before the rotation flush — so a writer either enqueued its
-  frame before the lock (and is drained by the flush) or parks on the lock
-  without appending. The event loop cannot interleave between the two, which
-  is why the critical section is quiescent after the flush and cannot deadlock.
+  section (one flush, the remaining tail copy, two renames + two `fsyncDir`s);
+  the bulk snapshot + tail copy happen concurrently with writes. Recovery is
+  always `load snapshot + replay WAL`, last-writer-wins, which is exactly what
+  makes the non-point-in-time snapshot converge to the correct latest state.
+- The rotation cannot slip a committed write: a writer's gate check
+  (`if (_rotateLock) await _rotateLock`) and its `wal.append()` are NOT in the
+  same synchronous segment (memory-budget checks yield microtasks in between),
+  so an in-flight op could previously append between the final flush and the
+  close and lose a resolved write into the abandoned old file. Sealing the old
+  WAL before the final flush closes that hole: such an op fails fast against
+  the sealed WAL and transparently retries against the new one.
 - The snapshot encoder runs on the main thread but **yields to the event loop
   every N entries** (chunked), so large snapshots stay responsive without a
   Worker. Offloading the encode to a `worker_thread` is the planned

packages/minidb/README.md

@@ -26,6 +26,8 @@ Written in **TypeScript**, with strict types and zero runtime dependencies.
   backed by a Redis-style **skip list** with O(log N) rank/range; unique and
   sparse supported; array fields indexed per element.
 - **RESP server** (optional): talk to it with `redis-cli` / `ioredis`.
+- **ClusterDb** (optional sharding layer): multi-process concurrent read/write
+  over N sharded MiniDb directories, scaling write throughput with shard count.
 - **Value codecs**: `buffer`, `string`, or `json`.
 - **Zero dependencies**, pure `node:*` built-ins.
 
@@ -261,6 +263,14 @@ bound the metadata/index footprint; it does not delete value bytes from disk
 because values are already stored outside RAM. `db.stats.evictions` and
 `db.stats.maxMemoryRejections` track the behavior.
 
+The budget is tracked in **approximate logical bytes** (key + value + dt
+metadata), not real JS heap usage. Actual per-key overhead is higher — on the
+order of several hundred bytes per key for the Map entry, the ordered index
+node, and bookkeeping — so a database of many small keys needs far more RAM
+than `store.bytes` suggests. (Measured: ~0.6 KB of heap per key for tiny
+records, so 1M small keys ≈ 600 MB of heap.) For millions of small keys,
+prefer `valueMode: 'disk'` and/or budget accordingly.
+
 ### Online backup / restore
 
 ```js
@@ -350,6 +360,14 @@ sorting in memory. On 10k sessions in one workspace, paginating with
 `compoundRange` is ~20–40× faster than "fetch-all + sort", and stays
 sub-millisecond at any offset.
 
+Multi-process `ClusterDb` scaling across process counts × shard counts (the
+headline: shard-affinity writes scale ~linearly until the single-shard speed
+is reached; uniform cross-shard traffic pays lock handoffs):
+
+```bash
+pnpm bench:cluster   # bench/cluster.ts — spawns real writer/reader processes
+```
+
 ## Testing
 
 Three layers of tests, run with the built-in `node:test` runner (no deps):
@@ -378,6 +396,13 @@ long-run behavior:
 | `boundary.test.js` | key-length limits, large values, many keys, empty db |
 | `soak.test.js` | sustained ops + heap stability (opt-in: `SOAK=30 npm run test:e2e`) |
 
+The **cluster suite** (`test/cluster/*.test.ts`) covers the `ClusterDb`
+sharding layer: topology/routing, merged scans, lock contention and lease
+renewal in-process, cross-shard indexes/compaction, true multi-process
+scenarios (concurrent writers on disjoint and shared shards, live cross-process
+read visibility, read/write storms) and crash takeovers (`kill -9` → contiguous
+recovery + stale-lock handoff).
+
 ## Design in one paragraph
 
 Log-structured engine: all writes append to a CRC-framed **WAL** (group
@@ -414,6 +439,49 @@ const r = await MiniDb.open({ dir: './data', readOnly: true });
 For many clients, run the **RESP server** (single minidb process, many TCP
 clients) — that is the intended concurrent-access model, like Redis.
 
+### ClusterDb: multi-process sharding
+
+When several **processes** must read and write the same logical database
+without a server, `ClusterDb` shards the key space over N ordinary minidb
+directories (each with its own WAL, snapshot, and `db.lock`):
+
+```js
+import { ClusterDb } from '@moonshot-ai/minidb/cluster';
+
+const db = await ClusterDb.open({ dir: './data', shardCount: 16, valueCodec: 'json' });
+await db.set('user:1', { name: 'alice' });      // routed by hash to one shard
+await db.mset([['a:1', v1], ['b:2', v2]]);      // grouped per shard, atomic per shard
+const all = await db.scan({ prefix: 'user:' }); // merged over all shards
+await db.close();
+```
+
+- **Writes** never meet a single global writer: each process acquires shard
+  write locks on demand (with retry up to `lockAcquireTimeoutMs`), caches them
+  briefly (`lockPoolMaxShards`, LRU), and yields a held shard after
+  `lockHoldMs` (default 250ms) so other processes are never starved. Throughput
+  scales with the number of distinct shards being written — up to the
+  single-shard speed of MiniDb per shard.
+- **Reads** never take locks: they use the cached writer when the local
+  process holds the shard, else a read-only instance that is revalidated
+  against the shard files on every use, so a read that starts after another
+  process's commit always observes it.
+- **Consistency**: single-key and same-shard batch ops are strongly
+  consistent (single writer per shard, atomic WAL frames). Cross-shard
+  `mset`/`mdel`/`batch` are best-effort (atomic per shard, not globally;
+  `crossShard: 'none'` rejects them instead). Scans merge per-shard snapshots,
+  so entries from different shards may reflect different points in time.
+- **Indexes** (`createIndex`/`createTextIndex`) are recorded in a cluster-wide
+  registry and applied by every shard writer on open; `findEq`/`findRange`/
+  `search` merge per-shard results (text scores are per-shard). Index
+  management acquires every shard writer — run it from one process, off the
+  hot path.
+- Crash recovery is per shard: a lock left by a dead PID is taken over by the
+  next opener, exactly like single MiniDb.
+
+`crossShard: '2pc'` is reserved for a future two-phase commit and is rejected
+today. Performance numbers across process/shard counts: run
+`pnpm bench:cluster` (see `bench/cluster.ts`).
+
 For a **rebuildable cache**, use `openOrRebuild`: a corrupt cache is discarded
 and reopened empty, while a live-locked db is never destroyed:
 
@@ -435,11 +503,20 @@ const db = await MiniDb.openOrRebuild(
   cold, very large postings list can briefly block the event loop.
 - Compaction is **non-blocking for writes**: the WAL itself acts as a
   `BGREWRITEAOF`-style rewrite buffer, so the (slow) snapshot is written while
-  writers keep appending. Writes pause only for a brief final rotation (a
-  flush, a small tail copy, and two atomic renames).
+  writers keep appending. Writes pause only for the final rotation (a flush, a
+  tail copy, and two atomic renames). A pre-copy drains most of the tail
+  beforehand when writes are slow enough; under sustained writes that outrun
+  the pre-copy, the rotation absorbs a larger tail — the same bounded
+  end-of-rewrite pause Redis accepts for its AOF diff flush — so compaction
+  always terminates. Mid-compaction crashes leave `db.*.tmp` files behind,
+  which the next writer open removes automatically.
 - Snapshot encoding runs on the main thread (chunked + yielding); offloading to
   a `worker_thread` is a planned optimization.
-- Single process / single writer.
+- Single minidb directory = single process / single writer. For multi-process
+  access use `ClusterDb` (above): sharding scales writes, but hash routing
+  means whole-range scans fan out to all shards, and uniform cross-shard write
+  patterns pay shard-lock handoff costs (`lockHoldMs` per yield) — workloads
+  with per-process shard affinity scale best.
 
 ## Credits
 

packages/minidb/bench/cluster-worker.ts

@@ -0,0 +1,91 @@
+// bench/cluster-worker.ts
+//
+// Worker process for the cluster concurrency benchmark. Spawned by
+// bench/cluster.ts; prints one JSON report line and exits.
+//
+//   write <dir> <shards> <prefix> <n> <valueBytes> <lockHoldMs> <allow>
+//     Writes n keys `${prefix}:${i}` (json values). `allow` is either 'all'
+//     or a comma-separated shard-id list; keys routing outside the allowed
+//     shards are skipped (shard-affinity workloads).
+//   read  <dir> <shards> <prefix> <n>
+
+import { ClusterDb } from '../src/cluster/index.js';
+import { LockError } from '../src/lockfile.js';
+
+const [, , mode, ...rest] = process.argv;
+
+function out(report: Record<string, unknown>): void {
+  process.stdout.write(JSON.stringify(report) + '\n');
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function main(): Promise<void> {
+  const [dir, shards, prefix, n, valueBytes = '64', lockHoldMs = '250', allow = 'all'] = rest;
+  const shardCount = Number(shards);
+  const allowed = allow === 'all' ? null : new Set(allow.split(',').map(Number));
+  const want = Number(n);
+
+  let retries = 0;
+  const withRetry = async <T>(fn: () => Promise<T>): Promise<T> => {
+    for (;;) {
+      try {
+        return await fn();
+      } catch (e) {
+        if ((e as { code?: string }).code !== 'ELOCKED' && !(e instanceof LockError)) throw e;
+        retries++;
+        await sleep(15 + Math.floor(Math.random() * 45));
+      }
+    }
+  };
+
+  if (mode === 'write') {
+    const db = await ClusterDb.open({
+      dir: dir!,
+      shardCount,
+      valueCodec: 'json',
+      fsyncPolicy: 'everysec',
+      lockHoldMs: Number(lockHoldMs),
+    });
+    const pad = 'x'.repeat(Number(valueBytes));
+    let written = 0;
+    const t0 = performance.now();
+    for (let i = 0; written < want; i++) {
+      const key = `${prefix}:${i}`;
+      if (allowed && !allowed.has(db.shardOf(key))) continue;
+      await withRetry(() => db.set(key, { p: prefix, i, pad }));
+      written++;
+    }
+    const ms = performance.now() - t0;
+    out({ ok: 1, mode, n: written, ms, retries, lockWaits: db.stats().lockWaits });
+    await db.close();
+    return;
+  }
+
+  if (mode === 'read') {
+    const db = await ClusterDb.open({ dir: dir!, shardCount, valueCodec: 'json', readOnly: true });
+    let found = 0;
+    const t0 = performance.now();
+    for (let i = 0; found < want; i++) {
+      if (i > want * shardCount * 8 + 1000) throw new Error(`read found only ${found}/${want} keys`);
+      const key = `${prefix}:${i}`;
+      if (allowed && !allowed.has(db.shardOf(key))) continue;
+      const v = (await db.get(key)) as { i?: number } | undefined;
+      if (v !== undefined && v.i === i) found++;
+    }
+    const ms = performance.now() - t0;
+    out({ ok: 1, mode, n: want, found, ms, retries });
+    await db.close();
+    return;
+  }
+
+  out({ ok: 0, error: `unknown mode: ${mode}` });
+  process.exit(1);
+}
+
+main().catch((e) => {
+  out({ ok: 0, mode, error: String(e && (e as Error).stack ? (e as Error).stack : e) });
+  process.exit(1);
+});