M5: KVEventStorage with dense commit keys, setIfAbsent and mutex modes, head cache, probe-based read

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE

CLAUDE.md

@@ -6,7 +6,8 @@ Record new decisions in DESIGN.md section 12.
 
 ## Commands
 
-- `npm test` — run all tests (Vitest; includes `*.test-d.ts` type tests)
+- `npm test` — `tsc --noEmit` then all Vitest tests (including `*.test-d.ts` type tests)
+- `npm run test:only -- <file>` — Vitest only, e.g. a single file
 - `npm run typecheck` — `tsc --noEmit` over src, test and bench
 - `npm run bench` — `vitest bench`
 - `npm run build` — emit `dist/`

DESIGN.md

@@ -459,6 +459,31 @@ Decisions taken where the brief was silent. Each is the simplest option consiste
 - **`set` of an indexed child with duplicate keys in one `setMany`**: last write wins, and index
   diffs are computed once per key.
 
+### M5 — Events
+
+- **`EventStorage` is an interface (type), not a class.** The brief's sample says `new EventStorage(kv)`
+  but also says it is the abstract interface; the concrete class is `KVEventStorage(kv, opts?)`, which
+  accepts a `KVStorage` or any `KVNode` as root. Options: `headCacheSize` (default 10 000),
+  `maxRetries` (default 5) and `now` (clock, default `Date.now`).
+- **Commit value**: `{ firstVersion, at, events }`, with `at` in ms since epoch. Read results are
+  `{ version, commit, at, data }`.
+- **The root node is not a stream** (consistent with the root having no own KV value). Non-root
+  nodes are both a stream and a container of child streams; `keys()` lists child streams, and a
+  node's `~e` subtree is reserved, so streams and children never mix.
+- **`version()` always reads storage** (one reverse scan with `limit: 1`) and refreshes the head cache.
+  Only `append` trusts the cache. When a cached head disagrees with a numeric `expectedVersion`,
+  the head is re-read before throwing, so a stale cache never produces a false `ConcurrencyError`.
+- **`append([])`** writes nothing; it still checks `expectedVersion` against a fresh head and returns it.
+- **Read arity**: `read()` and `read(opts)` are the self form (the first argument is a non-null
+  object). Any other first argument is a key, so `read(undefined)` throws like every other
+  passed-but-undefined key.
+- **Locating `from`/`to`**: commit `n` has `firstVersion >= n`, so the commit containing version `v` has
+  `seq <= v`. `read` probes `seq = v` first; with single-event commits that is the answer, in one
+  `getMany`. Otherwise it runs an 8-ary search, one `getMany` of up to 8 probes per round, which is
+  O(log₈ commits) round trips. The head is only read when the guess is past the end.
+- **Mutex mode** (driver without `setIfAbsent`): a per-stream in-process promise chain serialises
+  appends. It is safe only with a single writer process.
+
 <!-- decisions appended per milestone -->
 
 ## 13. Benchmark baselines

package.json

@@ -40,9 +40,10 @@
   "scripts": {
     "build": "tsc -p tsconfig.build.json",
     "typecheck": "tsc --noEmit",
-    "test": "vitest run",
+    "test": "tsc --noEmit && vitest run",
     "test:watch": "vitest",
-    "bench": "vitest bench --run"
+    "bench": "vitest bench --run",
+    "test:only": "vitest run"
   },
   "devDependencies": {
     "@types/node": "^22.20.5",

src/core/lru.ts

@@ -0,0 +1,30 @@
+/** Minimal bounded LRU over `Map` insertion order. */
+export class LRU<K, V> {
+  readonly #map = new Map<K, V>()
+
+  constructor(readonly capacity: number) {}
+
+  get(key: K): V | undefined {
+    const v = this.#map.get(key)
+    if (v !== undefined) {
+      this.#map.delete(key)
+      this.#map.set(key, v)
+    }
+    return v
+  }
+
+  set(key: K, value: V): void {
+    if (this.capacity <= 0) return
+    this.#map.delete(key)
+    this.#map.set(key, value)
+    if (this.#map.size > this.capacity) this.#map.delete(this.#map.keys().next().value as K)
+  }
+
+  delete(key: K): void {
+    this.#map.delete(key)
+  }
+
+  get size(): number {
+    return this.#map.size
+  }
+}

src/core/mutex.ts

@@ -0,0 +1,15 @@
+/** Per-key in-process mutex: calls with the same key run one at a time, in order. */
+export class KeyedMutex {
+  readonly #tails = new Map<string, Promise<unknown>>()
+
+  run<T>(key: string, fn: () => Promise<T>): Promise<T> {
+    const prev = this.#tails.get(key) ?? Promise.resolve()
+    const result = prev.then(fn, fn)
+    const tail = result.catch(() => {})
+    this.#tails.set(key, tail)
+    void tail.then(() => {
+      if (this.#tails.get(key) === tail) this.#tails.delete(key)
+    })
+    return result
+  }
+}

src/events/index.ts

@@ -0,0 +1,2 @@
+export * from './types.js'
+export * from './kv.js'

src/events/kv.ts

@@ -0,0 +1,288 @@
+import { LRU } from '../core/lru.js'
+import { KeyedMutex } from '../core/mutex.js'
+import { escapeSegment, RESERVED, succ, toSegment } from '../core/path.js'
+import { ConcurrencyError } from '../errors.js'
+import type { KVDriver } from '../kv/driver.js'
+import type { KVNode } from '../kv/node.js'
+import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueOf } from '../types/schema.js'
+import type { AppendOptions, AppendResult, EventNode, ReadOptions, StoredEvent, StreamKeysOptions } from './types.js'
+
+export interface KVEventStorageOptions {
+  /** Max streams whose head is cached. Default 10 000. */
+  headCacheSize?: number
+  /** Retries after a commit collision when `expectedVersion` is `'any'`. Default 5. */
+  maxRetries?: number
+  /** Clock for commit timestamps. Default `Date.now`. */
+  now?: () => number
+}
+
+/** Stored value of one commit key. */
+export interface CommitRecord<E = unknown> {
+  firstVersion: number
+  at: number
+  events: E[]
+}
+
+interface Head {
+  commit: number
+  version: number
+}
+
+interface Stream {
+  /** Stream's own key (escaped); identifies it in the cache and in errors. */
+  key: string
+  /** `<key>/~e/` — commit keys are `base + pad12(seq)`. */
+  base: string
+  /** Exclusive upper bound of commit keys. */
+  end: string
+}
+
+/** @internal */
+interface EventContext {
+  readonly driver: KVDriver
+  readonly sep: string
+  readonly heads: LRU<string, Head>
+  readonly mutex: KeyedMutex
+  readonly maxRetries: number
+  readonly now: () => number
+}
+
+const SEQ_WIDTH = 12
+const SEARCH_FANOUT = 8
+const EMPTY_HEAD: Head = { commit: 0, version: 0 }
+
+const pad = (seq: number) => String(seq).padStart(SEQ_WIDTH, '0')
+
+/**
+ * Event node over a `KVNode`. Each append is one commit key, densely numbered:
+ * `<stream>/~e/<commitSeq:12>` → `{ firstVersion, at, events }`. Commits are claimed with
+ * `setIfAbsent`, so any stale writer collides with the existing key. Without `setIfAbsent`,
+ * appends are serialised by an in-process per-stream mutex (single-writer only).
+ */
+export class KVEventNode<S extends Spec = AnySpec> implements EventNode<S> {
+  readonly #ctx: EventContext
+  readonly #kv: KVNode<AnySpec>
+
+  /** @internal Use `KVEventStorage` and `child()`. */
+  constructor(ctx: EventContext, kv: KVNode<AnySpec>) {
+    this.#ctx = ctx
+    this.#kv = kv
+  }
+
+  get path(): readonly string[] {
+    return this.#kv.path
+  }
+
+  /** The underlying KV node. */
+  get kv(): KVNode<AnySpec> {
+    return this.#kv
+  }
+
+  child<K extends KeyOf<S>>(key: K): KVEventNode<ChildOf<S, K>> {
+    return new KVEventNode(this.#ctx, this.#kv.child(key as string).as<AnySpec>())
+  }
+
+  as<S2 extends Spec>(): KVEventNode<S2> {
+    return this as unknown as KVEventNode<S2>
+  }
+
+  append(events: readonly ValueOf<S>[], opts?: AppendOptions): Promise<AppendResult>
+  append<K extends KeyOf<S>>(key: K, events: readonly ValueOf<ChildOf<S, K>>[], opts?: AppendOptions): Promise<AppendResult>
+  async append(...args: unknown[]): Promise<AppendResult> {
+    const self = Array.isArray(args[0])
+    const stream = self ? this.#self() : this.#child(args[0])
+    const events = args[self ? 0 : 1]
+    const opts = (args[self ? 1 : 2] ?? {}) as AppendOptions
+    if (!Array.isArray(events)) throw new TypeError('append() expects an array of events')
+    const expected = opts.expectedVersion ?? 'any'
+    if (expected !== 'any' && !(Number.isInteger(expected) && expected >= 0)) throw new TypeError(`Invalid expectedVersion ${String(expected)}`)
+    return this.#append(stream, events, expected)
+  }
+
+  read(opts?: ReadOptions): AsyncIterable<StoredEvent<ValueOf<S>>>
+  read<K extends KeyOf<S>>(key: K, opts?: ReadOptions): AsyncIterable<StoredEvent<ValueOf<ChildOf<S, K>>>>
+  async *read(...args: unknown[]): AsyncGenerator<StoredEvent<any>> {
+    const self = args.length === 0 || (typeof args[0] === 'object' && args[0] !== null)
+    const stream = self ? this.#self() : this.#child(args[0])
+    yield* this.#read(stream, ((self ? args[0] : args[1]) ?? {}) as ReadOptions)
+  }
+
+  version(): Promise<number>
+  version<K extends KeyOf<S>>(key: K): Promise<number>
+  async version(...args: unknown[]): Promise<number> {
+    const stream = args.length === 0 ? this.#self() : this.#child(args[0])
+    return (await this.#loadHead(stream)).version
+  }
+
+  keys(opts: StreamKeysOptions = {}): AsyncIterable<KeyString<S>> {
+    return this.#kv.keys(opts) as AsyncIterable<KeyString<S>>
+  }
+
+  // ---------------------------------------------------------------- internals
+
+  #self(): Stream {
+    if (this.#kv.isRoot) throw new TypeError('The root node is not a stream')
+    return this.#streamAt(this.#kv.selfKey)
+  }
+
+  #child(key: unknown): Stream {
+    return this.#streamAt(this.#kv.prefix + escapeSegment(toSegment(key), this.#ctx.sep))
+  }
+
+  #streamAt(key: string): Stream {
+    const sep = this.#ctx.sep
+    const head = key + sep + RESERVED.events
+    return { key, base: head + sep, end: head + succ(sep) }
+  }
+
+  async #loadHead(s: Stream): Promise<Head> {
+    for await (const [key, value] of this.#ctx.driver.scan({ gte: s.base, lt: s.end, reverse: true, limit: 1, values: true })) {
+      const c = value as CommitRecord
+      const head = { commit: Number(key.slice(s.base.length)), version: c.firstVersion + c.events.length - 1 }
+      this.#ctx.heads.set(s.key, head)
+      return head
+    }
+    return EMPTY_HEAD
+  }
+
+  async #append(s: Stream, events: readonly unknown[], expected: number | 'any'): Promise<AppendResult> {
+    const { driver, heads, maxRetries } = this.#ctx
+    if (events.length === 0) {
+      const head = await this.#loadHead(s)
+      if (expected !== 'any' && head.version !== expected) throw new ConcurrencyError(s.key, expected, head.version)
+      return { version: head.version, commit: head.commit }
+    }
+    if (!driver.setIfAbsent) return this.#ctx.mutex.run(s.key, () => this.#appendSingleWriter(s, events, expected))
+
+    for (let attempt = 0; ; attempt++) {
+      let head = heads.get(s.key)
+      if (!head) head = await this.#loadHead(s)
+      else if (expected !== 'any' && head.version !== expected) head = await this.#loadHead(s) // cache may be stale
+      if (expected !== 'any' && head.version !== expected) throw new ConcurrencyError(s.key, expected, head.version)
+
+      const seq = head.commit + 1
+      const record: CommitRecord = { firstVersion: head.version + 1, at: this.#ctx.now(), events: [...events] }
+      const [written] = await driver.setIfAbsent([[s.base + pad(seq), record]])
+      if (written) {
+        const next = { commit: seq, version: head.version + events.length }
+        heads.set(s.key, next)
+        return next
+      }
+      // Lost the race for this commit number.
+      heads.delete(s.key)
+      if (expected !== 'any' || attempt >= maxRetries) {
+        throw new ConcurrencyError(s.key, expected, (await this.#loadHead(s)).version)
+      }
+    }
+  }
+
+  async #appendSingleWriter(s: Stream, events: readonly unknown[], expected: number | 'any'): Promise<AppendResult> {
+    const head = this.#ctx.heads.get(s.key) ?? (await this.#loadHead(s))
+    if (expected !== 'any' && head.version !== expected) throw new ConcurrencyError(s.key, expected, head.version)
+    const seq = head.commit + 1
+    const record: CommitRecord = { firstVersion: head.version + 1, at: this.#ctx.now(), events: [...events] }
+    await this.#ctx.driver.setMany([[s.base + pad(seq), record]])
+    const next = { commit: seq, version: head.version + events.length }
+    this.#ctx.heads.set(s.key, next)
+    return next
+  }
+
+  /**
+   * Sequence number of the commit containing `version`, or 0 if the stream hasn't reached it.
+   * Commit `n` has `firstVersion >= n`, so the answer is at most `version`: probe that first
+   * (one round trip when commits hold single events), then k-ary search with `getMany` probes.
+   */
+  async #locate(s: Stream, version: number): Promise<number> {
+    const { driver } = this.#ctx
+    const [guess] = (await driver.getMany([s.base + pad(version)])) as (CommitRecord | undefined)[]
+    if (guess && guess.firstVersion <= version) return version
+    let hi: number
+    if (guess) hi = version - 1
+    else {
+      const head = await this.#loadHead(s)
+      if (head.version < version) return 0
+      hi = head.commit
+    }
+    let lo = 1 // commit 1 starts at version 1
+    while (lo < hi) {
+      const probes: number[] = []
+      for (let i = 1; i <= SEARCH_FANOUT; i++) {
+        const p = lo + Math.ceil(((hi - lo) * i) / SEARCH_FANOUT)
+        if (p !== probes.at(-1)) probes.push(p)
+      }
+      const found = (await driver.getMany(probes.map((p) => s.base + pad(p)))) as (CommitRecord | undefined)[]
+      for (let i = 0; i < probes.length; i++) {
+        const c = found[i]
+        if (c && c.firstVersion <= version) lo = probes[i]!
+        else {
+          hi = probes[i]! - 1
+          break
+        }
+      }
+    }
+    return lo
+  }
+
+  async *#read(s: Stream, opts: ReadOptions): AsyncGenerator<StoredEvent<unknown>> {
+    const from = Math.max(1, opts.from ?? 1)
+    const to = opts.to ?? Infinity
+    let remaining = opts.limit ?? Infinity
+    if (to < from || remaining <= 0) return
+    const { driver } = this.#ctx
+
+    if (!opts.reverse) {
+      let gte = s.base
+      if (from > 1) {
+        const seq = await this.#locate(s, from)
+        if (seq === 0) return
+        gte = s.base + pad(seq)
+      }
+      for await (const [key, value] of driver.scan({ gte, lt: s.end, values: true })) {
+        const c = value as CommitRecord
+        const commit = Number(key.slice(s.base.length))
+        for (let i = 0; i < c.events.length; i++) {
+          const version = c.firstVersion + i
+          if (version < from) continue
+          if (version > to) return
+          yield { version, commit, at: c.at, data: c.events[i] }
+          if (--remaining <= 0) return
+        }
+      }
+      return
+    }
+
+    let lt = s.end
+    if (to !== Infinity) {
+      const seq = await this.#locate(s, to)
+      if (seq !== 0) lt = s.base + pad(seq + 1)
+    }
+    for await (const [key, value] of driver.scan({ gte: s.base, lt, reverse: true, values: true })) {
+      const c = value as CommitRecord
+      const commit = Number(key.slice(s.base.length))
+      for (let i = c.events.length - 1; i >= 0; i--) {
+        const version = c.firstVersion + i
+        if (version > to) continue
+        if (version < from) return
+        yield { version, commit, at: c.at, data: c.events[i] }
+        if (--remaining <= 0) return
+      }
+    }
+  }
+}
+
+/** Event storage over a `KVStorage` (or any `KVNode` as root). */
+export class KVEventStorage<S extends Spec = AnySpec> extends KVEventNode<S> {
+  constructor(kv: KVNode<AnySpec>, opts: KVEventStorageOptions = {}) {
+    super(
+      {
+        driver: kv.driver,
+        sep: kv.separator,
+        heads: new LRU(opts.headCacheSize ?? 10_000),
+        mutex: new KeyedMutex(),
+        maxRetries: opts.maxRetries ?? 5,
+        now: opts.now ?? Date.now,
+      },
+      kv,
+    )
+  }
+}

src/events/types.ts

@@ -0,0 +1,62 @@
+import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueOf } from '../types/schema.js'
+
+export interface AppendOptions {
+  /** Version the stream must be at (0 = empty). Default `'any'`. */
+  expectedVersion?: number | 'any'
+}
+
+export interface AppendResult {
+  /** Stream version after the append (the last event's version). */
+  version: number
+  /** Commit sequence number of this append (the current head commit when no events were given). */
+  commit: number
+}
+
+export interface ReadOptions {
+  /** First version to read (inclusive). Default 1. */
+  from?: number
+  /** Last version to read (inclusive). Default: the head. */
+  to?: number
+  reverse?: boolean
+  limit?: number
+}
+
+export interface StoredEvent<E> {
+  version: number
+  commit: number
+  /** Commit timestamp (ms since epoch). */
+  at: number
+  data: E
+}
+
+export interface StreamKeysOptions {
+  limit?: number
+  after?: string | number
+  reverse?: boolean
+}
+
+/**
+ * One node of an event store: a stream (its own events) and a container of child streams.
+ * Same navigation and self/child arity rules as `KVNode`.
+ */
+export interface EventNode<S extends Spec = AnySpec> {
+  readonly path: readonly string[]
+  child<K extends KeyOf<S>>(key: K): EventNode<ChildOf<S, K>>
+  as<S2 extends Spec>(): EventNode<S2>
+
+  append(events: readonly ValueOf<S>[], opts?: AppendOptions): Promise<AppendResult>
+  append<K extends KeyOf<S>>(key: K, events: readonly ValueOf<ChildOf<S, K>>[], opts?: AppendOptions): Promise<AppendResult>
+
+  read(opts?: ReadOptions): AsyncIterable<StoredEvent<ValueOf<S>>>
+  read<K extends KeyOf<S>>(key: K, opts?: ReadOptions): AsyncIterable<StoredEvent<ValueOf<ChildOf<S, K>>>>
+
+  /** Current version (0 = empty). Always reads storage. */
+  version(): Promise<number>
+  version<K extends KeyOf<S>>(key: K): Promise<number>
+
+  /** Child stream keys. */
+  keys(opts?: StreamKeysOptions): AsyncIterable<KeyString<S>>
+}
+
+/** The root of an event store. Native backends (EventStoreDB, Kafka, …) implement this too. */
+export type EventStorage<S extends Spec = AnySpec> = EventNode<S>

src/index.ts

@@ -3,3 +3,4 @@ export * from './core/path.js'
 export * from './core/encoding.js'
 export * from './kv/index.js'
 export * from './types/schema.js'
+export * from './events/index.js'

test/events/kv.test.ts

@@ -0,0 +1,200 @@
+import { describe, expect, it } from 'vitest'
+import fc from 'fast-check'
+import { KVStorage } from '../../src/kv/node.js'
+import { InMemoryKVDriver } from '../../src/kv/memory.js'
+import { KVEventStorage } from '../../src/events/kv.js'
+import { ConcurrencyError } from '../../src/errors.js'
+import { collect, CountingDriver, type Capability } from '../helpers.js'
+
+function setup(hide: Capability[] = [], inner = new InMemoryKVDriver()) {
+  const driver = new CountingDriver(inner, hide)
+  let t = 1000
+  const events = new KVEventStorage(new KVStorage(driver), { now: () => t++ })
+  return { driver, events }
+}
+
+describe('KVEventStorage', () => {
+  it('appends and reads with versions and metadata', async () => {
+    const { events } = setup()
+    const orders = events.child('orders')
+    expect(await orders.append('o1', [{ type: 'created', total: 40 }], { expectedVersion: 0 })).toEqual({ version: 1, commit: 1 })
+    expect(await orders.child('o1').append([{ type: 'paid' }, { type: 'shipped' }])).toEqual({ version: 3, commit: 2 })
+    expect(await collect(orders.read('o1'))).toEqual([
+      { version: 1, commit: 1, at: 1000, data: { type: 'created', total: 40 } },
+      { version: 2, commit: 2, at: 1001, data: { type: 'paid' } },
+      { version: 3, commit: 2, at: 1001, data: { type: 'shipped' } },
+    ])
+    expect(await orders.version('o1')).toBe(3)
+    expect(await orders.child('o1').version()).toBe(3)
+    expect(await orders.version('nope')).toBe(0)
+  })
+
+  it('stores one key per commit under <stream>/~e/<seq:12>', async () => {
+    const { events, driver } = setup()
+    await events.child('orders').append('o1', ['a'])
+    await events.child('orders').append('o1', ['b', 'c'])
+    expect(await collect(driver.inner.scan({ values: true }))).toEqual([
+      ['orders/o1/~e/000000000001', { firstVersion: 1, at: 1000, events: ['a'] }],
+      ['orders/o1/~e/000000000002', { firstVersion: 2, at: 1001, events: ['b', 'c'] }],
+    ])
+  })
+
+  it('enforces expectedVersion', async () => {
+    const { events } = setup()
+    const s = events.child('s')
+    await s.append(['a'], { expectedVersion: 0 })
+    await expect(s.append(['b'], { expectedVersion: 0 })).rejects.toBeInstanceOf(ConcurrencyError)
+    await expect(s.append(['b'], { expectedVersion: 5 })).rejects.toMatchObject({ expectedVersion: 5, actualVersion: 1 })
+    expect(await s.append(['b'], { expectedVersion: 1 })).toEqual({ version: 2, commit: 2 })
+    await expect(s.append([], { expectedVersion: 1 })).rejects.toBeInstanceOf(ConcurrencyError)
+    expect(await s.append([])).toEqual({ version: 2, commit: 2 })
+  })
+
+  it('applies the arity rules', async () => {
+    const { events } = setup()
+    const orders = events.child('orders')
+    const id: string | undefined = undefined
+    await expect(orders.append(id as never, ['x'] as never)).rejects.toThrow(TypeError)
+    await expect(orders.version(id as never)).rejects.toThrow(TypeError)
+    await expect(collect(orders.read(id as never))).rejects.toThrow(TypeError)
+    await expect(events.append(['x'])).rejects.toThrow(/root/)
+    await expect(orders.append('o1', 'x' as never)).rejects.toThrow(/array/)
+  })
+
+  it('lists child streams and keeps nested streams separate', async () => {
+    const { events } = setup()
+    const orders = events.child('orders')
+    await orders.append('o1', ['a'])
+    await orders.append('o2', ['b'])
+    await orders.child('o1').child('shipments').append('s1', ['ship'])
+    await orders.child('o1').append('notes', ['n'])
+    expect(await collect(orders.keys())).toEqual(['o1', 'o2'])
+    expect(await collect(orders.child('o1').keys())).toEqual(['notes', 'shipments'])
+    expect((await collect(orders.read('o1'))).map((e) => e.data)).toEqual(['a'])
+  })
+})
+
+describe('read windows (property)', () => {
+  it('matches brute force for any commit layout', async () => {
+    await fc.assert(
+      fc.asyncProperty(
+        fc.array(fc.integer({ min: 1, max: 4 }), { maxLength: 25 }),
+        fc.record({
+          from: fc.option(fc.integer({ min: -1, max: 70 }), { nil: undefined }),
+          to: fc.option(fc.integer({ min: 0, max: 70 }), { nil: undefined }),
+          reverse: fc.boolean(),
+          limit: fc.option(fc.integer({ min: 0, max: 10 }), { nil: undefined }),
+        }),
+        async (sizes, opts) => {
+          const { events } = setup()
+          const s = events.child('s')
+          const all: number[] = []
+          for (const n of sizes) {
+            const batch = Array.from({ length: n }, (_, i) => all.length + i + 1)
+            await s.append(batch)
+            all.push(...batch)
+          }
+          let exp = all.filter((v) => v >= Math.max(1, opts.from ?? 1) && v <= (opts.to ?? Infinity))
+          if (opts.reverse) exp.reverse()
+          if (opts.limit !== undefined) exp = exp.slice(0, opts.limit)
+          const got = await collect(s.read(opts))
+          expect(got.map((e) => e.data)).toEqual(exp)
+          expect(got.map((e) => e.version)).toEqual(exp)
+        },
+      ),
+      { numRuns: 300 },
+    )
+  })
+})
+
+describe('costs', () => {
+  it('cached head: a warm append is a single setIfAbsent', async () => {
+    const { events, driver } = setup()
+    const s = events.child('s')
+    await s.append(['a'])
+    driver.reset()
+    await s.append(['b'])
+    await s.append(['c'], { expectedVersion: 2 })
+    expect(driver.calls).toEqual({ setIfAbsent: 2 })
+  })
+
+  it('cold head lookup is one reverse scan with limit 1', async () => {
+    const inner = new InMemoryKVDriver()
+    await setup([], inner).events.child('s').append(['a', 'b'])
+    const { events, driver } = setup([], inner)
+    expect(await events.child('s').append(['c'])).toEqual({ version: 3, commit: 2 })
+    expect(driver.calls).toEqual({ scan: 1, setIfAbsent: 1 })
+  })
+
+  it('read({ from }) finds single-event commits in one probe', async () => {
+    const { events, driver } = setup()
+    const s = events.child('s')
+    for (let i = 0; i < 200; i++) await s.append([i])
+    driver.reset()
+    expect((await collect(s.read({ from: 150, limit: 1 }))).map((e) => e.version)).toEqual([150])
+    expect(driver.calls).toEqual({ getMany: 1, scan: 1 })
+  })
+
+  it('read({ from }) uses O(log commits) probes for multi-event commits', async () => {
+    const { events, driver } = setup()
+    const s = events.child('s')
+    for (let i = 0; i < 1000; i++) await s.append(i % 3 === 0 ? [i, i] : [i])
+    driver.reset()
+    const [first] = await collect(s.read({ from: 777, limit: 1 }))
+    expect(first?.version).toBe(777)
+    // 1 guess probe + ceil(log8(1000)) search rounds, then the read scan. The guess hit an
+    // existing commit, so no head scan was needed.
+    expect(driver.calls.getMany).toBeLessThanOrEqual(1 + 4)
+    expect(driver.calls.scan).toBe(1)
+  })
+})
+
+describe('concurrency', () => {
+  it('a stale writer collides with the existing commit key', async () => {
+    const inner = new InMemoryKVDriver()
+    const a = setup([], inner).events.child('s')
+    const b = setup([], inner).events.child('s')
+    await a.append(['a1'])
+    await b.append(['b1']) // a's cached head is now stale
+    await expect(a.append(['a2'], { expectedVersion: 1 })).rejects.toMatchObject({ actualVersion: 2 })
+    // 'any' retries after refreshing the head
+    expect(await a.append(['a3'])).toEqual({ version: 3, commit: 3 })
+    expect((await collect(b.read())).map((e) => e.data)).toEqual(['a1', 'b1', 'a3'])
+  })
+
+  it('a stale cached head is refreshed before rejecting expectedVersion', async () => {
+    const inner = new InMemoryKVDriver()
+    const a = setup([], inner).events.child('s')
+    const b = setup([], inner).events.child('s')
+    await a.append(['a1'])
+    await b.append(['b1'])
+    expect(await a.append(['a2'], { expectedVersion: 2 })).toEqual({ version: 3, commit: 3 })
+  })
+
+  it('interleaved writers from two instances produce a dense, non-overlapping log', async () => {
+    const inner = new InMemoryKVDriver()
+    const a = setup([], inner).events.child('s')
+    const b = setup([], inner).events.child('s')
+    await Promise.all([a.append(['a1']), b.append(['b1']), a.append(['a2', 'a3']), b.append(['b2'])])
+    const log = await collect(a.read())
+    expect(log.map((e) => e.version)).toEqual([1, 2, 3, 4, 5])
+    expect(new Set(log.map((e) => e.data))).toEqual(new Set(['a1', 'a2', 'a3', 'b1', 'b2']))
+  })
+
+  it('exactly one of several optimistic writers at the same version wins', async () => {
+    const inner = new InMemoryKVDriver()
+    const writers = [0, 1, 2, 3].map(() => setup([], inner).events.child('s'))
+    const results = await Promise.allSettled(writers.map((w, i) => w.append([i], { expectedVersion: 0 })))
+    expect(results.filter((r) => r.status === 'fulfilled')).toHaveLength(1)
+    expect(results.filter((r) => r.status === 'rejected').every((r) => (r as PromiseRejectedResult).reason instanceof ConcurrencyError)).toBe(true)
+  })
+
+  it('without setIfAbsent, an in-process mutex serialises appends (single-writer mode)', async () => {
+    const { events, driver } = setup(['setIfAbsent'])
+    const s = events.child('s')
+    await Promise.all(Array.from({ length: 20 }, (_, i) => s.append([i])))
+    expect((await collect(s.read())).map((e) => e.version)).toEqual(Array.from({ length: 20 }, (_, i) => i + 1))
+    expect(driver.calls.setIfAbsent).toBeUndefined()
+    await expect(Promise.all([s.append(['x'], { expectedVersion: 20 }), s.append(['y'], { expectedVersion: 20 })])).rejects.toBeInstanceOf(ConcurrencyError)
+  })
+})