From 8bc1c27c400dc11ec211b986f1a833ec1d620fdc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 07:04:19 +0000 Subject: [PATCH] M4: KV indexes with pattern trie, atomic and non-atomic write paths, read-repair, range findBy Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE --- DESIGN.md | 25 ++++ src/types/schema.ts | 2 +- test/kv/indexes.test.ts | 245 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 271 insertions(+), 1 deletion(-) create mode 100644 test/kv/indexes.test.ts diff --git a/DESIGN.md b/DESIGN.md index e363137..a5a62d4 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -434,6 +434,31 @@ Decisions taken where the brief was silent. Each is the simplest option consiste - **Exposed low-level properties** on every node: `path`, `selfKey`, `prefix`, `end` (subtree upper bound), `driver`, `separator`, `isRoot`. The event and document layers build on these. +### M4 — KV indexes + +- **Pattern syntax**: segments separated by the storage separator; `*` matches one segment; `''` + names the root. A literal `*` key can't be targeted by a pattern. +- **Overlapping patterns** that declare the same index name throw at construction. +- **Index trie state is stepped per `child()` call** (O(1) in patterns), so each node resolves its + indexes in O(depth) overall. A node holds two bound index sets: one over its children (for child + writes) and its parent's (for self-form writes such as `users.child('1').set(v)`). +- **Index function results**: `undefined` → not indexed; arrays are multi-entry and de-duplicated + per child; anything that is not `string | number | boolean | null` (or `NaN`) throws a `TypeError` + at write time. Index functions are not called for missing (`undefined`) values. +- **`findBy` results** are `[key, value]` pairs (the value is re-read anyway for verification). + Range queries accept `gt`/`gte`/`lt`/`lte`. Results are in index-value order, then child-key order, + and range queries de-duplicate multi-entry hits. +- **`findBy` batching**: entry keys are read in chunks of 128 (or of the remaining `limit`, if + smaller) and values fetched with one `getMany` per chunk. Stale entries are deleted with one + `deleteMany` per chunk. +- **Read-repair race (non-atomic drivers only)**: a reader can see a concurrent writer's fresh + index add before that writer's value write lands, judge it stale, and delete it. With multiple + writers on a driver without `batch`, this can lose an index entry. Mitigation: `readRepair: false` + on `KVStorage` (stale entries are then only skipped, never deleted). With `batch`, entries never go + stale, so the race doesn't arise. +- **`set` of an indexed child with duplicate keys in one `setMany`**: last write wins, and index + diffs are computed once per key. + ## 13. Benchmark baselines diff --git a/src/types/schema.ts b/src/types/schema.ts index 1c9ea94..6089198 100644 --- a/src/types/schema.ts +++ b/src/types/schema.ts @@ -7,7 +7,7 @@ export type Spec = { value?: unknown; children?: Record } /** Untyped default: any value, any children. */ export interface AnySpec { value: unknown - children: Record + children: { [key: string]: AnySpec } } export type ValueOf = S extends { value: infer V } ? V : never diff --git a/test/kv/indexes.test.ts b/test/kv/indexes.test.ts new file mode 100644 index 0000000..acbc8dd --- /dev/null +++ b/test/kv/indexes.test.ts @@ -0,0 +1,245 @@ +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { KVStorage, type KVStorageOptions } from '../../src/kv/node.js' +import { InMemoryKVDriver } from '../../src/kv/memory.js' +import type { KVDriver } from '../../src/kv/driver.js' +import { encodeIndexValue } from '../../src/core/encoding.js' +import { collect, CountingDriver, type Capability } from '../helpers.js' + +type User = { email?: string; age?: number; tags?: string[] } +type Post = { createdAt: string; tags: string[] } + +const indexes: KVStorageOptions['indexes'] = { + users: { + byEmail: (u: User) => u.email, + byAge: (u: User) => u.age, + byTag: (u: User) => u.tags, + }, + 'users/*/posts': { byDate: (p: Post) => p.createdAt, byTag: (p: Post) => p.tags }, +} + +function setup(hide: Capability[] = []) { + const driver = new CountingDriver(new InMemoryKVDriver(), hide) + return { driver, db: new KVStorage(driver, { indexes }) } +} + +const modes: [string, Capability[]][] = [ + ['atomic batch', []], + ['non-atomic', ['batch']], +] + +describe.each(modes)('indexes (%s)', (_, hide) => { + it('writes entries under /~i///', async () => { + const { db, driver } = setup(hide) + await db.child('users').set('123', { email: 'sada@x.io', age: 30 }) + const keys = (await collect(driver.inner.scan({}))).map(([k]) => k) + expect(keys).toEqual([`users/123`, `users/~i/byAge/${encodeIndexValue(30)}/123`, `users/~i/byEmail/${encodeIndexValue('sada@x.io')}/123`]) + }) + + it('finds by exact value and keeps entries in sync on update and delete', async () => { + const { db } = setup(hide) + const users = db.child('users') + await users.set('1', { email: 'a@x', age: 30 }) + await users.set('2', { email: 'b@x', age: 30 }) + expect(await collect(users.findBy('byEmail', 'a@x'))).toEqual([['1', { email: 'a@x', age: 30 }]]) + expect((await collect(users.findBy('byAge', 30))).map(([k]) => k)).toEqual(['1', '2']) + await users.set('1', { email: 'c@x', age: 31 }) + expect(await collect(users.findBy('byEmail', 'a@x'))).toEqual([]) + expect((await collect(users.findBy('byEmail', 'c@x'))).map(([k]) => k)).toEqual(['1']) + await users.delete('2') + expect(await collect(users.findBy('byAge', 30))).toEqual([]) + }) + + it('self-form writes on a child update the parent index', async () => { + const { db } = setup(hide) + const u = db.child('users').child('7') + await u.set({ email: 'self@x' }) + expect((await collect(db.child('users').findBy('byEmail', 'self@x'))).map(([k]) => k)).toEqual(['7']) + await u.delete() + expect(await collect(db.child('users').findBy('byEmail', 'self@x'))).toEqual([]) + }) + + it('deleteTree removes the child entry', async () => { + const { db, driver } = setup(hide) + await db.child('users').set('7', { email: 'e' }) + await db.child('users').child('7').child('posts').set('p', { createdAt: '2024', tags: [] }) + await db.child('users').deleteTree('7') + expect(await collect(driver.inner.scan({}))).toEqual([]) + }) + + it('supports multi-entry values, de-duplicated in range queries', async () => { + const { db } = setup(hide) + const users = db.child('users') + await users.set('1', { tags: ['a', 'b', 'a'] }) + await users.set('2', { tags: ['b'] }) + expect((await collect(users.findBy('byTag', 'b'))).map(([k]) => k)).toEqual(['1', '2']) + expect((await collect(users.findBy('byTag', { gte: 'a', lt: 'c' }))).map(([k]) => k)).toEqual(['1', '2']) + await users.set('1', { tags: ['c'] }) + expect((await collect(users.findBy('byTag', 'b'))).map(([k]) => k)).toEqual(['2']) + }) + + it('range queries are ordered by value and honour bounds, limit and reverse', async () => { + const { db } = setup(hide) + const users = db.child('users') + await users.setMany([ + ['a', { age: 10 }], + ['b', { age: 9 }], + ['c', { age: -1.5 }], + ['d', { age: 100 }], + ['e', { age: 20 }], + ]) + const ages = async (q: object, o = {}) => (await collect(users.findBy('byAge', q as never, o))).map(([, v]) => (v as User).age) + expect(await ages({})).toEqual([-1.5, 9, 10, 20, 100]) + expect(await ages({ gte: 9, lt: 20 })).toEqual([9, 10]) + expect(await ages({ gt: 9, lte: 20 })).toEqual([10, 20]) + expect(await ages({ gte: 0 }, { limit: 2 })).toEqual([9, 10]) + expect(await ages({ lt: 100 }, { reverse: true })).toEqual([20, 10, 9, -1.5]) + }) + + it('applies pattern indexes with * to nested nodes only', async () => { + const { db } = setup(hide) + const posts = db.child('users').child('u1').child('posts') + await posts.set('p1', { createdAt: '2024-02', tags: ['x'] }) + await posts.set('p2', { createdAt: '2024-01', tags: ['y'] }) + expect((await collect(posts.findBy('byDate', { gte: '2024' }))).map(([k]) => k)).toEqual(['p2', 'p1']) + await expect(db.child('other').findBy('byDate', 'x').next()).rejects.toThrow(/No index/) + }) + + it('handles child keys and values needing escaping', async () => { + const { db } = setup(hide) + const users = db.child('users') + await users.set('a/b%~', { email: 'we/ird%~@x' }) + expect(await collect(users.findBy('byEmail', 'we/ird%~@x'))).toEqual([['a/b%~', { email: 'we/ird%~@x' }]]) + }) +}) + +describe('index costs', () => { + it('non-indexed nodes do not read before writing', async () => { + const { db, driver } = setup() + driver.reset() + await db.child('other').set('1', { email: 'x' }) + await db.child('other').setMany([['2', {}]]) + expect(driver.calls).toEqual({ setMany: 2 }) + }) + + it('indexed writes: one getMany then one atomic batch', async () => { + const { db, driver } = setup() + driver.reset() + await db.child('users').setMany([ + ['1', { email: 'a' }], + ['2', { email: 'b' }], + ]) + expect(driver.calls).toEqual({ getMany: 1, batch: 1 }) + }) + + it('findBy with a small limit makes one value read', async () => { + const { db, driver } = setup() + await db.child('users').setMany(Array.from({ length: 300 }, (_, i) => [String(i), { age: 1 }] as [string, User])) + driver.reset() + expect(await collect(db.child('users').findBy('byAge', 1, { limit: 1 }))).toHaveLength(1) + expect(driver.calls).toEqual({ scan: 1, getMany: 1 }) + driver.reset() + expect(await collect(db.child('users').findBy('byAge', 1))).toHaveLength(300) + expect(driver.calls).toEqual({ scan: 1, getMany: 3 }) + }) +}) + +describe('non-atomic crash recovery (read-repair)', () => { + /** Fails the n-th mutating call. */ + class CrashingDriver extends CountingDriver { + failAt = Infinity + #n = 0 + override setMany(entries: [string, unknown][]) { + if (++this.#n === this.failAt) return Promise.reject(new Error('crash')) + return super.setMany(entries) + } + override deleteMany(keys: string[]) { + if (++this.#n === this.failAt) return Promise.reject(new Error('crash')) + return super.deleteMany(keys) + } + } + + it('skips and deletes stale entries left by a crash after the index add', async () => { + const driver = new CrashingDriver(new InMemoryKVDriver(), ['batch']) + const users = new KVStorage(driver, { indexes }).child('users') + await users.set('1', { email: 'old' }) + driver.failAt = 2 + 2 // adds succeed, value write crashes + await expect(users.set('1', { email: 'new' })).rejects.toThrow('crash') + // The 'new' entry is a dangling hint; the value still says 'old'. + expect(await collect(users.findBy('byEmail', 'new'))).toEqual([]) + expect(await collect(users.findBy('byEmail', 'old'))).toEqual([['1', { email: 'old' }]]) + const entries = (await collect(driver.inner.scan({ gte: 'users/~i/' }))).map(([k]) => k) + expect(entries).toEqual([`users/~i/byEmail/${encodeIndexValue('old')}/1`]) + }) + + it('skips stale entries left by a crash before the index remove', async () => { + const driver = new CrashingDriver(new InMemoryKVDriver(), ['batch']) + const users = new KVStorage(driver, { indexes }).child('users') + await users.set('1', { email: 'old' }) + driver.failAt = 2 + 3 // adds + value succeed, remove crashes + await expect(users.set('1', { email: 'new' })).rejects.toThrow('crash') + expect(await collect(users.findBy('byEmail', 'old'))).toEqual([]) + expect(await collect(users.findBy('byEmail', 'new'))).toEqual([['1', { email: 'new' }]]) + }) + + it('can leave stale entries in place with readRepair: false', async () => { + const driver = new InMemoryKVDriver() + await driver.setMany([[`users/~i/byEmail/${encodeIndexValue('x')}/1`, true]]) + const users = new KVStorage(driver, { indexes, readRepair: false }).child('users') + expect(await collect(users.findBy('byEmail', 'x'))).toEqual([]) + expect(await collect(driver.scan({}))).toHaveLength(1) + }) +}) + +describe('index declarations', () => { + it('rejects the same index name on overlapping patterns', () => { + expect(() => new KVStorage(new InMemoryKVDriver(), { indexes: { 'a/*': { i: () => 1 }, '*/b': { i: () => 2 } } })).toThrow(/overlapping/) + expect(() => new KVStorage(new InMemoryKVDriver(), { indexes: { 'a/*': { i: () => 1 }, '*/b': { j: () => 2 } } })).not.toThrow() + }) + + it('rejects non-indexable index function results', async () => { + const db = new KVStorage(new InMemoryKVDriver(), { indexes: { '': { bad: () => ({}) as never } } }) + await expect(db.set('x', 1)).rejects.toThrow(/non-indexable/) + }) +}) + +describe('property: findBy equals brute-force filter', () => { + const val = fc.record({ age: fc.option(fc.integer({ min: -3, max: 3 }), { nil: undefined }), tags: fc.array(fc.constantFrom('a', 'b', 'c/d'), { maxLength: 3 }) }) + const op = fc.oneof( + fc.record({ t: fc.constant('set' as const), k: fc.constantFrom('1', '2', '3', 'x/y'), v: val }), + fc.record({ t: fc.constant('del' as const), k: fc.constantFrom('1', '2', '3', 'x/y') }), + ) + for (const [mode, hide] of modes) { + it(mode, async () => { + await fc.assert( + fc.asyncProperty(fc.array(op, { maxLength: 20 }), fc.integer({ min: -4, max: 4 }), fc.integer({ min: -4, max: 4 }), async (ops, lo, hi) => { + const driver: KVDriver = new CountingDriver(new InMemoryKVDriver(), hide) + const users = new KVStorage(driver, { indexes }).child('users') + const model = new Map() + for (const o of ops) { + if (o.t === 'set') { + await users.set(o.k, o.v) + model.set(o.k, o.v) + } else { + await users.delete(o.k) + model.delete(o.k) + } + } + const got = (await collect(users.findBy('byAge', { gte: lo, lt: hi }))).map(([k]) => k).sort() + const exp = [...model].filter(([, v]) => v.age !== undefined && v.age >= lo && v.age < hi).map(([k]) => k).sort() + expect(got).toEqual(exp) + for (const tag of ['a', 'b', 'c/d']) { + const g = (await collect(users.findBy('byTag', tag))).map(([k]) => k).sort() + expect(g).toEqual([...model].filter(([, v]) => v.tags?.includes(tag)).map(([k]) => k).sort()) + } + // No stale entries are ever left behind by clean writes. + let entryCount = 0 + for await (const _ of driver.scan({ gte: 'users/~i/', lt: 'users/~i0' })) entryCount++ + const expectedEntries = [...model.values()].reduce((n, v) => n + (v.age !== undefined ? 1 : 0) + new Set(v.tags).size, 0) + expect(entryCount).toBe(expectedEntries) + }), + { numRuns: 200 }, + ) + }) + } +})