diff --git a/CLAUDE.md b/CLAUDE.md index 72263e3..0a8cbe4 100644 --- a/CLAUDE.md +++ b/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 -- ` — 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/` diff --git a/DESIGN.md b/DESIGN.md index a5a62d4..a921750 100644 --- a/DESIGN.md +++ b/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. + ## 13. Benchmark baselines diff --git a/package.json b/package.json index b98a2b6..ce27cd5 100644 --- a/package.json +++ b/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", diff --git a/src/core/lru.ts b/src/core/lru.ts new file mode 100644 index 0000000..c66c07a --- /dev/null +++ b/src/core/lru.ts @@ -0,0 +1,30 @@ +/** Minimal bounded LRU over `Map` insertion order. */ +export class LRU { + readonly #map = new Map() + + 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 + } +} diff --git a/src/core/mutex.ts b/src/core/mutex.ts new file mode 100644 index 0000000..3df05ab --- /dev/null +++ b/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>() + + run(key: string, fn: () => Promise): Promise { + 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 + } +} diff --git a/src/events/index.ts b/src/events/index.ts new file mode 100644 index 0000000..057ee0f --- /dev/null +++ b/src/events/index.ts @@ -0,0 +1,2 @@ +export * from './types.js' +export * from './kv.js' diff --git a/src/events/kv.ts b/src/events/kv.ts new file mode 100644 index 0000000..963b2e5 --- /dev/null +++ b/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 { + 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 + /** `/~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 + 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: + * `/~e/` → `{ 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 implements EventNode { + readonly #ctx: EventContext + readonly #kv: KVNode + + /** @internal Use `KVEventStorage` and `child()`. */ + constructor(ctx: EventContext, kv: KVNode) { + this.#ctx = ctx + this.#kv = kv + } + + get path(): readonly string[] { + return this.#kv.path + } + + /** The underlying KV node. */ + get kv(): KVNode { + return this.#kv + } + + child>(key: K): KVEventNode> { + return new KVEventNode(this.#ctx, this.#kv.child(key as string).as()) + } + + as(): KVEventNode { + return this as unknown as KVEventNode + } + + append(events: readonly ValueOf[], opts?: AppendOptions): Promise + append>(key: K, events: readonly ValueOf>[], opts?: AppendOptions): Promise + async append(...args: unknown[]): Promise { + 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>> + read>(key: K, opts?: ReadOptions): AsyncIterable>>> + async *read(...args: unknown[]): AsyncGenerator> { + 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 + version>(key: K): Promise + async version(...args: unknown[]): Promise { + const stream = args.length === 0 ? this.#self() : this.#child(args[0]) + return (await this.#loadHead(stream)).version + } + + keys(opts: StreamKeysOptions = {}): AsyncIterable> { + return this.#kv.keys(opts) as AsyncIterable> + } + + // ---------------------------------------------------------------- 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 { + 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 { + 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 { + 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 { + 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> { + 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 extends KVEventNode { + constructor(kv: KVNode, 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, + ) + } +} diff --git a/src/events/types.ts b/src/events/types.ts new file mode 100644 index 0000000..34c76f6 --- /dev/null +++ b/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 { + 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 { + readonly path: readonly string[] + child>(key: K): EventNode> + as(): EventNode + + append(events: readonly ValueOf[], opts?: AppendOptions): Promise + append>(key: K, events: readonly ValueOf>[], opts?: AppendOptions): Promise + + read(opts?: ReadOptions): AsyncIterable>> + read>(key: K, opts?: ReadOptions): AsyncIterable>>> + + /** Current version (0 = empty). Always reads storage. */ + version(): Promise + version>(key: K): Promise + + /** Child stream keys. */ + keys(opts?: StreamKeysOptions): AsyncIterable> +} + +/** The root of an event store. Native backends (EventStoreDB, Kafka, …) implement this too. */ +export type EventStorage = EventNode diff --git a/src/index.ts b/src/index.ts index 6aabc88..3f1e020 100644 --- a/src/index.ts +++ b/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' diff --git a/test/events/kv.test.ts b/test/events/kv.test.ts new file mode 100644 index 0000000..a1a393c --- /dev/null +++ b/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 /~e/', 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) + }) +})