diff --git a/DESIGN.md b/DESIGN.md index fbd9a6b..64fe994 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -57,28 +57,32 @@ bench/ ## 4. KV storage -`BaseKVStorage` primitives (absolute keys). Implementations provide the reads and the protected -`apply*` writes; the public `writeMany` / `removeMany` / `writeIfAbsent` live in the base class and -emit a `change` event after each write, so every store emits change events. +`BaseKVStorage` primitives (absolute keys). Implementations provide two reads and one protected +write; the public `writeMany(changes)` / `removeMany(keys)` live in the base class and emit a +`change` event after each write, so every store emits change events. -| Primitive | | -| ------------------------------------------------------ | ---------------------------------------------------------------------------------------------------- | -| `readMany(keys)` | values in order, `undefined` for missing | -| `scan({ gte, lt, limit, reverse, values })` | key-ordered; must keep working if keys change mid-scan (keys present throughout appear exactly once) | -| `applyWrite(entries)` → public `writeMany` | | -| `applyRemove(keys)` → public `removeMany` | | -| `applyWriteIfAbsent(entries)` → public `writeIfAbsent` | `true` = written; atomic per key (needed by events) | +| Primitive | | +| ------------------------------------------- | ---------------------------------------------------------------------------------------------------- | +| `readMany(keys)` | values in order, `undefined` for missing | +| `scan({ gte, lt, limit, reverse, values })` | key-ordered; must keep working if keys change mid-scan (keys present throughout appear exactly once) | +| `write(changes)` → public `writeMany` | `{ key, value }` sets, `{ key }` deletes, in one batch | + +There is no conditional write. Stores execute exactly the operations they are given, and the last +write wins. Data that can be written concurrently belongs in event storage (section 5), where +concurrent appends never touch the same key. Data whose key is the hash of its content is +immutable, so concurrent writes of it are identical. **Values** are JSON plus `bigint` and `Uint8Array` (as in JSON, `-0` becomes `0` and `undefined` properties are dropped). Anything crossing the network is encoded as Extended JSON (``, ``); IndexedDB stores them natively. **Events** (`EventEmitter`, shared by all nodes of a storage): `change` → -`{ changes: { key, value? }[], origin: 'local' | 'remote', id? }` (no `value` = removed), and -`resync` → changes may have been missed, so reload. +`{ changes: { key, value? }[], origin: 'local' | 'remote' }` (no `value` = removed), and +`resync` → changes may have been missed, so re-read. Node API (defined once): `child`, `as`, `get(key?)`, `getMany`, `set(value)` / `set(key, value)`, -`setMany`, `delete(key?)`, `deleteMany`, `deleteTree(key?)`, `keys({ limit, after, reverse })`, -`entries(...)`. The root has no own value. A node's own value and its children are independent keys. +`setMany`, `setFields({ a, b })` / `getFields()` (direct children as an object), `delete(key?)`, +`deleteMany`, `deleteTree(key?)`, `keys({ limit, after, reverse })`, `entries(...)`. The root has +no own value. A node's own value and its children are independent keys. - **`keys()`** is a skip-scan: one seek per child that has descendants; siblings without descendants share one iterator. It lists children that only have descendants, and skips @@ -89,23 +93,22 @@ Node API (defined once): `child`, `as`, `get(key?)`, `getMany`, `set(value)` / ` - `KVStorageInMemory`: `SortedMap` (sorted chunks of ≤1024 keys); values deep-copied. - `KVStorageIndexedDB`: one object store with string keys. Scans read in batches, each in its own transaction, resuming after the last key (IndexedDB transactions close when the code awaits - anything else). `writeIfAbsent` runs in one readwrite transaction. + anything else). A write is one readwrite transaction. - `KVStorageHttp` (client) and `KVStorageHttpServer` (wraps any KV storage; - `handle(Request) → Response`): `POST /` with Extended JSON bodies; `scan` streams - NDJSON. Unreachable server (fetch failure, 502/503/504) → `StorageUnavailableError`. The server - exposes the whole underlying store (primitives use absolute keys). + `handle(Request) → Response`): `POST /readMany`, `/write`, `/scan` with Extended JSON bodies; + `scan` streams NDJSON. Unreachable server (fetch failure, 502/503/504) → + `StorageUnavailableError`. The server executes exactly what clients send and never reads values + for decisions, so values can be encrypted end to end. - `KVStorageComposite`: see 4.2. -- **Change stream (SSE).** `POST /events` with `{ paths }` streams changes to those keys and their - subtrees as `event: change` with id `-` (epoch = server instance). The server keeps the last - `historySize` (10 000) change batches. - - A client reconnecting with `Last-Event-ID` gets the batches it missed, or `event: resync` if it - can't be resumed (another epoch, or too old). - - Every catch-up ends with `event: ready`. - - Keep-alive comments are sent every 15 s. - - `KVStorageHttp.subscribe(path = this.key)` / `unsubscribe(path)` manage the subscribed paths. - Each change reconnects the stream (vendored `SSESession`, backoff) with the new list, resuming - from `lastEventId`. The client emits `change` (`origin: 'remote'`) and `resync`, and `subscribe` - resolves once caught up. A 401/403 rejects `subscribe` instead of retrying. +- **Change stream (SSE).** `POST /events` with `{ paths }` streams every change to those keys and + their subtrees as `event: change` (data: the changes), including the subscriber's own writes. + `event: ready` follows each connection; keep-alive comments are sent every 15 s. The server keeps + no history: nothing is replayed after a reconnect. + - `KVStorageHttp.subscribe(path = this.key)` / `unsubscribe(path)` / `close()` manage the + subscribed paths. Each change reconnects the stream (vendored `SSESession`, backoff) with the + new list; `subscribe` resolves once connected. A 401/403 rejects instead of retrying. + - The client emits each pushed batch as `change` (`origin: 'remote'`), and `resync` whenever the + stream reconnects after a drop, since changes may have been missed meanwhile. - Limitation: a server only sees writes made through its own storage instance, so several server processes over one database need a shared pub/sub (not built). - **Wrappers** (e.g. a future encryption layer) extend `BaseKVStorage` and delegate the primitives. @@ -130,83 +133,67 @@ development and tests). It can't be left open by omission. dependencies; Node 22+ and current browsers; keys persist as PKCS #8). Others, e.g. secp256k1 via libauth, implement `RequestSigner` (`algorithm`, `publicKey`, `sign`) and `SignatureVerifier` (`algorithm`, `verify`). The server can accept several algorithms at once. -- **Authorization (403): the public key is the user id.** Every key in `readMany`, `writeMany`, - `removeMany` and `writeIfAbsent`, and every subscription path, must be `` or under +- **Authorization (403): the public key is the user id.** Every key in `readMany` and `write`, and + every subscription path, must be `` or under `/`. A scan must stay within `[, 0)`, i.e. the key and its subtree (escaping guarantees no other key falls in that range, including lookalikes such as `x`). -- **Scoped events.** Each subscriber only receives changes under its own (authorized) paths, live - and when resuming. Event ids are a global sequence, so a client can infer the server's overall - write volume, though nothing about the contents. +- **Scoped events.** Each subscriber only receives changes under its own (authorized) paths. - Proxies must preserve the URL path, because the signature covers it. -### 4.2 `KVStorageComposite({ source, prefix?, replicas, retry? })` +### 4.2 `KVStorageComposite({ source, replicas, queue?, prefix?, retryMs? })` Keeps local copies of a source (e.g. `KVStorageHttp`) so reads are local and the app works offline. With `prefix` (e.g. the user's public key) only that key and its subtree are synced, cached and subscribed to; keys outside it throw `RangeError`. Without it, the whole source is copied, which -is only appropriate when the client may read all of it. Replicas are listed fastest first (e.g. `[inMemory, indexedDb]`). The last replica is the -persistent one: it seeds the others on start and stores the composite's internal keys under `~q/` -(hidden from reads): `~q/meta` (`{ lastEventId }`, present once a sync has completed) and the -outbox `~q/o/`. +is only appropriate when the client may read all of it. -- **Start** (lazy, on first use): - 1. Load the outbox and copy the persistent replica into the faster ones. - 2. If a previous sync exists, serve reads locally straight away. - 3. Watch the source, resuming from `lastEventId`. - 4. Run a full sync only when there's no resume point, when the source sends `resync`, or when the - source can't be watched. - 5. Before the first sync completes, reads go to the source. -- **Remote changes** (and full syncs) are applied in order through one queue. A change that arrives - during a full sync waits behind it, so the end state always matches the source. `lastEventId` is - saved after each applied batch. Changes the replicas already hold are skipped. -- **Full sync** merge-scans the source against the persistent replica (both key-ordered) and applies - the difference to every replica in batches of 500. -- **Writes** go to the source, then every replica. - - On `StorageUnavailableError` (or while anything is queued, to keep order), the write is applied - locally and appended to the outbox. The outbox is replayed in order with exponential backoff, - and also whenever the source shows signs of life. Last writer wins on conflicts. - - Remote changes to keys with queued writes are ignored until those writes are replayed. - - `writeIfAbsent` always needs the source. -- **Echo tracking.** Each write sent to a watched source expects one echo per key. A remote change - to a key with an outstanding echo is consumed instead of applied. The stream is in server order, - so that change is either the echo or older than it, and this avoids briefly showing a stale value. - A full sync clears the counts. -- Write through the composite, not directly through the same source instance (its `local`-origin - changes are not mirrored). -- `start()`, `synced()` (complete copy, all received changes applied), `queued` (outbox size), - `close()`. - -**Structure.** The composite only orchestrates (what to read from, when to sync, whether to queue a -write). The work is done by small classes: - -| Class | Job | -| --------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | -| `KVReplicaSet` | the replicas of one `KeyRange`: seed the faster ones, apply changes, scan locally (hiding `~q/` bookkeeping), diff against the source, sync metadata | -| `KVOutbox` | the persisted offline queue (`~q/o/`), its pending-key counts, and in-order replay with backoff | -| `KeyRange` (utils) | a key plus its subtree: `contains`, `clamp` scan bounds, `covers` a scan; also used by the server's authorization | -| `KeyCounter` (utils) | per-key counts: pending outbox writes and expected echoes | -| `SerialQueue` (utils) | one-at-a-time tasks, optionally per key: applying remote changes, subscription updates, event-sourced writes | -| `RetryTimer` (utils) | backoff retries, one timer per job (full sync and outbox replay retry independently) | +- **Replicas** are listed fastest first (e.g. `[inMemory, indexedDb]`); the last one seeds the + others on start. Reads come from the first. +- **The queue** is any KV store (default: in memory). It holds writes the source couldn't take + (`writes/`) and whether a sync has completed (`synced`). Use a persistent node outside the + synced range, e.g. `indexedDb.child('~sync')` with `prefix: userKey`; a queue inside the range is + refused, since a sync would delete it. +- **Writes** are applied to every replica, then sent to the source, exactly as given. + - Applying locally first means the source relaying the write back finds nothing new. + - A write the source refuses (e.g. 403) is undone locally and rethrown. + - On `StorageUnavailableError`, or while a sync runs or writes are queued (to keep order), the + write goes to the queue and is sent later. The last write to reach the source wins. +- **Remote changes** pushed by the source are applied and announced (`change`, `origin: 'remote'`) + only where they differ from what the first replica holds. No write tagging or echo tracking is + needed: echoes of this client's writes change nothing. +- **`sync()`**: replay the queue in order (send, then re-apply locally) → merge-scan the source + against the last replica (both key-ordered), applying the difference in batches of 500 → + replay again (writes made meanwhile) → mark `synced`. It runs on start, on `resync` (the stream + reconnected), and `retryMs` after a failure while offline. +- **Reads** wait for the first sync. Until one has completed, they go to the source. With a + persistent queue, a restart reads locally straight away (offline too). +- `start()` (also lazy on first use), `sync()`, `queued`, `close()`. +- Limitations: catching up re-scans the whole range (no server history); a remote change can + overwrite a value whose write is still queued until that write is replayed (last write wins); + replicas, source and queue must be root stores (the primitives use absolute keys). ## 5. Event storage -`BaseEventStorage` primitives over stream paths: `appendTo`, `readFrom`, `versionOf`, `streamsOf`. -Node API: `append(events, opts?)` / `append(key, events, opts?)`, `read(opts?)` / `read(key, opts?)`, -`version(key?)`, `keys()`. The root is not a stream. An empty stream is version 0; the first event is 1. -`expectedVersion: number | 'any'` (default `'any'`). +`BaseEventStorage` primitives over stream paths: `appendTo(stream, events)` → ids, +`readFrom(stream, { after, limit, reverse })`, `streamsOf`. Node API: `append(events)` / +`append(key, events)` → ids, `read(opts?)` / `read(key, opts?)` → `{ id, at, data }`, `keys()`. +The root is not a stream. -**`EventStorageKV`** (on any KV storage): +**Appends never conflict.** Concurrent writers each add their own events, and every reader sees all +of them in the same order. There are no versions and no `expectedVersion`; folding events into +state is where concurrent changes meet (see event-sourced documents). -- One key per append, densely numbered: `/~e/` → `{ firstVersion, at, events }`. - Keying by commit number (not version) means any writer with a stale view collides with the - existing key. -- Append: head from a bounded cache (else one reverse scan, `limit: 1`) → check `expectedVersion` - (re-reading the head before rejecting, in case the cache is stale) → `writeIfAbsent`. On a lost race: - `ConcurrencyError` if a version was expected, otherwise retry (default 5). -- `read({ from })`: commit `n` starts at a version ≥ n, so try commit `from` first (exact for - single-event commits), else binary-search; then forward-scan. -- Multi-writer safe on any store, because `writeIfAbsent` is a required primitive. +**`EventStorageKV`** (on any KV storage): each event is one plain KV write, +`/~e/_` → the event. + +- Keys sort by time, so a stream reads in time order with one scan; `after` is an exclusive id. +- Events from different writers in the same millisecond order by hash: arbitrary, but the same for + every reader. One instance never reuses a time, so its own appends keep their order. +- The id can be recomputed from the content, and re-sending an event (e.g. a retried request) + writes the same key, so it can't be duplicated. +- An append is one `writeMany`, so it works through `KVStorageComposite` offline and syncs like any + other data. ## 6. Document storage @@ -240,14 +227,14 @@ Node API: `get(key?)`, `getMany`, `set(doc)` / `set(key, doc)`, `insert(doc)` - `EventSourcedDocumentStorage(events, projections)`: each document's `DocChange`s are appended to its own stream. Current state lives in **any** `BaseDocumentStorage` used as the projection store, which serves all reads, queries and indexes. - - Projections carry `_version` (reserved; stripped from results), used as `expectedVersion` on the - next append. - - If a write crashes between append and projection update, the next write conflicts, rebuilds the - projection from the stream, and retries. `catchUp(key?)` rebuilds on demand. - - Missing projections with stream history (deleted documents, crashed first writes) are rebuilt - before writing. - - Writes to one document are serialised in-process, so retries only happen across processes. - - Extras: `history(key?)`, `getAt(version)` / `getAt(key, version)`. Children keep these methods + - A write reads the projection, resolves the change (`$inc` → `set: { n: 6 }`), appends it, and + applies it to the projection. + - Concurrent edits to different fields both survive; the same field ends with the later event's + value. (Increments from two devices don't add up: each resolves to a final value.) + - Events written elsewhere (another device, arriving through sync) are folded in by + `catchUp(key?)`, which rebuilds the projection from the whole stream. It also repairs a + projection left behind by a crash between append and projection write. + - Extras: `history(key?)`, `getAt(eventId)` / `getAt(key, eventId)`. Children keep these methods in their types through interface declaration merging (types only). ## 7. Changes from the original brief @@ -260,14 +247,16 @@ Node API: `get(key?)`, `getMany`, `set(doc)` / `set(key, doc)`, `insert(doc)` - Separator fixed to `/`, and every character ≤ `/` is escaped so child keys stay contiguous. This removed the subtree-order workaround in child listing. - Event-sourced documents are events + a projection store instead of a bespoke projection format. -- `writeIfAbsent` is a required primitive (no single-writer mutex fallback; no optional `batch` / - `deleteRange`). +- No conditional writes and no versions: KV stores execute exactly what they're given (so a server + never needs to read values, which may be encrypted), and events get time + content-hash ids so + concurrent appends never collide. Optimistic concurrency (`expectedVersion`) was dropped. - `tree()` removed (use `entries()` for field-per-key reads). ## 8. Open questions -- Composite: lazily syncing parts of a large subtree on first read; conflict policies beyond last - writer wins for offline writes. +- Composite: lazily syncing parts of a large subtree on first read; catching up without a full + re-scan (would need server-side history or a change log in event storage). +- Event-sourced documents: subscribing to event streams to run `catchUp` automatically. - Authorization beyond "public key = user id" (sharing a subtree with other keys). - Multi-process servers need a shared change feed (pub/sub) for SSE. @@ -285,9 +274,10 @@ Node API: `get(key?)`, `getMany`, `set(doc)` / `set(key, doc)`, `insert(doc)` | KV `get` / `set` @ 100k keys | 2.0 / 2.3 µs | | KV `getMany` × 100 | 25–35 µs | | `keys()` on 1k children × 100 descendants (skip-scan vs full scan) | 2.8 ms vs 34 ms | -| Event `append` (warm cache) | 7.8 µs | -| Event `read` 100 from version 5 000 of 10k | 131 µs | -| Event `read` full 10k stream | 18 ms | +| Event `append` (one SHA-256 via WebCrypto + one write) | 72 µs | +| Event `read` 100 from the middle of 10k | 119 µs | +| Event `read` full 10k stream | 11 ms | +| Event-sourced document `update ($inc)` | 86 µs | | Documents in-memory @ 10k: `get` / `update` | 1.8 / 6.6 µs | | Documents in-memory: `find` indexed (~100 hits) / scan (~200 of 10k) | 0.20 / 11 ms | | Event-sourced @ 10k: `get` / `update` | 1.6 / 17 µs | diff --git a/bench/event-storage.bench.ts b/bench/event-storage.bench.ts index 0c1a695..571a889 100644 --- a/bench/event-storage.bench.ts +++ b/bench/event-storage.bench.ts @@ -4,15 +4,19 @@ import { KVStorageInMemory } from '../src/kv-storage/kv-storage-in-memory.js' const events = new EventStorageKV(new KVStorageInMemory()) const streams = events.child('streams') -for (let i = 0; i < 10_000; i++) await streams.append('long', [{ i }]) +const ids = await streams.append( + 'long', + Array.from({ length: 10_000 }, (_, i) => ({ i })), +) +const middle = ids[5000] let s = 0 describe('events', () => { - bench('append 1 event (warm cache)', async () => { + bench('append 1 event', async () => { await streams.append(`s${s++ % 1000}`, [{ type: 'e' }]) }) - bench('read 100 events from version 5000 of 10k', async () => { - for await (const _ of streams.read('long', { from: 5000, limit: 100 })) { + bench('read 100 events from the middle of 10k', async () => { + for await (const _ of streams.read('long', { after: middle, limit: 100 })) { } }) bench('read full 10k-event stream', async () => { diff --git a/bench/kv-storage.bench.ts b/bench/kv-storage.bench.ts index 8db217c..01d8141 100644 --- a/bench/kv-storage.bench.ts +++ b/bench/kv-storage.bench.ts @@ -1,9 +1,10 @@ import { bench, describe } from 'vitest' +import type { KVChange } from '../src/kv-storage/base-kv-storage.js' import { KVStorageInMemory } from '../src/kv-storage/kv-storage-in-memory.js' for (const n of [10_000, 100_000]) { const db = new KVStorageInMemory() - await db.writeMany(Array.from({ length: n }, (_, i) => [`users/u${i}`, { name: `user ${i}`, n: i }])) + await db.writeMany(Array.from({ length: n }, (_, i) => ({ key: `users/u${i}`, value: { name: `user ${i}`, n: i } }))) const users = db.child('users') const keys = Array.from({ length: 100 }, (_, i) => `u${(i * 7919) % n}`) let i = 0 @@ -21,10 +22,10 @@ for (const n of [10_000, 100_000]) { } const tree = new KVStorageInMemory() -const writes: [string, number][] = [] +const writes: KVChange[] = [] for (let c = 0; c < 1000; c++) { - writes.push([`n/c${c}`, c]) - for (let d = 0; d < 100; d++) writes.push([`n/c${c}/d${d}`, d]) + writes.push({ key: `n/c${c}`, value: c }) + for (let d = 0; d < 100; d++) writes.push({ key: `n/c${c}/d${d}`, value: d }) } await tree.writeMany(writes) diff --git a/src/document-storage/document-storage-kv.ts b/src/document-storage/document-storage-kv.ts index 7a915b4..85a512a 100644 --- a/src/document-storage/document-storage-kv.ts +++ b/src/document-storage/document-storage-kv.ts @@ -26,7 +26,7 @@ export class DocumentStorageKV extends BaseDocumentSto const resolved = change(current) if (!resolved) return current const next = DocumentUpdate.apply(current, resolved) - if (next) await this.kv.writeMany([[key, next]]) + if (next) await this.kv.writeMany([{ key, value: next }]) else await this.kv.removeMany([key]) return next } diff --git a/src/document-storage/event-sourced-document-storage.ts b/src/document-storage/event-sourced-document-storage.ts index 1a3f957..360cfad 100644 --- a/src/document-storage/event-sourced-document-storage.ts +++ b/src/document-storage/event-sourced-document-storage.ts @@ -2,16 +2,9 @@ import type { BaseEventStorage, StoredEvent } from '../event-storage/base-event- import type { Doc } from '../utils/document-path.js' import type { DocumentQuery, FindResult } from '../utils/document-query.js' import { DocumentUpdate, type DocChange } from '../utils/document-update.js' -import { ConcurrencyError } from '../utils/errors.js' -import { SerialQueue } from '../utils/serial-queue.js' import type { AnySpec, ChildOf, KeyOf, Spec, ValueAt } from '../utils/schema.js' import { BaseDocumentStorage } from './base-document-storage.js' -export interface EventSourcedDocumentStorageOptions { - /** Retries after a write conflict, each preceded by a catch-up. Default 5. */ - maxRetries?: number -} - // Types only: children of an event-sourced store are event-sourced stores too (child() creates // them with Object.create, so they share this class's prototype). export interface EventSourcedDocumentStorage { @@ -19,103 +12,66 @@ export interface EventSourcedDocumentStorage { } /** - * Documents as event streams: each document's changes are appended to its own stream, and the - * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all - * reads and queries. + * Documents as event streams: each change to a document is appended to its own stream, and the + * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all reads + * and queries. * - * Projections carry a `_version` field (stripped from results) used as `expectedVersion` on the - * next append. If a write crashes between appending and updating the projection, the next - * write's append conflicts, the projection is rebuilt from the stream, and the write retries. - * `_version` is therefore reserved in documents. + * A document's state is its events folded in order. Changes are stored as resolved values + * (`$inc` becomes `set: { n: 6 }`), so two devices editing the same field concurrently end with + * the later event's value; edits to different fields both survive. + * + * The projection is updated by this store's own writes. Events written elsewhere (another device, + * arriving through sync) are folded in by `catchUp(key)`, which rebuilds the document from its + * whole stream; call it when such events arrive. */ export class EventSourcedDocumentStorage extends BaseDocumentStorage { - private readonly maxRetries: number - /** Writes to one document run one at a time, so writes in one process don't conflict with each other. */ - private readonly writes = new SerialQueue() - constructor( private readonly events: BaseEventStorage, private readonly projections: BaseDocumentStorage, - options: EventSourcedDocumentStorageOptions = {}, ) { super() - this.maxRetries = options.maxRetries ?? 5 } - async readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> { - return (await this.projections.readMany(paths)).map((doc) => EventSourcedDocumentStorage.strip(doc)) + readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> { + return this.projections.readMany(paths) } - async *query(parent: readonly string[], query: DocumentQuery): AsyncGenerator { - for await (const { key, data } of this.projections.query(parent, query)) yield { key, data: EventSourcedDocumentStorage.strip(data)! } + query(parent: readonly string[], query: DocumentQuery): AsyncIterable { + return this.projections.query(parent, query) } - mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise { - return this.writes.run(() => this.write(path, change), JSON.stringify(path)) + async mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise { + const [current] = await this.projections.readMany([path]) + const resolved = change(current) + if (!resolved) return current + await this.events.appendTo(path, [resolved]) + // Apply the change itself (not the state computed above), so concurrent writes in this + // process each land on the latest projection. + return this.projections.mutate(path, () => resolved) } - private async write(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise { - for (let attempt = 0; ; attempt++) { - const [stored] = await this.projections.readMany([path]) - const version = (stored?._version as number | undefined) ?? (await this.events.versionOf(path)) - if (!stored && version > 0) { - // No projection but the stream has history (deleted, or a crashed first write): rebuild. - if ((await this.rebuild(path)) !== undefined) continue - } - const current = EventSourcedDocumentStorage.strip(stored) - const resolved = change(current) - if (!resolved) return current - const next = DocumentUpdate.apply(current, resolved) - try { - await this.events.appendTo(path, [resolved], version) - } catch (error) { - if (!(error instanceof ConcurrencyError) || attempt >= this.maxRetries) throw error - await this.rebuild(path) - continue - } - await this.project(path, next, version + 1) - return next - } - } - - /** Events of this document (or a child). */ + /** Events of this document (or a child), oldest first. */ history(key?: KeyOf): AsyncIterable> { return this.events.readFrom(this.pathOf(arguments.length > 0, key), {}) as AsyncIterable> } - /** State as of `version`: `getAt(version)` for this document, `getAt(key, version)` for a child. */ - async getAt>(...args: [version: number] | [key: K, version: number]): Promise | undefined> { - const path = this.pathOf(args.length > 1, args[0]) + /** State right after event `eventId`: `getAt(eventId)` for this document, `getAt(key, eventId)` for a child. */ + async getAt>(...args: [eventId: string] | [key: K, eventId: string]): Promise | undefined> { + const until = args[args.length - 1] as string let state: Doc | undefined - for await (const event of this.events.readFrom(path, { to: args[args.length - 1] as number })) { + for await (const event of this.events.readFrom(this.pathOf(args.length > 1, args[0]), {})) { + if (event.id > until) break state = DocumentUpdate.apply(state, event.data as DocChange) } return state as ValueAt | undefined } - /** Rebuilds the projection of this document (or a child) from its stream. Returns the state. */ - catchUp(key?: KeyOf): Promise { - return this.rebuild(this.pathOf(arguments.length > 0, key)) - } - - private async rebuild(path: readonly string[]): Promise { + /** Rebuilds the projection of this document (or a child) from all its events. Returns the state. */ + async catchUp(key?: KeyOf): Promise { + const path = this.pathOf(arguments.length > 0, key) let state: Doc | undefined - let version = 0 - for await (const event of this.events.readFrom(path, {})) { - state = DocumentUpdate.apply(state, event.data as DocChange) - version = event.version - } - await this.project(path, state, version) - return state - } - - private async project(path: readonly string[], state: Doc | undefined, version: number): Promise { - await this.projections.mutate(path, () => (state ? { op: 'set', doc: { ...state, _version: version } } : { op: 'delete' })) - } - - private static strip(doc: Doc | undefined): Doc | undefined { - if (!doc) return undefined - const { _version, ...state } = doc + for await (const event of this.events.readFrom(path, {})) state = DocumentUpdate.apply(state, event.data as DocChange) + await this.projections.mutate(path, () => (state ? { op: 'set', doc: state } : { op: 'delete' })) return state } } diff --git a/src/event-storage/base-event-storage.ts b/src/event-storage/base-event-storage.ts index 4daaabf..9eaef89 100644 --- a/src/event-storage/base-event-storage.ts +++ b/src/event-storage/base-event-storage.ts @@ -1,27 +1,17 @@ import { Keys } from '../utils/keys.js' import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueAt, ValueOf } from '../utils/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. */ - version: number -} - export interface ReadOptions { - /** First version to read (inclusive). Default 1. */ - from?: number - /** Last version to read (inclusive). Default: the head. */ - to?: number + /** Start after this event id (exclusive), in the direction of travel. */ + after?: string limit?: number + reverse?: boolean } export interface StoredEvent { - version: number - /** Append timestamp (ms since epoch). */ + /** Unique, time-ordered id (see the implementation for its format). */ + id: string + /** Append time (ms since epoch). */ at: number data: E } @@ -34,16 +24,20 @@ export interface StreamKeysOptions { /** * Append-only event streams. Every node is a stream (its own events) and a parent of child - * streams. Implementations provide four primitives over stream paths; the user-facing, + * streams. Implementations provide three primitives over stream paths; the user-facing, * self-or-child API is defined here once. + * + * Appends never conflict: concurrent writers each add their own events, and every reader sees all + * of them in the same (time) order. Folding events into state is where concurrent changes meet. */ export abstract class BaseEventStorage { /** Unescaped path of this node (`[]` for the root, which is not a stream). */ readonly path: readonly string[] = [] - abstract appendTo(stream: readonly string[], events: unknown[], expectedVersion: number | 'any'): Promise + /** Appends events in order; returns their ids. */ + abstract appendTo(stream: readonly string[], events: unknown[]): Promise + /** Events in id (time) order. */ abstract readFrom(stream: readonly string[], options: ReadOptions): AsyncIterable> - abstract versionOf(stream: readonly string[]): Promise abstract streamsOf(parent: readonly string[], options: StreamKeysOptions): AsyncIterable child>(key: K): BaseEventStorage> { @@ -59,15 +53,12 @@ export abstract class BaseEventStorage { return this as unknown as BaseEventStorage } - /** `append(events, options?)` on this stream; `append(key, events, options?)` on a child stream. */ - append>( - ...args: [events: ValueOf[], options?: AppendOptions] | [key: K, events: ValueOf>[], options?: AppendOptions] - ): Promise { + /** `append(events)` to this stream; `append(key, events)` to a child stream. Returns the event ids. */ + append>(...args: [events: ValueOf[]] | [key: K, events: ValueOf>[]]): Promise { const hasKey = !Array.isArray(args[0]) const events = args[hasKey ? 1 : 0] - const options = (args[hasKey ? 2 : 1] ?? {}) as AppendOptions if (!Array.isArray(events)) return Promise.reject(new TypeError('append() expects an array of events')) - return this.withStream(hasKey, args[0], (stream) => this.appendTo(stream, events, options.expectedVersion ?? 'any')) + return this.withStream(hasKey, args[0], (stream) => this.appendTo(stream, events)) } /** `read(options?)` this stream; `read(key, options?)` a child stream. */ @@ -77,11 +68,6 @@ export abstract class BaseEventStorage { yield* this.readFrom(stream, ((hasKey ? args[1] : args[0]) ?? {}) as ReadOptions) as AsyncIterable>> } - /** Current version of this stream (or a child's); 0 when empty. */ - version(key?: KeyOf): Promise { - return this.withStream(arguments.length > 0, key, (stream) => this.versionOf(stream)) - } - /** Keys of child streams. */ keys(options: StreamKeysOptions = {}): AsyncIterable> { return this.streamsOf(this.path, options) as AsyncIterable> diff --git a/src/event-storage/event-storage-kv.ts b/src/event-storage/event-storage-kv.ts index aa5e00a..6487e04 100644 --- a/src/event-storage/event-storage-kv.ts +++ b/src/event-storage/event-storage-kv.ts @@ -1,151 +1,86 @@ import type { BaseKVStorage } from '../kv-storage/base-kv-storage.js' -import { ConcurrencyError } from '../utils/errors.js' +import { ExtendedJson } from '../utils/extended-json.js' +import { Hex } from '../utils/hex.js' import { Keys } from '../utils/keys.js' -import type { AnySpec, Spec } from '../utils/schema.js' -import { BaseEventStorage, type AppendResult, type ReadOptions, type StoredEvent, type StreamKeysOptions } from './base-event-storage.js' - -/** Stored value of one commit key. */ -interface Commit { - firstVersion: number - at: number - events: unknown[] -} - -interface Head { - commit: number - version: number -} +import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js' +import { BaseEventStorage, type ReadOptions, type StoredEvent, type StreamKeysOptions } from './base-event-storage.js' export interface EventStorageKVOptions { - /** Retries after losing a commit race when `expectedVersion` is `'any'`. Default 5. */ - maxRetries?: number - /** Max streams whose head is cached. Default 10 000. */ - cacheSize?: number now?: () => number } +// Types only: children of an EventStorageKV are EventStorageKV nodes too (child() creates them with +// Object.create, so they share this class's prototype). +export interface EventStorageKV { + child>(key: K): EventStorageKV> +} + /** - * Event storage on any KV storage. Each append is one commit key, densely numbered: - * `/~e/` → `{ firstVersion, at, events }`. Commits are claimed with - * `writeIfAbsent`, so a writer with a stale view always collides with the existing key. + * Event storage on any KV storage. Each event is one plain write: + * `/~e/_` → the event, where `hash` is the SHA-256 (hex) of the time and + * the event's canonical Extended JSON. + * + * - Concurrent appends from different writers get different keys, so all are kept. + * - Keys sort by time, so a stream reads in time order; equal times order by hash, the same way + * for every reader. One instance never reuses a time, so its own appends keep their order. + * - The key can be recomputed from the content, and writing the same event again writes the same + * key (e.g. a retried request), so it can't be duplicated. */ export class EventStorageKV extends BaseEventStorage { - private readonly heads = new Map() - private readonly maxRetries: number - private readonly cacheSize: number private readonly now: () => number + /** Mutable state lives here: child nodes share this object (see `child()`). */ + private readonly state = { lastTime: 0 } constructor( private readonly kv: BaseKVStorage, options: EventStorageKVOptions = {}, ) { super() - this.maxRetries = options.maxRetries ?? 5 - this.cacheSize = options.cacheSize ?? 10_000 this.now = options.now ?? Date.now } - async appendTo(stream: readonly string[], events: unknown[], expectedVersion: number | 'any'): Promise { - const base = this.commitPrefix(stream) - for (let attempt = 0; ; attempt++) { - let head = this.heads.get(base) ?? (await this.loadHead(base)) - // The cache may be stale; only trust a mismatch after re-reading. - if (expectedVersion !== 'any' && head.version !== expectedVersion) head = await this.loadHead(base) - if (expectedVersion !== 'any' && head.version !== expectedVersion) throw new ConcurrencyError(base, expectedVersion, head.version) - if (events.length === 0) return { version: head.version } - - const commit: Commit = { firstVersion: head.version + 1, at: this.now(), events } - const [written] = await this.kv.writeIfAbsent([[base + EventStorageKV.pad(head.commit + 1), commit]]) - if (written) { - const next = { commit: head.commit + 1, version: head.version + events.length } - this.cacheHead(base, next) - return { version: next.version } - } - this.heads.delete(base) - if (expectedVersion !== 'any' || attempt >= this.maxRetries) { - throw new ConcurrencyError(base, expectedVersion, (await this.loadHead(base)).version) - } - } + async appendTo(stream: readonly string[], events: unknown[]): Promise { + const base = this.eventsPrefix(stream) + const entries = await Promise.all( + events.map(async (data) => { + this.state.lastTime = Math.max(this.now(), this.state.lastTime + 1) + const id = await EventStorageKV.id(this.state.lastTime, data) + return { key: base + id, value: data } + }), + ) + await this.kv.writeMany(entries) + return entries.map(({ key }) => key.slice(base.length)) } async *readFrom(stream: readonly string[], options: ReadOptions): AsyncGenerator> { - const from = Math.max(1, options.from ?? 1) - const to = options.to ?? Infinity - let remaining = options.limit ?? Infinity - if (to < from || remaining <= 0) return - const base = this.commitPrefix(stream) - const start = from === 1 ? 1 : await this.findCommit(base, from) - if (start === 0) return - for await (const [, value] of this.kv.scan({ gte: base + EventStorageKV.pad(start), lt: this.commitEnd(base), values: true })) { - const commit = value as Commit - for (let i = 0; i < commit.events.length; i++) { - const version = commit.firstVersion + i - if (version < from) continue - if (version > to || remaining-- <= 0) return - yield { version, at: commit.at, data: commit.events[i] } - } + const base = this.eventsPrefix(stream) + const { after, reverse = false, limit } = options + // `after` is exclusive: start at it and skip it. + const bounds = reverse + ? { gte: base, lt: after === undefined ? base.slice(0, -1) + Keys.afterSeparator : base + after } + : { gte: base + (after ?? ''), lt: base.slice(0, -1) + Keys.afterSeparator } + let remaining = limit ?? Infinity + for await (const [key, data] of this.kv.scan({ ...bounds, reverse, values: true })) { + if (remaining <= 0) return + const id = key.slice(base.length) + if (id === after) continue + remaining-- + yield { id, at: Number(id.slice(0, 15)), data } } } - async versionOf(stream: readonly string[]): Promise { - return (await this.loadHead(this.commitPrefix(stream))).version - } - async *streamsOf(parent: readonly string[], options: StreamKeysOptions): AsyncGenerator { yield* parent.reduce((node, segment) => node.child(segment), this.kv).keys(options) } - /** Last commit of a stream: one reverse scan with limit 1. Refreshes the cache. */ - private async loadHead(base: string): Promise { - for await (const [key, value] of this.kv.scan({ gte: base, lt: this.commitEnd(base), reverse: true, limit: 1, values: true })) { - const commit = value as Commit - const head = { commit: Number(key.slice(base.length)), version: commit.firstVersion + commit.events.length - 1 } - this.cacheHead(base, head) - return head - } - return { commit: 0, version: 0 } + /** `_`. */ + static async id(time: number, data: unknown): Promise { + const stamp = String(time).padStart(15, '0') + const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(stamp + ExtendedJson.canonical(data))) + return `${stamp}_${Hex.encode(new Uint8Array(digest))}` } - /** - * Commit containing `version`, or 0 if the stream is shorter. Commit `n` starts at a version - * >= n, so it's at most `version`: try that first (exact when commits hold one event each), - * then binary-search the commits below it. - */ - private async findCommit(base: string, version: number): Promise { - const firstVersion = async (commit: number) => ((await this.kv.readMany([base + EventStorageKV.pad(commit)]))[0] as Commit | undefined)?.firstVersion - const guess = await firstVersion(version) - if (guess !== undefined && guess <= version) return version - let hi = version - 1 - if (guess === undefined) { - const head = await this.loadHead(base) - if (head.version < version) return 0 - hi = head.commit - } - let lo = 1 - while (lo < hi) { - const mid = (lo + hi + 1) >>> 1 - if ((await firstVersion(mid))! <= version) lo = mid - else hi = mid - 1 - } - return lo - } - - private cacheHead(base: string, head: Head): void { - this.heads.delete(base) - this.heads.set(base, head) - if (this.heads.size > this.cacheSize) this.heads.delete(this.heads.keys().next().value!) - } - - private commitPrefix(stream: readonly string[]): string { - const prefix = this.kv.prefix + Keys.join(stream) - return `${prefix}/~e/` - } - - private commitEnd(base: string): string { - return base.slice(0, -1) + Keys.afterSeparator - } - - private static pad(commit: number): string { - return String(commit).padStart(12, '0') + private eventsPrefix(stream: readonly string[]): string { + return `${this.kv.prefix}${Keys.join(stream)}/~e/` } } diff --git a/src/index.ts b/src/index.ts index 29b67db..db472e8 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,8 +4,6 @@ export * from './kv-storage/kv-storage-http.js' export * from './kv-storage/kv-storage-http-server.js' export * from './kv-storage/kv-storage-indexeddb.js' export * from './kv-storage/kv-storage-composite.js' -export * from './kv-storage/kv-replica-set.js' -export * from './kv-storage/kv-outbox.js' export * from './event-storage/base-event-storage.js' export * from './event-storage/event-storage-kv.js' export * from './document-storage/base-document-storage.js' @@ -19,10 +17,7 @@ export * from './utils/errors.js' export * from './utils/event-emitter.js' export * from './utils/extended-json.js' export * from './utils/hex.js' -export * from './utils/key-counter.js' export * from './utils/key-range.js' -export * from './utils/retry-timer.js' -export * from './utils/serial-queue.js' export * from './utils/request-auth.js' export * from './utils/json.js' export * from './utils/keys.js' diff --git a/src/kv-storage/base-kv-storage.ts b/src/kv-storage/base-kv-storage.ts index 08dc823..8127f96 100644 --- a/src/kv-storage/base-kv-storage.ts +++ b/src/kv-storage/base-kv-storage.ts @@ -30,20 +30,22 @@ export interface KVChangeEvent { changes: KVChange[] /** `local`: made through this instance. `remote`: observed from elsewhere (e.g. pushed by a server). */ origin: 'local' | 'remote' - /** Position in the source's change stream, when it has one (used to resume). */ - id?: string } export type KVStorageEvents = { change: KVChangeEvent - /** Changes may have been missed (e.g. a resume point was too old): consumers should reload. */ - resync: { id?: string } + /** Changes may have been missed (e.g. a dropped connection was restored): consumers should re-read. */ + resync: void } /** - * Ordered key-value storage. Implementations provide five raw primitives over absolute keys - * (`readMany`, `scan`, and the protected `apply*` writes); everything else (navigation, self/child - * reads and writes, child listing, change events) is defined here once. + * Ordered key-value storage. Implementations provide three primitives over absolute keys: + * `readMany`, `scan` and the protected `write`. Everything else (navigation, self/child reads and + * writes, child listing, change events) is defined here once. + * + * Writes are plain operations: set a key, or delete it. There are no versions or conditions; + * concurrent writers to one key end with whichever write arrived last. Use event storage when + * concurrent writes must all be kept. * * Every write emits a `change` event from the storage (shared by all its child nodes). * @@ -63,31 +65,20 @@ export abstract class BaseKVStorage extends EventEmitt abstract readMany(keys: string[]): Promise /** Entries in key order (descending when `reverse`). */ abstract scan(options?: ScanOptions): AsyncIterable<[key: string, value: unknown]> - protected abstract applyWrite(entries: [key: string, value: unknown][]): Promise - protected abstract applyRemove(keys: string[]): Promise - /** Writes each entry only if its key is absent; `true` = written. Must be atomic per key. */ - protected abstract applyWriteIfAbsent(entries: [key: string, value: unknown][]): Promise + /** Applies changes in order: sets each key to its value, or deletes it when the value is `undefined`. */ + protected abstract write(changes: KVChange[]): Promise // ------------------------------------------------------------------ raw writes (absolute keys) - async writeMany(entries: [key: string, value: unknown][]): Promise { - if (entries.length === 0) return - await this.applyWrite(entries) - this.emit('change', { changes: entries.map(([key, value]) => ({ key, value })), origin: 'local' }) + /** Applies changes (`value: undefined` deletes) and emits them as a `change` event. */ + async writeMany(changes: KVChange[]): Promise { + if (changes.length === 0) return + await this.write(changes) + this.emit('change', { changes, origin: 'local' }) } async removeMany(keys: string[]): Promise { - if (keys.length === 0) return - await this.applyRemove(keys) - this.emit('change', { changes: keys.map((key) => ({ key })), origin: 'local' }) - } - - async writeIfAbsent(entries: [key: string, value: unknown][]): Promise { - if (entries.length === 0) return [] - const written = await this.applyWriteIfAbsent(entries) - const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })) - if (changes.length) this.emit('change', { changes, origin: 'local' }) - return written + await this.writeMany(keys.map((key) => ({ key }))) } // ------------------------------------------------------------------ navigation @@ -127,12 +118,24 @@ export abstract class BaseKVStorage extends EventEmitt async set>(...args: [value: ValueOf] | [key: K, value: ValueOf>]): Promise { const value = args[args.length - 1] if (value === undefined) throw new TypeError('Cannot set undefined; use delete()') - await this.writeMany([[this.keyOf(args.length > 1, args[0]), value]]) + await this.writeMany([{ key: this.keyOf(args.length > 1, args[0]), value }]) } async setMany(entries: readonly EntryOf[]): Promise { if (entries.some(([, v]) => v === undefined)) throw new TypeError('Cannot set undefined; use delete()') - await this.writeMany(entries.map(([k, v]) => [this.keyOf(true, k), v])) + await this.writeMany(entries.map(([k, value]) => ({ key: this.keyOf(true, k), value }))) + } + + /** Writes each property of `fields` as a child key (one write). Nested objects stay single values. */ + async setFields(fields: { [K in KeyString]?: ValueOf>> }): Promise { + await this.setMany(Object.entries(fields).filter(([, v]) => v !== undefined) as unknown as EntryOf[]) + } + + /** Reads the direct children that have a value into an object. */ + async getFields(): Promise<{ [K in KeyString]?: ValueOf>> }> { + const fields: Record = {} + for await (const [key, value] of this.entries()) fields[key] = value + return fields as { [K in KeyString]?: ValueOf>> } } /** Deletes this node's own value (or a child's), not descendants. */ diff --git a/src/kv-storage/kv-outbox.ts b/src/kv-storage/kv-outbox.ts deleted file mode 100644 index ead5a91..0000000 --- a/src/kv-storage/kv-outbox.ts +++ /dev/null @@ -1,91 +0,0 @@ -import { StorageUnavailableError } from '../utils/errors.js' -import { KeyCounter } from '../utils/key-counter.js' -import { KeyRange } from '../utils/key-range.js' -import type { RetryTimer } from '../utils/retry-timer.js' -import type { BaseKVStorage, KVChange } from './base-kv-storage.js' - -/** A write waiting to be sent to the source. */ -export type OutboxEntry = { type: 'write'; entries: [string, unknown][] } | { type: 'remove'; keys: string[] } - -const ENTRIES = new KeyRange('~q/o') - -/** - * Writes made while the source was unreachable, persisted as `~q/o/` in a store and - * replayed to the source in order. Replay stops at the first `StorageUnavailableError` and retries - * with backoff; any other error is logged and also retried. - */ -export class KVOutbox { - /** Queued writes per key. */ - readonly pending = new KeyCounter() - private seq = 0 - private replaying: Promise | undefined - - constructor( - private readonly store: BaseKVStorage, - private readonly send: (entry: OutboxEntry) => Promise, - private readonly retry: RetryTimer, - ) {} - - /** Number of queued writes. */ - get size(): number { - return this.pending.size - } - - /** Restores the queue left by a previous session and starts replaying it. */ - async load(): Promise { - for await (const [key, entry] of this.entries()) { - this.seq = Number(key.slice(ENTRIES.prefix.length + 1)) - this.pending.add(KVOutbox.keysOf(entry)) - } - if (this.size > 0) void this.replay() - } - - /** Queues a write and starts replaying. */ - async push(entry: OutboxEntry): Promise { - const key = `${ENTRIES.prefix}/${String(++this.seq).padStart(12, '0')}` - this.pending.add(KVOutbox.keysOf(entry)) - await this.store.writeMany([[key, entry]]) - void this.replay() - } - - /** Sends queued writes in order until the queue is empty or the source is unreachable. */ - replay(): Promise { - const run = async () => { - try { - for (let next = await this.first(); next; next = await this.first()) { - const [key, entry] = next - await this.send(entry) - await this.store.removeMany([key]) - for (const k of KVOutbox.keysOf(entry)) this.pending.take(k) - this.retry.succeeded() - } - } catch (error) { - if (!(error instanceof StorageUnavailableError)) console.error(error) - this.retry.schedule(() => void this.replay()) - } finally { - this.replaying = undefined - } - } - return (this.replaying ??= run()) - } - - static keysOf(entry: OutboxEntry): string[] { - return entry.type === 'write' ? entry.entries.map(([key]) => key) : entry.keys - } - - static changesOf(entry: OutboxEntry): KVChange[] { - return entry.type === 'write' ? entry.entries.map(([key, value]) => ({ key, value })) : entry.keys.map((key) => ({ key })) - } - - private async first(): Promise<[string, OutboxEntry] | undefined> { - for await (const entry of this.entries(1)) return entry - return undefined - } - - private async *entries(limit?: number): AsyncGenerator<[string, OutboxEntry]> { - // Only the entries themselves (`~q/o/...`), not `~q/o` itself. - for await (const [key, value] of this.store.scan({ gte: ENTRIES.prefix + '/', lt: ENTRIES.lt, limit, values: true })) { - yield [key, value as OutboxEntry] - } - } -} diff --git a/src/kv-storage/kv-replica-set.ts b/src/kv-storage/kv-replica-set.ts deleted file mode 100644 index 75a1409..0000000 --- a/src/kv-storage/kv-replica-set.ts +++ /dev/null @@ -1,105 +0,0 @@ -import { Json } from '../utils/json.js' -import { KeyRange } from '../utils/key-range.js' -import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js' - -/** Sync bookkeeping kept in the persistent replica. */ -export interface SyncMeta { - /** Resume position in the source's change stream. */ - lastEventId?: string -} - -/** Keys reserved for sync bookkeeping (`~q/...`); never synced or shown. */ -export const SYNC_INTERNAL = new KeyRange('~q') -const META_KEY = '~q/meta' -const BATCH = 500 - -/** - * Local copies of one key range of a source, fastest first. The last replica is the persistent - * one: it seeds the others and holds the sync metadata (and the outbox, see `KVOutbox`). - */ -export class KVReplicaSet { - readonly fastest: BaseKVStorage - readonly persistent: BaseKVStorage - - constructor( - private readonly replicas: BaseKVStorage[], - readonly range: KeyRange, - ) { - if (replicas.length === 0) throw new TypeError('At least one replica is required') - this.fastest = replicas[0]! - this.persistent = replicas.at(-1)! - } - - async readMeta(): Promise { - return (await this.persistent.readMany([META_KEY]))[0] as SyncMeta | undefined - } - - async writeMeta(meta: SyncMeta): Promise { - await this.persistent.writeMany([[META_KEY, meta]]) - } - - /** Copies the persistent replica into the faster ones. */ - async seed(): Promise { - const faster = this.replicas.slice(0, -1) - if (faster.length === 0) return - let batch: [string, unknown][] = [] - const flush = async () => { - await Promise.all(faster.map((replica) => replica.writeMany(batch))) - batch = [] - } - for await (const entry of this.scanPersistent()) if (batch.push(entry) >= BATCH) await flush() - await flush() - } - - async apply(changes: KVChange[]): Promise { - const writes = changes.filter((c) => c.value !== undefined).map((c): [string, unknown] => [c.key, c.value]) - const removes = changes.filter((c) => c.value === undefined).map((c) => c.key) - await Promise.all(this.replicas.map(async (replica) => (await replica.writeMany(writes), await replica.removeMany(removes)))) - } - - /** The changes whose values differ from what the replicas hold. */ - async unapplied(changes: KVChange[]): Promise { - const current = await this.fastest.readMany(changes.map(({ key }) => key)) - return changes.filter(({ value }, i) => !Json.equal(value, current[i])) - } - - /** Scans the fastest replica within the range, hiding bookkeeping keys. */ - async *scan(options: ScanOptions): AsyncGenerator<[string, unknown]> { - const scoped = this.range.clamp(options) - let remaining = options.limit ?? Infinity - if (!scoped || remaining <= 0) return - // Filter before applying the limit. - for await (const entry of this.fastest.scan({ ...scoped, limit: undefined })) { - if (SYNC_INTERNAL.contains(entry[0])) continue - yield entry - if (--remaining <= 0) return - } - } - - /** Changes that turn the replicas into `source` within the range (merge of two ordered scans). */ - async *diff(source: BaseKVStorage): AsyncGenerator { - const remote = source.scan({ gte: this.range.gte, lt: this.range.lt, values: true })[Symbol.asyncIterator]() - const local = this.scanPersistent() - let r = await remote.next() - let l = await local.next() - while (!r.done || !l.done) { - if (l.done || (!r.done && r.value[0] < l.value[0])) { - yield { key: r.value![0], value: r.value![1] } - r = await remote.next() - } else if (r.done || l.value[0] < r.value[0]) { - yield { key: l.value[0] } - l = await local.next() - } else { - if (!Json.equal(r.value[1], l.value[1])) yield { key: r.value[0], value: r.value[1] } - r = await remote.next() - l = await local.next() - } - } - } - - private async *scanPersistent(): AsyncGenerator<[string, unknown]> { - for await (const entry of this.persistent.scan({ gte: this.range.gte, lt: this.range.lt, values: true })) { - if (!SYNC_INTERNAL.contains(entry[0])) yield entry - } - } -} diff --git a/src/kv-storage/kv-storage-composite.ts b/src/kv-storage/kv-storage-composite.ts index 1595bfc..3d2d325 100644 --- a/src/kv-storage/kv-storage-composite.ts +++ b/src/kv-storage/kv-storage-composite.ts @@ -1,37 +1,33 @@ import { StorageUnavailableError } from '../utils/errors.js' import type { OffCallback } from '../utils/event-emitter.js' -import { KeyCounter } from '../utils/key-counter.js' +import { Json } from '../utils/json.js' import { KeyRange } from '../utils/key-range.js' -import { RetryTimer, type RetryTimerOptions } from '../utils/retry-timer.js' import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js' -import { SerialQueue } from '../utils/serial-queue.js' -import { BaseKVStorage, type KVChange, type KVChangeEvent, type ScanOptions } from './base-kv-storage.js' -import { KVOutbox, type OutboxEntry } from './kv-outbox.js' -import { KVReplicaSet, SYNC_INTERNAL } from './kv-replica-set.js' +import { Ulid } from '../utils/ulid.js' +import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' +import { KVStorageInMemory } from './kv-storage-in-memory.js' export interface KVStorageCompositeOptions { - /** The source of truth (e.g. `KVStorageHttp`). */ + /** The source of truth, e.g. `KVStorageHttp`. */ source: BaseKVStorage - /** - * Only this key and its subtree are synced, cached and subscribed to (e.g. the user's public - * key). Keys outside it are rejected. Default `''`: the whole source (only for sources the - * client may read entirely). - */ - prefix?: string - /** - * Local copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be persistent: - * it seeds the others on start and holds the offline outbox and sync metadata. - */ + /** Local copies, fastest first, e.g. `[inMemory, indexedDb]`. The last one seeds the others on start. */ replicas: BaseKVStorage[] - /** Backoff for replaying queued writes and retrying a failed sync. */ - retry?: RetryTimerOptions + /** + * Holds writes made while the source is unreachable (under its `writes` child) and whether a + * sync has completed (`synced`). Use a persistent node outside `prefix`, e.g. + * `indexedDb.child('sync')`, so queued writes survive a reload. Default: in memory. + */ + queue?: BaseKVStorage + /** Only this key and its subtree are synced (e.g. the user's public key). Default: everything. */ + prefix?: string + /** Delay before retrying when the source is unreachable. Default 5 000 ms. */ + retryMs?: number } -/** Implemented by sources that push remote changes (e.g. `KVStorageHttp`). */ -interface SubscribableSource { +/** Sources that push remote changes (e.g. `KVStorageHttp`). */ +interface Subscribable { subscribe(path: string): Promise unsubscribe(path: string): Promise - lastEventId: string | undefined } // Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates @@ -41,239 +37,236 @@ export interface KVStorageComposite { } /** - * Keeps local replicas of (part of) a source in sync, so reads, including `keys()`, `entries()` - * and document queries, are local, and the app keeps working offline. + * Combines a source with local replicas: reads come from the first replica, writes go to every + * replica and then the source, exactly as given (no merging: the last write wins). A write the + * source refuses (e.g. 403) is undone locally and rethrown. * - * - **Start** (lazily, on first use): the persistent replica seeds the faster ones. If it holds a - * previous sync, reads are served locally straight away (possibly stale until caught up). - * - **Staying in sync:** remote changes pushed by the source are applied to every replica, in - * order. On restart the source resumes from the saved `lastEventId`; a full sync (diff of the - * source against the replicas) only runs the first time, when the source can't resume - * (`resync`), or when it can't push changes at all. - * - **Writes** go to the source, then the replicas. When the source is unreachable they are - * applied locally and queued in the outbox (`KVOutbox`), which replays them in order. While - * anything is queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source. - * - * Write through the composite, not the source: changes made directly through the same source - * instance (`origin: 'local'`) are not mirrored. + * - Changes pushed by the source (`subscribe`) are applied to the replicas, and announced as + * `change` events, only where they differ from what the replicas hold. The source relaying this + * client's own writes therefore changes nothing. + * - A write the source can't take (`StorageUnavailableError`) is applied locally and stored in + * the `queue` store; later writes queue behind it, so order is kept. + * - `sync()` replays the queue, then copies what changed on the source into the replicas. It runs + * on start, when the source reconnects (`resync`), and after a retry delay while offline. + * - Reads wait for the first sync (until one succeeds, they go to the source). Whether a sync has + * completed is kept in the `queue` store, so with a persistent one a restart reads offline. */ export class KVStorageComposite extends BaseKVStorage { private readonly source: BaseKVStorage + private readonly replicas: BaseKVStorage[] + private readonly queue: BaseKVStorage + private readonly writes: BaseKVStorage private readonly range: KeyRange - private readonly replicas: KVReplicaSet - private readonly outbox: KVOutbox - /** Remote changes and full syncs are applied strictly in order. */ - private readonly applying = new SerialQueue() - private readonly syncRetry: RetryTimer - private readonly outboxRetry: RetryTimer - /** - * Echoes still expected per key: one per write sent to a subscribable source. A remote change to - * such a key is the echo or older than it (the stream is in server order), so it is consumed - * instead of applied; this avoids briefly showing a stale value. - */ - private readonly echoes = new KeyCounter() - private readonly offs: OffCallback[] = [] + private readonly retryMs: number + private readonly ids = new Ulid() /** Mutable state lives here: child nodes share this object (see `child()`). */ private readonly state: { started?: Promise - /** The replicas hold a complete (possibly stale) copy, so reads can be served locally. */ - hasSnapshot: boolean - snapshotTaken: () => void - snapshot: Promise - /** - * Keys written while a full sync runs. Its diff may have scanned the source before such a - * write and the replica after it, so these keys are left out of the diff. - */ - writtenDuringSync?: Set - } + syncing?: Promise + /** A sync has completed at some point, so the replicas hold a full copy. */ + synced: boolean + /** Writes waiting in the queue. */ + queued: number + /** Remote changes are applied one after another. */ + applying: Promise + retry?: ReturnType + offs: OffCallback[] + } = { synced: false, queued: 0, applying: Promise.resolve(), offs: [] } constructor(options: KVStorageCompositeOptions) { super() + if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica') this.source = options.source + this.replicas = options.replicas + this.queue = options.queue ?? new KVStorageInMemory() + this.writes = this.queue.child('writes') this.range = new KeyRange(options.prefix ?? '') - this.replicas = new KVReplicaSet(options.replicas, this.range) - this.syncRetry = new RetryTimer(options.retry) - this.outboxRetry = new RetryTimer(options.retry) - this.outbox = new KVOutbox(this.replicas.persistent, (entry) => this.send(entry), this.outboxRetry) - let snapshotTaken!: () => void - const snapshot = new Promise((resolve) => (snapshotTaken = resolve)) - this.state = { hasSnapshot: false, snapshot, snapshotTaken } + this.retryMs = options.retryMs ?? 5000 + // A full sync makes the replicas match the source within the range, which would delete queued + // writes stored inside that range. + const queueRoot = KVStorageComposite.rootOf(this.queue) + if (this.replicas.some((replica) => KVStorageComposite.rootOf(replica) === queueRoot) && this.range.contains(this.queue.key)) { + throw new TypeError('The queue must be outside the synced range (e.g. a separate store, or a node outside `prefix`)') + } } - /** Starts syncing (also happens on first use). Resolves once local data is loaded. */ + /** Loads local data and starts syncing (also happens on first use). */ start(): Promise { return (this.state.started ??= this.open()) } - /** Resolves once the replicas hold a complete copy and all received remote changes are applied. */ - async synced(): Promise { - await this.start() - await this.state.snapshot - await this.applying.idle() + /** Replays queued writes, then brings the replicas up to date with the source (starts first if needed). */ + sync(): Promise { + return (this.state.syncing ??= this.runSync().finally(() => (this.state.syncing = undefined))) } - /** Number of writes waiting to be sent to the source. */ + /** Number of writes waiting for the source. */ get queued(): number { - return this.outbox.size + return this.state.queued } - /** Stops following the source and cancels retries. */ async close(): Promise { - this.syncRetry.stop() - this.outboxRetry.stop() - for (const off of this.offs.splice(0)) off() - if (this.subscribable) await this.subscribable.unsubscribe(this.range.prefix) + clearTimeout(this.state.retry) + for (const off of this.state.offs.splice(0)) off() + if (KVStorageComposite.isSubscribable(this.source)) await this.source.unsubscribe(this.range.prefix) } async readMany(keys: string[]): Promise { this.checkRange(keys) - await this.start() - return (this.state.hasSnapshot ? this.replicas.fastest : this.source).readMany(keys) + return (await this.reader()).readMany(keys) } async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { - await this.start() - if (this.state.hasSnapshot) return yield* this.replicas.scan(options) const scoped = this.range.clamp(options) - if (scoped) yield* this.source.scan(scoped) + if (scoped) yield* (await this.reader()).scan(scoped) } - protected async applyWrite(entries: [string, unknown][]): Promise { - await this.write({ type: 'write', entries }) - } - - protected async applyRemove(keys: string[]): Promise { - await this.write({ type: 'remove', keys }) - } - - protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { - this.checkRange(entries.map(([key]) => key)) + protected async write(changes: KVChange[]): Promise { + const keys = changes.map(({ key }) => key) + this.checkRange(keys) await this.start() - const written = await this.source.writeIfAbsent(entries) - const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })) - for (const { key } of changes) this.state.writtenDuringSync?.add(key) - await this.replicas.apply(changes) - return written + // Applied locally first, so the source relaying this write back finds nothing new. + const previous = await this.replicas[0]!.readMany(keys) + await this.apply(changes) + // While syncing or with writes queued, queue this one too so the source sees writes in order. + if (this.state.queued === 0 && !this.state.syncing) { + try { + return await this.source.writeMany(changes) + } catch (error) { + if (!(error instanceof StorageUnavailableError)) { + await this.apply(keys.map((key, i) => ({ key, value: previous[i] }))) + throw error + } + } + } + await this.writes.set(this.ids.generate(), changes) + this.state.queued++ + this.retryLater() } - // ------------------------------------------------------------------ start + /** The first replica once it holds a full copy; until then, a sync is attempted first and the source is the fallback. */ + private async reader(): Promise { + if (!this.state.synced) await this.sync() + return this.state.synced ? this.replicas[0]! : this.source + } private async open(): Promise { - const meta = await this.replicas.readMeta() - await this.outbox.load() - await this.replicas.seed() - if (meta) this.snapshotTaken() - - this.offs.push( + this.state.synced = (await this.queue.get('synced')) === true + for await (const _ of this.writes.keys()) this.state.queued++ + await this.seed() + this.state.offs.push( this.source.on('change', (event) => { - if (event.origin === 'remote') this.enqueue(() => this.applyRemote(event)) - if (this.outbox.size > 0) void this.outbox.replay() // the source is reachable again + if (event.origin !== 'remote') return + const changes = event.changes.filter(({ key }) => this.range.contains(key)) + this.state.applying = this.state.applying.then(() => this.applyRemote(changes)).catch((error) => console.error(error)) }), - this.source.on('resync', () => this.enqueue(() => this.fullSync())), + this.source.on('resync', () => void this.sync()), ) - - const source = this.subscribable - if (source) { - source.lastEventId = meta?.lastEventId - // Without a resume point, catch up on the stream first so no change after the snapshot is missed. - const subscribed = source.subscribe(this.range.prefix) - if (!meta?.lastEventId) void subscribed.then(() => this.enqueue(() => this.fullSync())).catch((error) => console.error(error)) - else void subscribed.catch((error) => console.error(error)) + if (KVStorageComposite.isSubscribable(this.source)) { + void this.source + .subscribe(this.range.prefix) + .then(() => this.sync()) + .catch((error) => console.error(error)) } else { - this.enqueue(() => this.fullSync()) + void this.sync() } } - /** Queues a sync task; failures are logged (an unreachable source is retried elsewhere). */ - private enqueue(task: () => Promise): void { - this.applying.run(task).catch((error) => { + private async runSync(): Promise { + await this.start() + try { + await this.replay() + let batch: KVChange[] = [] + for await (const change of this.diff()) if (batch.push(change) >= 500) await this.apply(batch.splice(0), true) + await this.apply(batch, true) + await this.replay() // writes made during the diff: send them, and re-apply them over it + await this.queue.set('synced', true) + this.state.synced = true + } catch (error) { if (!(error instanceof StorageUnavailableError)) console.error(error) - }) + this.retryLater() + } } - private snapshotTaken(): void { - this.state.hasSnapshot = true - this.state.snapshotTaken() + /** Sends queued writes to the source in order, and re-applies each locally (a remote change may have overwritten it meanwhile). */ + private async replay(): Promise { + for (;;) { + let next: [string, unknown] | undefined + for await (const entry of this.writes.entries({ limit: 1 })) next = entry + if (!next) return + const changes = next[1] as KVChange[] + await this.source.writeMany(changes) + await this.apply(changes) + await this.writes.delete(next[0]) + this.state.queued-- + } } - private get subscribable(): SubscribableSource | undefined { - const source = this.source as unknown as Partial - return typeof source.subscribe === 'function' ? (source as SubscribableSource) : undefined + /** What the source has that differs from the last replica (both scanned in key order). */ + private async *diff(): AsyncGenerator { + const options = { gte: this.range.gte, lt: this.range.lt, values: true } + const remote = this.source.scan(options)[Symbol.asyncIterator]() + const local = this.replicas.at(-1)!.scan(options)[Symbol.asyncIterator]() + let r = await remote.next() + let l = await local.next() + while (!r.done || !l.done) { + if (l.done || (!r.done && r.value[0] < l.value[0])) { + yield { key: r.value![0], value: r.value![1] } + r = await remote.next() + } else if (r.done || l.value[0] < r.value[0]) { + yield { key: l.value[0] } + l = await local.next() + } else { + if (!Json.equal(r.value[1], l.value[1])) yield { key: r.value[0], value: r.value[1] } + ;[r, l] = await Promise.all([remote.next(), local.next()]) + } + } + } + + /** Applies pushed changes that differ from what the replicas hold. */ + private async applyRemote(changes: KVChange[]): Promise { + const current = await this.replicas[0]!.readMany(changes.map(({ key }) => key)) + await this.apply( + changes.filter(({ value }, i) => !Json.equal(value, current[i])), + true, + ) + } + + /** Writes changes to every replica; `remote` ones are also announced as `change` events. */ + private async apply(changes: KVChange[], remote = false): Promise { + if (changes.length === 0) return + await Promise.all(this.replicas.map((replica) => replica.writeMany(changes))) + if (remote) this.emit('change', { changes, origin: 'remote' }) + } + + /** Copies the last (persistent) replica into the faster ones. */ + private async seed(): Promise { + if (this.replicas.length < 2) return + let batch: KVChange[] = [] + const flush = () => Promise.all(this.replicas.slice(0, -1).map((replica) => replica.writeMany(batch.splice(0)))) + for await (const [key, value] of this.replicas.at(-1)!.scan({ gte: this.range.gte, lt: this.range.lt, values: true })) { + if (batch.push({ key, value }) >= 500) await flush() + } + await flush() + } + + private retryLater(): void { + clearTimeout(this.state.retry) + this.state.retry = setTimeout(() => void this.sync(), this.retryMs) + ;(this.state.retry as { unref?: () => void }).unref?.() } private checkRange(keys: string[]): void { for (const key of keys) if (!this.range.contains(key)) throw new RangeError(`"${key}" is outside this composite's prefix "${this.range.prefix}"`) } - // ------------------------------------------------------------------ source → replicas - - private async applyRemote(event: KVChangeEvent): Promise { - const relevant = event.changes.filter( - ({ key }) => this.range.contains(key) && !SYNC_INTERNAL.contains(key) && !this.echoes.take(key) && !this.outbox.pending.has(key), - ) - const changes = await this.replicas.unapplied(relevant) - await this.replicas.apply(changes) - if (event.id) await this.replicas.writeMeta({ lastEventId: event.id }) - if (changes.length) this.emit('change', { changes, origin: 'remote', id: event.id }) + /** The storage instance a node belongs to (child nodes are frozen views over it). */ + private static rootOf(node: BaseKVStorage): BaseKVStorage { + while (Object.isFrozen(node)) node = Object.getPrototypeOf(node) as BaseKVStorage + return node } - /** Makes the replicas match the source (within the range). */ - private async fullSync(): Promise { - // The diff re-reads the source, so outstanding echoes no longer matter (and may never arrive). - this.echoes.clear() - const writtenDuringSync = (this.state.writtenDuringSync = new Set()) - const skip = ({ key }: KVChange) => this.outbox.pending.has(key) || writtenDuringSync.has(key) - try { - const lastEventId = this.subscribable?.lastEventId - let batch: KVChange[] = [] - const flush = async () => { - const changes = batch.filter((change) => !skip(change)) - await this.replicas.apply(changes) - if (changes.length) this.emit('change', { changes, origin: 'remote' }) - batch = [] - } - for await (const change of this.replicas.diff(this.source)) if (batch.push(change) >= 500) await flush() - await flush() - await this.replicas.writeMeta({ lastEventId }) - this.snapshotTaken() - this.syncRetry.succeeded() - } catch (error) { - if (!(error instanceof StorageUnavailableError)) throw error - this.syncRetry.schedule(() => this.enqueue(() => this.fullSync())) - } finally { - this.state.writtenDuringSync = undefined - } - } - - // ------------------------------------------------------------------ writes → source - - private async write(entry: OutboxEntry): Promise { - const keys = KVOutbox.keysOf(entry) - this.checkRange(keys) - await this.start() - for (const key of keys) this.state.writtenDuringSync?.add(key) - if (this.outbox.size === 0) { - try { - await this.send(entry) - return this.replicas.apply(KVOutbox.changesOf(entry)) - } catch (error) { - if (!(error instanceof StorageUnavailableError)) throw error - } - } - // Offline, or earlier writes are still queued: apply locally and queue for the source. - await this.outbox.push(entry) - await this.replicas.apply(KVOutbox.changesOf(entry)) - } - - /** Sends a write to the source, expecting its echo when the source pushes changes. */ - private async send(entry: OutboxEntry): Promise { - const keys = this.subscribable ? KVOutbox.keysOf(entry) : [] - this.echoes.add(keys) - try { - await (entry.type === 'write' ? this.source.writeMany(entry.entries) : this.source.removeMany(entry.keys)) - } catch (error) { - for (const key of keys) this.echoes.take(key) - throw error - } + private static isSubscribable(source: BaseKVStorage): source is BaseKVStorage & Subscribable { + return typeof (source as Partial).subscribe === 'function' } } diff --git a/src/kv-storage/kv-storage-http-server.ts b/src/kv-storage/kv-storage-http-server.ts index 32ee001..7db6759 100644 --- a/src/kv-storage/kv-storage-http-server.ts +++ b/src/kv-storage/kv-storage-http-server.ts @@ -13,50 +13,33 @@ export interface KVStorageHttpServerOptions { auth: 'none' | ({ verifiers: SignatureVerifier[] } & RequestAuthOptions) /** Path prefix to strip, e.g. `/kv`. Default `''`. */ basePath?: string - /** Recent change batches kept for clients resuming with `Last-Event-ID`. Default 10 000. */ - historySize?: number /** Interval of SSE keep-alive comments. Default 15 000 ms. */ keepAliveMs?: number } -interface HistoryEntry { - seq: number - changes: KVChange[] -} - -interface Subscriber { - /** Keys (with their subtrees) this subscriber receives; `''` = everything. */ - ranges: KeyRange[] - send: (frame: string) => void -} - -type RequestBody = { keys: string[]; entries: [string, unknown][]; paths: string[] } & ScanOptions +type RequestBody = { keys: string[]; changes: KVChange[]; paths: string[] } & ScanOptions /** * Serves any KV storage over HTTP for `KVStorageHttp`. Framework-agnostic: `handle(Request)` - * returns a `Response`. Every route is `POST` with an Extended JSON body: + * returns a `Response`. It executes exactly the operations clients send; it doesn't know about + * versions or conflicts. Every route is `POST` with an Extended JSON body: * - * - `/readMany`, `/writeMany`, `/removeMany`, `/writeIfAbsent`, `/scan` (streams NDJSON). - * - `/events` with `{ paths }`: Server-Sent Events of changes to those keys and their subtrees. - * Event ids are `-` (epoch = this server instance). A client sending - * `Last-Event-ID` gets the matching changes it missed if they are still in the history, - * otherwise a `resync` event; either way a `ready` event ends the catch-up. + * - `/readMany { keys }`, `/write { changes }`, `/scan { gte, lt, limit, reverse, values }` (NDJSON). + * - `/events { paths }`: Server-Sent Events relaying each change to those keys (and their + * subtrees) as it happens, including the subscriber's own writes. A `ready` event follows the + * connection. Nothing is replayed after a reconnect; clients catch up by reading. * * With authentication, each request is checked first (401), then every key, scan range and * subscription path must be the caller's public key or under it (403). * - * Only changes made through the wrapped storage instance are seen; several server processes over - * one database each see only their own writes. + * Only changes made through the wrapped storage instance are relayed; several server processes + * over one database each relay only their own writes. */ export class KVStorageHttpServer { private readonly auth: RequestAuth | undefined private readonly basePath: string - private readonly historySize: number private readonly keepAliveMs: number - private readonly epoch = Math.random().toString(36).slice(2, 10) - private seq = 0 - private readonly history: HistoryEntry[] = [] - private readonly subscribers = new Set() + private readonly subscribers = new Set<{ ranges: KeyRange[]; send: (frame: string) => void }>() constructor( private readonly storage: BaseKVStorage, @@ -64,9 +47,13 @@ export class KVStorageHttpServer { ) { this.auth = options.auth === 'none' ? undefined : new RequestAuth(options.auth.verifiers, options.auth) this.basePath = options.basePath ?? '' - this.historySize = options.historySize ?? 10_000 this.keepAliveMs = options.keepAliveMs ?? 15_000 - storage.on('change', ({ changes }) => this.publish(changes)) + storage.on('change', ({ changes }) => { + for (const subscriber of this.subscribers) { + const matching = changes.filter(({ key }) => subscriber.ranges.some((range) => range.contains(key))) + if (matching.length) subscriber.send(`event: change\ndata: ${ExtendedJson.stringify(matching)}\n\n`) + } + }) } async handle(request: Request): Promise { @@ -81,32 +68,22 @@ export class KVStorageHttpServer { this.authorize(owner, body.keys) const values = await this.storage.readMany(body.keys) const missing = values.flatMap((v, i) => (v === undefined ? [i] : [])) - return KVStorageHttpServer.json({ values: values.map((v) => v ?? null), missing }) + return new Response(ExtendedJson.stringify({ values: values.map((v) => v ?? null), missing }), { headers: { 'content-type': 'application/json' } }) } - case '/writeMany': + case '/write': this.authorize( owner, - body.entries.map(([key]) => key), + body.changes.map(({ key }) => key), ) - await this.storage.writeMany(body.entries) + await this.storage.writeMany(body.changes) return new Response(null, { status: 204 }) - case '/removeMany': - this.authorize(owner, body.keys) - await this.storage.removeMany(body.keys) - return new Response(null, { status: 204 }) - case '/writeIfAbsent': - this.authorize( - owner, - body.entries.map(([key]) => key), - ) - return KVStorageHttpServer.json(await this.storage.writeIfAbsent(body.entries)) case '/scan': - this.authorizeRange(owner, body) + if (owner !== undefined && !new KeyRange(owner).covers(body.gte, body.lt)) throw new AuthError(403, 'Scan outside your keys') return this.scan(body) case '/events': if (!Array.isArray(body.paths) || body.paths.length === 0) return KVStorageHttpServer.error(400, 'Subscribe to at least one path') this.authorize(owner, body.paths) - return this.events(body.paths, request.headers.get('last-event-id')) + return this.events(body.paths) default: return KVStorageHttpServer.error(404, 'Not found') } @@ -116,8 +93,6 @@ export class KVStorageHttpServer { } } - // ------------------------------------------------------------------ authorization - /** Every key must be the owner's key or under it (no-op without authentication). */ private authorize(owner: string | undefined, keys: string[]): void { if (owner === undefined) return @@ -127,33 +102,15 @@ export class KVStorageHttpServer { } } - /** A scan must stay inside the owner's key and its subtree. */ - private authorizeRange(owner: string | undefined, options: ScanOptions): void { - if (owner !== undefined && !new KeyRange(owner).covers(options.gte, options.lt)) throw new AuthError(403, 'Scan outside your keys') - } - - // ------------------------------------------------------------------ change stream - - private publish(changes: KVChange[]): void { - const entry = { seq: ++this.seq, changes } - this.history.push(entry) - if (this.history.length > this.historySize) this.history.splice(0, this.history.length - this.historySize) - for (const subscriber of this.subscribers) { - const frame = this.changeFrame(entry, subscriber.ranges) - if (frame) subscriber.send(frame) - } - } - - private events(paths: string[], lastEventId: string | null): Response { + private events(paths: string[]): Response { const encoder = new TextEncoder() - let subscriber: Subscriber | undefined + let subscriber: { ranges: KeyRange[]; send: (frame: string) => void } | undefined let keepAlive: ReturnType | undefined const stream = new ReadableStream({ start: (controller) => { - const ranges = paths.map((path) => new KeyRange(path)) - subscriber = { ranges, send: (frame) => controller.enqueue(encoder.encode(frame)) } - subscriber.send(this.catchUpFrames(ranges, lastEventId)) + subscriber = { ranges: paths.map((path) => new KeyRange(path)), send: (frame) => controller.enqueue(encoder.encode(frame)) } this.subscribers.add(subscriber) + subscriber.send('event: ready\ndata: null\n\n') keepAlive = setInterval(() => subscriber!.send(': keep-alive\n\n'), this.keepAliveMs) ;(keepAlive as { unref?: () => void }).unref?.() }, @@ -165,38 +122,6 @@ export class KVStorageHttpServer { return new Response(stream, { headers: { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' } }) } - /** - * What a (re)connecting client gets before live changes: the matching changes it missed, or - * `resync` when they can't be replayed; then always `ready` (caught up as of that id). - */ - private catchUpFrames(ranges: KeyRange[], lastEventId: string | null): string { - const ready = this.frame('ready', this.seq, null) - if (lastEventId === null) return ready - const [epoch, seqText] = lastEventId.split('-') - const seq = Number(seqText) - const oldest = this.history[0]?.seq ?? this.seq + 1 - // Resumable only if it's our epoch and nothing after `seq` has been dropped from the history. - if (epoch !== this.epoch || !Number.isInteger(seq) || seq > this.seq || seq < oldest - 1) return this.frame('resync', this.seq, null) + ready - return ( - this.history - .filter((entry) => entry.seq > seq) - .map((entry) => this.changeFrame(entry, ranges)) - .join('') + ready - ) - } - - /** A `change` frame with only the changes inside `ranges`, or `''` if none match. */ - private changeFrame(entry: HistoryEntry, ranges: KeyRange[]): string { - const changes = entry.changes.filter(({ key }) => ranges.some((range) => range.contains(key))) - return changes.length ? this.frame('change', entry.seq, changes) : '' - } - - private frame(event: string, seq: number, data: unknown): string { - return `id: ${this.epoch}-${seq}\nevent: ${event}\ndata: ${ExtendedJson.stringify(data)}\n\n` - } - - // ------------------------------------------------------------------ scan - private scan(options: ScanOptions): Response { const iterator = this.storage.scan(options)[Symbol.asyncIterator]() const encoder = new TextEncoder() @@ -221,10 +146,6 @@ export class KVStorageHttpServer { return new Response(stream, { headers: { 'content-type': 'application/x-ndjson' } }) } - private static json(body: unknown): Response { - return new Response(ExtendedJson.stringify(body), { headers: { 'content-type': 'application/json' } }) - } - private static error(status: number, message: string): Response { return Response.json({ error: message }, { status }) } diff --git a/src/kv-storage/kv-storage-http.ts b/src/kv-storage/kv-storage-http.ts index fd8bb64..da16133 100644 --- a/src/kv-storage/kv-storage-http.ts +++ b/src/kv-storage/kv-storage-http.ts @@ -3,7 +3,6 @@ import { ExtendedJson } from '../utils/extended-json.js' import { RequestAuth, type RequestSigner } from '../utils/request-auth.js' import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js' import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js' -import { SerialQueue } from '../utils/serial-queue.js' import { SSESession } from '../utils/sse-session/sse-session.js' import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' @@ -27,10 +26,10 @@ export interface KVStorageHttp { /** * KV storage client for a `KVStorageHttpServer`. Same API as any other KV storage. * - * `subscribe(path)` receives changes under a key (pushed over SSE) as `change` events with - * `origin: 'remote'`; `resync` is emitted when the server can't resume from `lastEventId`. With a - * `signer`, every request is signed, and an authenticated server only allows keys under the - * signer's public key. Values travel as Extended JSON (`bigint` and `Uint8Array` survive). + * `subscribe(path)` relays every change the server pushes for a key and its subtree (including + * this client's own writes) as `change` events with `origin: 'remote'`. When a dropped stream reconnects, `resync` is emitted: changes may have been + * missed, so re-read what you need. With a `signer`, every request is signed. Values travel as + * Extended JSON (`bigint` and `Uint8Array` survive). */ export class KVStorageHttp extends BaseKVStorage { private readonly url: string @@ -38,10 +37,8 @@ export class KVStorageHttp extends BaseKVStorage { private readonly headers: Record private readonly fetcher: (url: string, init: RequestInit) => Promise private readonly onSubscriptionError: (error: Error) => void - /** Subscription changes are applied one at a time. */ - private readonly updates = new SerialQueue() /** Mutable state lives here: child nodes share this object (see `child()`). */ - private readonly state: { paths: Set; session?: SSESession; lastEventId?: string } = { paths: new Set() } + private readonly state: { paths: Set; session?: SSESession; updating: Promise } = { paths: new Set(), updating: Promise.resolve() } constructor(options: KVStorageHttpOptions) { super() @@ -52,23 +49,9 @@ export class KVStorageHttp extends BaseKVStorage { this.onSubscriptionError = options.onSubscriptionError ?? (() => {}) } - /** Position in the server's change stream; sent as `Last-Event-ID` when (re)connecting. */ - get lastEventId(): string | undefined { - return this.state.lastEventId - } - - set lastEventId(id: string | undefined) { - this.state.lastEventId = id - } - - /** Keys currently subscribed to. */ - get subscriptions(): ReadonlySet { - return this.state.paths - } - async readMany(keys: string[]): Promise { if (keys.length === 0) return [] - const { values, missing } = (await this.post('readMany', { keys })) as { values: unknown[]; missing: number[] } + const { values, missing } = ExtendedJson.parse(await (await this.request('readMany', { keys })).text()) as { values: unknown[]; missing: number[] } for (const i of missing) values[i] = undefined return values } @@ -96,15 +79,15 @@ export class KVStorageHttp extends BaseKVStorage { } /** - * Receives changes to `path` (default: this node's key) and everything under it. Resolves once - * the stream is connected and caught up (missed changes replayed, or `resync` emitted). While - * the server is unreachable this stays pending, so don't await it on a path that must work offline. + * Relays changes to `path` (default: this node's key) and everything under it. Resolves once + * the stream is connected. While the server is unreachable this stays pending, so don't await it + * on a path that must work offline. A 401/403 rejects instead of retrying. */ subscribe(path: string = this.key): Promise { return this.updateSubscriptions(() => this.state.paths.add(path)) } - /** Stops receiving changes to `path` (default: this node's key). */ + /** Stops relaying changes to `path` (default: this node's key). */ unsubscribe(path: string = this.key): Promise { return this.updateSubscriptions(() => this.state.paths.delete(path)) } @@ -114,21 +97,11 @@ export class KVStorageHttp extends BaseKVStorage { return this.updateSubscriptions(() => this.state.paths.clear()) } - protected async applyWrite(entries: [string, unknown][]): Promise { - await this.post('writeMany', { entries }) + protected async write(changes: KVChange[]): Promise { + await this.request('write', { changes }) } - protected async applyRemove(keys: string[]): Promise { - await this.post('removeMany', { keys }) - } - - protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { - return (await this.post('writeIfAbsent', { entries })) as boolean[] - } - - // ------------------------------------------------------------------ subscriptions - - /** Applies a subscription change, then reconnects the stream with the new path list. */ + /** Applies a subscription change (one at a time), then reconnects with the new path list. */ private updateSubscriptions(change: () => unknown): Promise { const run = async () => { const before = JSON.stringify([...this.state.paths]) @@ -138,7 +111,9 @@ export class KVStorageHttp extends BaseKVStorage { this.state.session = undefined if (this.state.paths.size > 0) await this.connect([...this.state.paths]) } - return this.updates.run(run) + const result = this.state.updating.then(run) + this.state.updating = result.catch(() => {}) + return result } private async connect(paths: string[]): Promise { @@ -153,39 +128,23 @@ export class KVStorageHttp extends BaseKVStorage { onError: this.onSubscriptionError, retry: KVStorageHttp.retryUnlessDenied(), // Signed on every (re)connect, so the timestamp is always fresh. - onRequest: async (request) => ({ - ...request, - headers: { - ...(request.headers as Record), - ...(this.state.lastEventId ? { 'Last-Event-ID': this.state.lastEventId } : {}), - ...(await this.authHeaders(url, body)), - }, - }), + onRequest: async (request) => ({ ...request, headers: { ...(request.headers as Record), ...(await this.authHeaders(url, body)) } }), }) + let readies = 0 session.on('message', (message) => { - if (message.id) this.state.lastEventId = message.id - if (message.event === 'change') { - this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote', id: message.id }) - } else if (message.event === 'resync') { - this.emit('resync', { id: message.id }) - } + if (message.event === 'change') this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote' }) + // Every `ready` after the first is a reconnect: changes may have been missed meanwhile. + if (message.event === 'ready' && readies++ > 0) this.emit('resync', undefined) }) this.state.session = session - const caughtUp = session.waitFor('message', (message) => message.event === 'ready') + const ready = session.waitFor('message', (message) => message.event === 'ready') try { await session.connect() } catch (error) { this.state.session = undefined throw (error as Error).cause instanceof AuthError ? (error as Error).cause : error } - await caughtUp - } - - // ------------------------------------------------------------------ requests - - private async post(path: string, body: unknown): Promise { - const response = await this.request(path, body) - return response.status === 204 ? undefined : ExtendedJson.parse(await response.text()) + await ready } private async request(path: string, payload: unknown): Promise { @@ -207,6 +166,10 @@ export class KVStorageHttp extends BaseKVStorage { return response } + private async authHeaders(url: string, body: string): Promise> { + return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {} + } + /** Retries connecting with backoff, except when the server denies access (401/403): that won't fix itself. */ private static retryUnlessDenied(): { run(fn: () => Promise): Promise } { const backoff = new ExponentialBackoff({ baseDelay: 1000, maxDelay: 10_000, maxAttempts: 0, growthRate: 1.3, jitter: 0.3 }) @@ -218,8 +181,4 @@ export class KVStorageHttp extends BaseKVStorage { }), } } - - private async authHeaders(url: string, body: string): Promise> { - return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {} - } } diff --git a/src/kv-storage/kv-storage-in-memory.ts b/src/kv-storage/kv-storage-in-memory.ts index 3ac5f6e..dbadf5c 100644 --- a/src/kv-storage/kv-storage-in-memory.ts +++ b/src/kv-storage/kv-storage-in-memory.ts @@ -1,7 +1,7 @@ import { Json } from '../utils/json.js' import { SortedMap } from '../utils/sorted-map.js' import type { AnySpec, Spec } from '../utils/schema.js' -import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js' +import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' /** In-memory KV storage. Values are deep-copied on write and read. */ export class KVStorageInMemory extends BaseKVStorage { @@ -11,18 +11,6 @@ export class KVStorageInMemory extends BaseKVStorage Json.clone(this.data.get(k))) } - protected async applyWrite(entries: [string, unknown][]): Promise { - for (const [k, v] of entries) this.data.set(k, Json.clone(v)) - } - - protected async applyRemove(keys: string[]): Promise { - for (const k of keys) this.data.delete(k) - } - - protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { - return entries.map(([k, v]) => !this.data.has(k) && (this.data.set(k, Json.clone(v)), true)) - } - async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { let remaining = options.limit ?? Infinity if (remaining <= 0) return @@ -31,4 +19,11 @@ export class KVStorageInMemory extends BaseKVStorage { + for (const { key, value } of changes) { + if (value === undefined) this.data.delete(key) + else this.data.set(key, Json.clone(value)) + } + } } diff --git a/src/kv-storage/kv-storage-indexeddb.ts b/src/kv-storage/kv-storage-indexeddb.ts index 4c0e7c7..9ca99b4 100644 --- a/src/kv-storage/kv-storage-indexeddb.ts +++ b/src/kv-storage/kv-storage-indexeddb.ts @@ -1,5 +1,5 @@ import type { AnySpec, Spec } from '../utils/schema.js' -import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js' +import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' export interface KVStorageIndexedDBOptions { /** Database name. */ @@ -56,32 +56,15 @@ export class KVStorageIndexedDB extends BaseKVStorage< ;(await this.db).close() } - protected async applyWrite(entries: [string, unknown][]): Promise { + protected async write(changes: KVChange[]): Promise { const store = await this.store('readwrite') - for (const [key, value] of entries) store.put(value, key) + for (const { key, value } of changes) { + if (value === undefined) store.delete(key) + else store.put(value, key) + } await KVStorageIndexedDB.committed(store.transaction) } - protected async applyRemove(keys: string[]): Promise { - const store = await this.store('readwrite') - for (const key of keys) store.delete(key) - await KVStorageIndexedDB.committed(store.transaction) - } - - protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { - // One readwrite transaction: checks and writes can't interleave with other writers. - const store = await this.store('readwrite') - const written = await Promise.all( - entries.map(async ([key, value]) => { - if ((await KVStorageIndexedDB.done(store.getKey(key))) !== undefined) return false - store.add(value, key) - return true - }), - ) - await KVStorageIndexedDB.committed(store.transaction) - return written - } - /** Up to `count` entries after `last` (exclusive), in scan order, within one transaction. */ private async readBatch(options: ScanOptions, last: string | undefined, count: number): Promise<[string, unknown][]> { const reverse = options.reverse ?? false diff --git a/src/utils/errors.ts b/src/utils/errors.ts index 5c15807..be137ec 100644 --- a/src/utils/errors.ts +++ b/src/utils/errors.ts @@ -1,15 +1,3 @@ -/** An optimistic-concurrency check failed (stale `expectedVersion` or a lost commit race). */ -export class ConcurrencyError extends Error { - constructor( - readonly stream: string, - readonly expectedVersion: number | 'any', - readonly actualVersion: number, - ) { - super(`Concurrency conflict on "${stream}": expected version ${expectedVersion}, actual ${actualVersion}`) - this.name = 'ConcurrencyError' - } -} - /** A key argument was passed but is `undefined`/`null`, empty, or not a string/number. */ export class InvalidKeyError extends TypeError { constructor(message: string) { diff --git a/src/utils/extended-json.ts b/src/utils/extended-json.ts index f7ada03..8318bf9 100644 --- a/src/utils/extended-json.ts +++ b/src/utils/extended-json.ts @@ -18,6 +18,19 @@ export class ExtendedJson { return JSON.stringify(value, ExtendedJson.replacer) } + /** Like `stringify`, with object keys sorted, so equal values always give the same text (for hashing). */ + static canonical(value: unknown): string { + return JSON.stringify(value, (key, v) => { + const replaced = ExtendedJson.replacer(key, v) + if (typeof replaced !== 'object' || replaced === null || Array.isArray(replaced)) return replaced + return Object.fromEntries( + Object.keys(replaced) + .sort() + .map((k) => [k, (replaced as Record)[k]]), + ) + }) + } + static parse(text: string): unknown { return JSON.parse(text, ExtendedJson.reviver) } diff --git a/src/utils/key-counter.ts b/src/utils/key-counter.ts deleted file mode 100644 index 9e5e016..0000000 --- a/src/utils/key-counter.ts +++ /dev/null @@ -1,36 +0,0 @@ -/** A count per key (a multiset). */ -export class KeyCounter { - private readonly counts = new Map() - private total = 0 - - /** Sum of all counts. */ - get size(): number { - return this.total - } - - has(key: string): boolean { - return this.counts.has(key) - } - - add(keys: Iterable): void { - for (const key of keys) { - this.counts.set(key, (this.counts.get(key) ?? 0) + 1) - this.total++ - } - } - - /** Decrements `key`; returns whether it had a count. */ - take(key: string): boolean { - const count = this.counts.get(key) - if (!count) return false - if (count > 1) this.counts.set(key, count - 1) - else this.counts.delete(key) - this.total-- - return true - } - - clear(): void { - this.counts.clear() - this.total = 0 - } -} diff --git a/src/utils/retry-timer.ts b/src/utils/retry-timer.ts deleted file mode 100644 index ac04b2b..0000000 --- a/src/utils/retry-timer.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { ExponentialBackoff } from './sse-session/exponential-backoff.js' - -export interface RetryTimerOptions { - baseDelay?: number - maxDelay?: number -} - -/** Schedules one pending retry at a time, with exponential backoff that resets on success. */ -export class RetryTimer { - private readonly options: { baseDelay: number; maxDelay: number } - private timer: ReturnType | undefined - private attempt = 0 - private stopped = false - - constructor(options: RetryTimerOptions = {}) { - this.options = { baseDelay: 1000, maxDelay: 30_000, ...options } - } - - /** Runs `task` after the next backoff delay, replacing any retry already scheduled. */ - schedule(task: () => void): void { - if (this.stopped) return - clearTimeout(this.timer) - const delay = ExponentialBackoff.calculateDelay({ ...this.options, growthRate: 2, jitter: 0.2, maxAttempts: 0 }, this.attempt++) - this.timer = setTimeout(task, delay) - ;(this.timer as { unref?: () => void }).unref?.() - } - - succeeded(): void { - this.attempt = 0 - } - - stop(): void { - this.stopped = true - clearTimeout(this.timer) - } -} diff --git a/src/utils/serial-queue.ts b/src/utils/serial-queue.ts deleted file mode 100644 index 943a634..0000000 --- a/src/utils/serial-queue.ts +++ /dev/null @@ -1,17 +0,0 @@ -/** Runs tasks one at a time, in the order given; independently per key when a key is passed. */ -export class SerialQueue { - private readonly tails = new Map>() - - run(task: () => Promise, key: string = ''): Promise { - const result = (this.tails.get(key) ?? Promise.resolve()).then(task) - const tail = result.catch(() => {}) - this.tails.set(key, tail) - void tail.then(() => this.tails.get(key) === tail && this.tails.delete(key)) - return result - } - - /** Resolves once every task queued so far (for `key`) has settled. */ - async idle(key: string = ''): Promise { - while (this.tails.has(key)) await this.tails.get(key) - } -} diff --git a/test/document-storage/document-storage.test.ts b/test/document-storage/document-storage.test.ts index d4add11..54a1b9d 100644 --- a/test/document-storage/document-storage.test.ts +++ b/test/document-storage/document-storage.test.ts @@ -42,43 +42,48 @@ describe('EventSourcedDocumentStorage', () => { return { kv, events, projections, docs: new EventSourcedDocumentStorage(events, projections) } } - it('records resolved changes as history and replays any version', async () => { + it('records resolved changes as history and replays any event', async () => { const users = setup().docs.child('users') await users.set('1', { n: 5 }) await users.update('1', { $inc: { n: 1 } }) await users.delete('1') await users.set('1', { n: 10 }) - expect((await collect(users.history('1'))).map((e) => e.data)).toEqual([ + const history = await collect(users.history('1')) + expect(history.map((e) => e.data)).toEqual([ { op: 'set', doc: { n: 5 } }, { op: 'patch', set: { n: 6 }, unset: [] }, { op: 'delete' }, { op: 'set', doc: { n: 10 } }, ]) - expect(await users.getAt('1', 2)).toEqual({ n: 6 }) - expect(await users.child('1').getAt(3)).toBeUndefined() + expect(await users.getAt('1', history[1]!.id)).toEqual({ n: 6 }) + expect(await users.child('1').getAt(history[2]!.id)).toBeUndefined() expect(await users.get('1')).toEqual({ n: 10 }) }) - it('heals a projection left behind by a crash between append and projection write', async () => { + it('catchUp folds in events written elsewhere', async () => { const { docs, events } = setup() const users = docs.child('users') await users.set('1', { n: 1 }) - await events.child('users').append('1', [{ op: 'patch', set: { n: 2 }, unset: [] }]) // projection never written + await events.child('users').append('1', [{ op: 'patch', set: { m: 2 }, unset: [] }]) // e.g. synced from another device expect(await users.get('1')).toEqual({ n: 1 }) - expect(await users.update('1', { $inc: { n: 1 } })).toEqual({ n: 3 }) - await events.child('users').append('2', [{ op: 'set', doc: { n: 7 } }]) // first write crashed - expect(await users.catchUp('2')).toEqual({ n: 7 }) - expect(await users.get('2')).toEqual({ n: 7 }) + expect(await users.catchUp('1')).toEqual({ n: 1, m: 2 }) + expect(await users.get('1')).toEqual({ n: 1, m: 2 }) + await events.child('users').append('2', [{ op: 'set', doc: { n: 7 } }, { op: 'delete' }]) + expect(await users.catchUp('2')).toBeUndefined() + expect(await users.get('2')).toBeUndefined() }) - it('two instances incrementing the same document converge', async () => { + it('two instances editing different fields both keep their edits; the same field ends with the later value', async () => { const kv = new KVStorageInMemory() - const make = () => new EventSourcedDocumentStorage(new EventStorageKV(kv), new DocumentStorageKV(kv.child('projections'))).child('users') + let t = 1000 + const make = () => + new EventSourcedDocumentStorage(new EventStorageKV(kv, { now: () => t++ }), new DocumentStorageKV(kv.child('projections'))).child('users') const a = make() const b = make() - await a.set('1', { n: 0 }) - await Promise.all([...Array(5)].flatMap(() => [a.update('1', { $inc: { n: 1 } }), b.update('1', { $inc: { n: 1 } })])) - expect(await a.get('1')).toEqual({ n: 10 }) + await a.set('1', { name: 'x', age: 1 }) + await Promise.all([a.update('1', { $set: { name: 'Ada' } }), b.update('1', { $set: { age: 2 } })]) + await b.update('1', { $set: { name: 'Bob' } }) + expect(await a.catchUp('1')).toEqual({ name: 'Bob', age: 2 }) }) it('a write is a projection read, an append and a projection write', async () => { @@ -86,6 +91,6 @@ describe('EventSourcedDocumentStorage', () => { await docs.child('users').set('1', { n: 1 }) kv.reset() await docs.child('users').update('1', { $inc: { n: 1 } }) - expect(kv.calls).toEqual({ readMany: 2, writeIfAbsent: 1, writeMany: 1 }) + expect(kv.calls).toEqual({ readMany: 2, writeMany: 2 }) }) }) diff --git a/test/event-storage/event-storage-kv.test.ts b/test/event-storage/event-storage-kv.test.ts index 4b0af71..fb38830 100644 --- a/test/event-storage/event-storage-kv.test.ts +++ b/test/event-storage/event-storage-kv.test.ts @@ -1,45 +1,70 @@ import { describe, expect, it } from 'vitest' import fc from 'fast-check' import { EventStorageKV } from '../../src/event-storage/event-storage-kv.js' +import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' -import { ConcurrencyError } from '../../src/utils/errors.js' import { collect, KVStorageCounting } from '../helpers.js' -function setup(kv = new KVStorageCounting()) { - let t = 1000 +function setup(kv: BaseKVStorage = new KVStorageCounting(), start = 1000) { + let t = start return { kv, events: new EventStorageKV(kv, { now: () => t++ }) } } describe('EventStorageKV', () => { it('appends and reads in self and child forms', async () => { const orders = setup().events.child('orders') - expect(await orders.append('o1', [{ type: 'created' }], { expectedVersion: 0 })).toEqual({ version: 1 }) - expect(await orders.child('o1').append([{ type: 'paid' }, { type: 'shipped' }])).toEqual({ version: 3 }) + const [created] = await orders.append('o1', [{ type: 'created' }]) + const ids = await orders.child('o1').append([{ type: 'paid' }, { type: 'shipped' }]) expect(await collect(orders.read('o1'))).toEqual([ - { version: 1, at: 1000, data: { type: 'created' } }, - { version: 2, at: 1001, data: { type: 'paid' } }, - { version: 3, at: 1001, data: { type: 'shipped' } }, + { id: created, at: 1000, data: { type: 'created' } }, + { id: ids[0], at: 1001, data: { type: 'paid' } }, + { id: ids[1], at: 1002, data: { type: 'shipped' } }, ]) - expect(await collect(orders.child('o1').read({ from: 3 }))).toHaveLength(1) - expect(await orders.version('o1')).toBe(3) - expect(await orders.version('none')).toBe(0) + expect((await collect(orders.child('o1').read({ after: ids[0] }))).map((e) => e.data)).toEqual([{ type: 'shipped' }]) + expect((await collect(orders.read('o1', { reverse: true, limit: 2 }))).map((e) => e.id)).toEqual([ids[1], ids[0]]) + expect((await collect(orders.read('o1', { reverse: true, after: ids[0] }))).map((e) => e.id)).toEqual([created]) }) - it('stores one key per commit', async () => { + it('ids are the time and a hash of the time and content', async () => { const { kv, events } = setup() - await events.child('s').append(['a']) - await events.child('s').append(['b', 'c']) - expect((await collect(kv.scan())).map(([k]) => k)).toEqual(['s/~e/000000000001', 's/~e/000000000002']) + const [id] = await events.child('s').append([{ b: 1, a: 2 }]) + expect(id).toMatch(/^000000000001000_[0-9a-f]{64}$/) + expect(id).toBe(await EventStorageKV.id(1000, { a: 2, b: 1 })) + expect((await collect(kv.scan())).map(([k]) => k)).toEqual([`s/~e/${id}`]) }) - it('enforces expectedVersion and the arity rules', async () => { + it('one append is one write; one instance never reuses a time', async () => { + const kv = new KVStorageCounting() + const events = new EventStorageKV(kv, { now: () => 5 }) + const ids = await events.child('s').append(['a', 'a']) + expect(kv.calls).toEqual({ writeMany: 1 }) + expect(ids.map((id) => id.slice(0, 15))).toEqual(['000000000000005', '000000000000006']) + expect((await collect(events.child('s').read())).map((e) => e.data)).toEqual(['a', 'a']) + }) + + it('keeps concurrent appends from different writers, ordered by time', async () => { + const kv = new KVStorageInMemory() + const a = setup(kv, 1000).events.child('s') + const b = setup(kv, 1000).events.child('s') + await Promise.all([a.append(['a1', 'a2']), b.append(['b1', 'b2'])]) + const read = await collect(a.read()) + expect(read).toHaveLength(4) + expect(read.map((e) => e.at)).toEqual([1000, 1000, 1001, 1001]) + expect(read.map((e) => e.id)).toEqual([...read.map((e) => e.id)].sort()) + expect(await collect(b.read())).toEqual(read) + }) + + it('writing the same event again does not duplicate it', async () => { + const kv = new KVStorageInMemory() + await setup(kv).events.child('s').append(['x']) + await setup(kv).events.child('s').append(['x']) + expect(await collect(new EventStorageKV(kv).child('s').read())).toHaveLength(1) + }) + + it('checks the arity rules', async () => { const { events } = setup() - const s = events.child('s') - await s.append(['a']) - await expect(s.append(['b'], { expectedVersion: 0 })).rejects.toBeInstanceOf(ConcurrencyError) await expect(events.append(['x'])).rejects.toThrow(/root/) await expect(events.child('o').append(undefined as never, ['x'] as never)).rejects.toThrow(TypeError) - await expect(events.child('o').version(undefined as never)).rejects.toThrow(TypeError) }) it('lists child streams', async () => { @@ -51,46 +76,26 @@ describe('EventStorageKV', () => { expect(await collect(orders.child('o1').keys())).toEqual(['notes']) }) - it('a warm append is one writeIfAbsent; a cold one adds one reverse scan', async () => { - const { kv, events } = setup() - await events.child('s').append(['a']) - kv.reset() - await events.child('s').append(['b']) - expect(kv.calls).toEqual({ writeIfAbsent: 1 }) - const cold = setup(kv).events - kv.reset() - await cold.child('s').append(['c']) - expect(kv.calls).toEqual({ scan: 1, writeIfAbsent: 1 }) - }) - - it('a writer with a stale cache collides; with "any" it retries', async () => { - const kv = new KVStorageInMemory() - const a = setup(new KVStorageCounting(kv)).events.child('s') - const b = setup(new KVStorageCounting(kv)).events.child('s') - await a.append(['a1']) - await b.append(['b1']) - await expect(a.append(['a2'], { expectedVersion: 1 })).rejects.toMatchObject({ actualVersion: 2 }) - expect(await a.append(['a3'])).toEqual({ version: 3 }) - const results = await Promise.allSettled([a.append(['x'], { expectedVersion: 3 }), b.append(['y'], { expectedVersion: 3 })]) - expect(results.filter((r) => r.status === 'fulfilled')).toHaveLength(1) - }) - it('reads any window correctly (property)', async () => { await fc.assert( fc.asyncProperty( fc.array(fc.integer({ min: 1, max: 4 }), { maxLength: 20 }), fc.record({ - from: fc.option(fc.integer({ min: 0, max: 60 }), { nil: undefined }), - to: fc.option(fc.integer({ min: 0, max: 60 }), { nil: undefined }), + after: fc.option(fc.nat(60), { nil: undefined }), limit: fc.option(fc.nat(10), { nil: undefined }), + reverse: fc.boolean(), }), - async (sizes, options) => { + async (sizes, { after, limit, reverse }) => { const s = setup().events.child('s') - let n = 0 - for (const size of sizes) await s.append(Array.from({ length: size }, () => ++n)) - let expected = Array.from({ length: n }, (_, i) => i + 1).filter((v) => v >= (options.from ?? 1) && v <= (options.to ?? Infinity)) - if (options.limit !== undefined) expected = expected.slice(0, options.limit) - expect((await collect(s.read(options))).map((e) => e.data)).toEqual(expected) + const ids: string[] = [] + for (const size of sizes) ids.push(...(await s.append(Array.from({ length: size }, () => ids.length)))) + const all = await collect(s.read()) + expect(all.map((e) => e.id)).toEqual(ids) + const ordered = reverse ? [...ids].reverse() : ids + const afterId = after === undefined ? undefined : ids[after] + let expected = afterId === undefined ? ordered : ordered.slice(ordered.indexOf(afterId) + 1) + if (limit !== undefined) expected = expected.slice(0, limit) + expect((await collect(s.read({ after: afterId, limit, reverse }))).map((e) => e.id)).toEqual(expected) }, ), { numRuns: 200 }, diff --git a/test/helpers.ts b/test/helpers.ts index 65550dc..9ca1b66 100644 --- a/test/helpers.ts +++ b/test/helpers.ts @@ -1,4 +1,4 @@ -import { BaseKVStorage, type ScanOptions } from '../src/kv-storage/base-kv-storage.js' +import { BaseKVStorage, type KVChange, type ScanOptions } from '../src/kv-storage/base-kv-storage.js' import { KVStorageInMemory } from '../src/kv-storage/kv-storage-in-memory.js' export async function collect(iterable: AsyncIterable): Promise { @@ -9,8 +9,8 @@ export async function collect(iterable: AsyncIterable): Promise { /** * Wraps another KV storage and counts primitive calls. Also shows the wrapper pattern - * (e.g. an encryption layer): extend BaseKVStorage and delegate the five primitives. - * Counts are reported under the public method names (writeMany, removeMany, writeIfAbsent). + * (e.g. an encryption layer): extend BaseKVStorage and delegate the three primitives. + * Writes are counted as `writeMany`. */ export class KVStorageCounting extends BaseKVStorage { calls: Record = {} @@ -26,14 +26,8 @@ export class KVStorageCounting extends BaseKVStorage { readMany(keys: string[]) { return this.count('readMany', () => this.inner.readMany(keys)) } - protected applyWrite(entries: [string, unknown][]) { - return this.count('writeMany', () => this.inner.writeMany(entries)) - } - protected applyRemove(keys: string[]) { - return this.count('removeMany', () => this.inner.removeMany(keys)) - } - protected applyWriteIfAbsent(entries: [string, unknown][]) { - return this.count('writeIfAbsent', () => this.inner.writeIfAbsent(entries)) + protected write(changes: KVChange[]) { + return this.count('writeMany', () => this.inner.writeMany(changes)) } scan(options?: ScanOptions) { return this.count('scan', () => this.inner.scan(options)) diff --git a/test/kv-storage/kv-outbox.test.ts b/test/kv-storage/kv-outbox.test.ts deleted file mode 100644 index e4dc134..0000000 --- a/test/kv-storage/kv-outbox.test.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { describe, expect, it, vi } from 'vitest' -import { KVOutbox, type OutboxEntry } from '../../src/kv-storage/kv-outbox.js' -import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' -import { RetryTimer } from '../../src/utils/retry-timer.js' -import { StorageUnavailableError } from '../../src/utils/errors.js' - -describe('KVOutbox', () => { - it('replays entries in order once the source is reachable, and survives a restart', async () => { - const store = new KVStorageInMemory() - let online = false - const sent: OutboxEntry[] = [] - const send = async (entry: OutboxEntry) => { - if (!online) throw new StorageUnavailableError('offline') - sent.push(entry) - } - const firstRetry = new RetryTimer({ baseDelay: 5, maxDelay: 5 }) - const first = new KVOutbox(store, send, firstRetry) - await first.push({ type: 'write', entries: [['a', 1]] }) - await first.push({ type: 'remove', keys: ['a'] }) - expect(first.size).toBe(2) - expect(first.pending.has('a')).toBe(true) - - firstRetry.stop() // the first session ends... - await first.replay() // ...once its in-flight replay attempt has failed - // A new session over the same store picks up the queue. - const second = new KVOutbox(store, send, new RetryTimer({ baseDelay: 5, maxDelay: 5 })) - online = true - await second.load() - await vi.waitFor(() => expect(second.size).toBe(0)) - expect(sent).toEqual([ - { type: 'write', entries: [['a', 1]] }, - { type: 'remove', keys: ['a'] }, - ]) - expect(await store.readMany(['~q/o/000000000001', '~q/o/000000000002'])).toEqual([undefined, undefined]) - }) -}) diff --git a/test/kv-storage/kv-storage-composite.test.ts b/test/kv-storage/kv-storage-composite.test.ts index aef18cb..93bfedf 100644 --- a/test/kv-storage/kv-storage-composite.test.ts +++ b/test/kv-storage/kv-storage-composite.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' -import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js' +import type { BaseKVStorage, KVChangeEvent } from '../../src/kv-storage/base-kv-storage.js' import { DocumentStorageKV } from '../../src/document-storage/document-storage-kv.js' import { collect, KVStorageCounting } from '../helpers.js' import { runKVStorageConformance } from './kv-storage.conformance.js' @@ -23,21 +23,21 @@ afterEach(async () => { await network?.close() }) -function composite(persistent: BaseKVStorage = new KVStorageInMemory()) { - const c = new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), persistent], retry: { baseDelay: 5, maxDelay: 20 } }) +function composite(persistent: BaseKVStorage = new KVStorageInMemory(), queue?: BaseKVStorage) { + const c = new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), persistent], queue, retryMs: 10 }) composites.push(c) return c } describe('KVStorageComposite sync', () => { - it('loads existing data with one full sync, then serves reads locally', async () => { + it('loads existing data with one sync, then serves reads locally', async () => { network = new TestNetwork() await network.store.writeMany([ - ['users/1', { name: 'Ada' }], - ['users/2', { name: 'Bob' }], + { key: 'users/1', value: { name: 'Ada' } }, + { key: 'users/2', value: { name: 'Bob' } }, ]) const db = composite() - await db.synced() + await db.sync() network.store.reset() expect(await db.child('users').get('1')).toEqual({ name: 'Ada' }) expect(await collect(db.child('users').keys())).toEqual(['1', '2']) @@ -48,7 +48,7 @@ describe('KVStorageComposite sync', () => { network = new TestNetwork() const memory = new KVStorageInMemory() const db = composite(memory) - await db.synced() + await db.sync() const seen = db.waitFor('change', (e) => e.origin === 'remote') await network.client().child('users').set('9', { name: 'Remote' }) expect((await seen).changes).toEqual([{ key: 'users/9', value: { name: 'Remote' } }]) @@ -56,10 +56,29 @@ describe('KVStorageComposite sync', () => { expect(await memory.readMany(['users/9'])).toEqual([{ name: 'Remote' }]) }) + it('announces only remote changes that differ from local values', async () => { + network = new TestNetwork() + const db = composite() + await db.sync() + const events: KVChangeEvent[] = [] + db.on('change', (e) => events.push(e)) + await db.set('mine', 1) // relayed back by the server: same value, so no remote event + const other = db.waitFor('change', (e) => e.origin === 'remote') + await network.client().writeMany([ + { key: 'mine', value: 1 }, + { key: 'theirs', value: 2 }, + ]) + await other + expect(events).toEqual([ + { changes: [{ key: 'mine', value: 1 }], origin: 'local' }, + { changes: [{ key: 'theirs', value: 2 }], origin: 'remote' }, + ]) + }) + it('queues writes while offline and replays them in order when back online', async () => { network = new TestNetwork() const db = composite() - await db.synced() + await db.sync() network.online = false await db.child('users').set('1', { v: 1 }) await db.child('users').set('1', { v: 2 }) @@ -67,68 +86,97 @@ describe('KVStorageComposite sync', () => { await db.child('users').set('2', { v: 3 }) expect(db.queued).toBe(4) expect(await db.child('users').get('2')).toEqual({ v: 3 }) // served locally - await expect(db.writeIfAbsent([['x', 1]])).rejects.toThrow(/Cannot reach/) network.online = true await vi.waitFor(() => expect(db.queued).toBe(0)) expect(await network.store.readMany(['users/1', 'users/2'])).toEqual([undefined, { v: 3 }]) }) - it('keeps queued local values when the server pushes an older value for the same key', async () => { + it('the last write wins when the server pushes a change while a local write is queued', async () => { network = new TestNetwork() const db = composite() - await db.synced() + await db.sync() network.online = false - await db.child('k').set({ from: 'local' }) - await network.store.writeMany([['k', { from: 'remote' }]]) // pushed over the still-open stream - await new Promise((r) => setTimeout(r, 20)) - expect(await db.child('k').get()).toEqual({ from: 'local' }) + await db.set('k', { from: 'local' }) + await network.store.writeMany([{ key: 'k', value: { from: 'remote' } }]) // pushed over the still-open stream + await vi.waitFor(async () => expect(await db.get('k')).toEqual({ from: 'remote' })) network.online = true - await vi.waitFor(() => expect(db.queued).toBe(0)) + await vi.waitFor(() => expect(db.queued).toBe(0)) // the queued write is sent last, so it wins expect(await network.store.readMany(['k'])).toEqual([{ from: 'local' }]) + expect(await db.get('k')).toEqual({ from: 'local' }) }) - it('restarts from the persistent replica and resumes without a full sync', async () => { + it('restarts from the persistent replica, then catches up', async () => { network = new TestNetwork() const persistent = new KVStorageInMemory() - const first = composite(persistent) - await first.child('a').set(1) - await first.synced() + const queue = new KVStorageInMemory() + const first = composite(persistent, queue) + await first.set('a', 1) + await first.set('gone', 1) + await first.sync() await first.close() - await network.client().writeMany([['b', 2]]) // missed while "closed" + await network.client().writeMany([{ key: 'b', value: 2 }, { key: 'gone' }]) // missed while closed - network.store.reset() - const second = composite(persistent) - expect(await second.child('a').get()).toBe(1) // available before contacting the server - await vi.waitFor(async () => expect(await second.child('b').get()).toBe(2)) - expect(network.store.calls.scan).toBeUndefined() // resumed via Last-Event-ID, no full scan - }) - - it('falls back to a full sync when the server cannot resume', async () => { - network = new TestNetwork({ historySize: 1 }) - const persistent = new KVStorageInMemory() - const first = composite(persistent) - await first.child('a').set(1) - await first.synced() - await first.close() - await network.client().writeMany([['b', 2]]) - await network.client().writeMany([['c', 3]]) - await network.client().removeMany(['a']) - const second = composite(persistent) - await vi.waitFor(async () => expect(await collect(second.keys())).toEqual(['b', 'c'])) - }) - - it('replays an outbox left by a previous session', async () => { - network = new TestNetwork() - const persistent = new KVStorageInMemory() - const first = composite(persistent) - await first.synced() network.online = false - await first.child('offline').set('queued') + const second = composite(persistent, queue) + expect(await second.get('a')).toBe(1) // available without the server + network.online = true + await second.sync() + expect(await collect(second.scan({ values: true }))).toEqual([ + ['a', 1], + ['b', 2], + ]) + }) + + it('replays writes queued by a previous session from a persistent queue', async () => { + network = new TestNetwork() + const persistent = new KVStorageInMemory() + const queue = persistent.child('~sync') + const first = new KVStorageComposite({ source: network.client(), replicas: [persistent], queue, prefix: 'data' }) + composites.push(first) + await first.sync() + network.online = false + await first.child('data').set('offline', 'queued') await first.close() network.online = true - const second = composite(persistent) - await second.start() - await vi.waitFor(async () => expect(await network.store.readMany(['offline'])).toEqual(['queued'])) + const second = new KVStorageComposite({ source: network.client(), replicas: [persistent], queue, prefix: 'data' }) + composites.push(second) + expect(second.queued).toBe(0) + await second.sync() + expect(second.queued).toBe(0) + expect(await network.store.readMany(['data/offline'])).toEqual(['queued']) + }) + + it('catches up on changes missed while the stream was down', async () => { + network = new TestNetwork() + const db = composite() + await db.sync() + const resynced = vi.fn() + db.on('change', resynced) + network.online = false + network.dropStreams() + await network.store.writeMany([{ key: 'missed', value: 1 }]) + network.online = true + await vi.waitFor(async () => expect(await db.get('missed')).toBe(1), { timeout: 5000 }) + expect(resynced).toHaveBeenCalledWith({ changes: [{ key: 'missed', value: 1 }], origin: 'remote' }) + }) + + it('syncs only its prefix and rejects keys outside it', async () => { + network = new TestNetwork() + await network.store.writeMany([ + { key: 'alice/a', value: 1 }, + { key: 'bob/b', value: 2 }, + ]) + const db = new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory()], prefix: 'alice' }) + composites.push(db) + await db.sync() + expect(await collect(db.scan({ values: true }))).toEqual([['alice/a', 1]]) + await expect(db.child('bob').get('b')).rejects.toBeInstanceOf(RangeError) + }) + + it('refuses a queue inside the synced range of a replica', () => { + const replica = new KVStorageInMemory() + expect(() => new KVStorageComposite({ source: new KVStorageInMemory(), replicas: [replica], queue: replica.child('q') })).toThrow(/outside/) + expect(() => new KVStorageComposite({ source: new KVStorageInMemory(), replicas: [replica], queue: replica.child('q'), prefix: 'data' })).not.toThrow() }) it('serves document queries from the local replica, offline too', async () => { @@ -137,12 +185,12 @@ describe('KVStorageComposite sync', () => { const users = new DocumentStorageKV(db).child('users') await users.set('1', { age: 30 }) await users.set('2', { age: 40 }) - await db.synced() + await db.sync() network.online = false expect((await collect(users.find({ age: { $gt: 35 } }))).map((r) => r.key)).toEqual(['2']) }) - it('does not undo writes made while a full sync is scanning', async () => { + it('does not undo writes made while a sync is scanning', async () => { // A source whose scan reads ahead (like the HTTP stream) and pauses after its first entry. let release!: () => void let reached!: () => void @@ -150,8 +198,8 @@ describe('KVStorageComposite sync', () => { const paused = new Promise((r) => (reached = r)) const inner = new KVStorageInMemory() await inner.writeMany([ - ['a', 1], - ['c', 3], + { key: 'a', value: 1 }, + { key: 'c', value: 3 }, ]) const source = new KVStorageCounting(inner) const scan = source.scan.bind(source) @@ -163,13 +211,15 @@ describe('KVStorageComposite sync', () => { } } const persistent = new KVStorageInMemory() - await persistent.writeMany([['a0', 'stale']]) // a replica copy from before, without sync metadata + await persistent.writeMany([{ key: 'a0', value: 'stale' }]) const db = new KVStorageComposite({ source, replicas: [persistent] }) + composites.push(db) void db.start() await paused // the diff has passed 'b' in the source but not in the replica - await db.writeMany([['b', 2]]) + await db.set('b', 2) release() - await db.synced() + await vi.waitFor(() => expect(db.queued).toBe(0)) + await db.sync() expect(await collect(db.scan({ values: true }))).toEqual([ ['a', 1], ['b', 2], diff --git a/test/kv-storage/kv-storage-http-auth.test.ts b/test/kv-storage/kv-storage-http-auth.test.ts index 79575a8..280a140 100644 --- a/test/kv-storage/kv-storage-http-auth.test.ts +++ b/test/kv-storage/kv-storage-http-auth.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import type { KVChangeEvent } from '../../src/kv-storage/base-kv-storage.js' import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js' +import { KVStorageHttpServer } from '../../src/kv-storage/kv-storage-http-server.js' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' import { AuthError } from '../../src/utils/errors.js' import { Ed25519Signer, Ed25519Verifier } from '../../src/utils/request-auth.js' @@ -16,9 +17,9 @@ beforeEach(async () => { alice = await Ed25519Signer.generate() bob = await Ed25519Signer.generate() await network.store.writeMany([ - [`${alice.publicKey}/profile`, { name: 'Alice' }], - [`${bob.publicKey}/profile`, { name: 'Bob' }], - [`${bob.publicKey}/posts/1`, { title: 'secret' }], + { key: `${alice.publicKey}/profile`, value: { name: 'Alice' } }, + { key: `${bob.publicKey}/profile`, value: { name: 'Bob' } }, + { key: `${bob.publicKey}/posts/1`, value: { title: 'secret' } }, ]) }) afterEach(() => network.close()) @@ -67,22 +68,9 @@ describe('authenticated KVStorageHttpServer', () => { await expect(network.client(alice).subscribe('')).rejects.toMatchObject({ status: 403 }) }) - it('replays only matching changes when resuming', async () => { - const watcher = network.client(alice) - const mine = watcher.child(alice.publicKey) - await mine.subscribe() - await mine.unsubscribe() - await network.client(bob).child(bob.publicKey).set('x', 1) - await network.client(alice).child(alice.publicKey).set('y', 2) - const replayed: KVChangeEvent[] = [] - watcher.on('change', (e) => replayed.push(e)) - await mine.subscribe() - expect(replayed.flatMap((e) => e.changes)).toEqual([{ key: `${alice.publicKey}/y`, value: 2 }]) - }) - it('a composite scoped to the user syncs only their data', async () => { const db = new KVStorageComposite({ source: network.client(alice), prefix: alice.publicKey, replicas: [new KVStorageInMemory()] }) - await db.synced() + await db.sync() expect(await collect(db.scan({ values: true }))).toEqual([[`${alice.publicKey}/profile`, { name: 'Alice' }]]) expect(await collect(db.keys())).toEqual([alice.publicKey]) await expect(db.child(bob.publicKey).get('profile')).rejects.toBeInstanceOf(RangeError) @@ -92,4 +80,15 @@ describe('authenticated KVStorageHttpServer', () => { expect((await pushed).changes).toEqual([{ key: `${alice.publicKey}/fromElsewhere`, value: 1 }]) await db.close() }) + + it('a composite undoes a local write the server refuses', async () => { + const replica = new KVStorageInMemory() + const db = new KVStorageComposite({ source: network.client(alice), prefix: alice.publicKey, replicas: [replica] }) + await db.sync() + await db.child(alice.publicKey).set('x', 'before') + network.server = new KVStorageHttpServer(network.store, { auth: { verifiers: [{ algorithm: 'ed25519', verify: async () => false }] } }) // now rejects every signature + await expect(db.child(alice.publicKey).set('x', 'after')).rejects.toMatchObject({ status: 401 }) + expect(await db.child(alice.publicKey).get('x')).toBe('before') + await db.close() + }) }) diff --git a/test/kv-storage/kv-storage-http-watch.test.ts b/test/kv-storage/kv-storage-http-watch.test.ts index 31dc777..c802edc 100644 --- a/test/kv-storage/kv-storage-http-watch.test.ts +++ b/test/kv-storage/kv-storage-http-watch.test.ts @@ -7,49 +7,47 @@ let network: TestNetwork afterEach(() => network?.close()) describe('KVStorageHttp.subscribe (SSE)', () => { - it('receives changes made by other clients as remote change events', async () => { + it('relays every change under the path as remote change events, its own writes included', async () => { network = new TestNetwork() const watcher = network.client() - await watcher.subscribe('') - const received = watcher.waitFor('change', (e) => e.origin === 'remote') + await watcher.child('users').subscribe() + const received: KVChangeEvent[] = [] + watcher.on('change', (e) => e.origin === 'remote' && received.push(e)) + const other = watcher.waitFor('change', (e) => e.origin === 'remote') await network.client().child('users').set('1', { name: 'Ada', n: 1n }) - const event = await received - expect(event.changes).toEqual([{ key: 'users/1', value: { name: 'Ada', n: 1n } }]) - expect(event.id).toMatch(/^\w+-1$/) - expect(watcher.lastEventId).toBe(event.id) + await other + const own = watcher.waitFor('change', (e) => e.origin === 'remote') + await network.client().set('elsewhere', 1) // not subscribed + await watcher.child('users').delete('1') + await own + expect(received.map((e) => e.changes)).toEqual([[{ key: 'users/1', value: { name: 'Ada', n: 1n } }], [{ key: 'users/1' }]]) }) - it('resumes from lastEventId and replays missed changes', async () => { + it('emits resync when a dropped stream reconnects', async () => { network = new TestNetwork() - const writer = network.client() const watcher = network.client() await watcher.subscribe('') - const first = watcher.waitFor('change', (e) => e.origin === 'remote') - await writer.writeMany([['a', 1]]) - await first - await watcher.unsubscribe('') - await writer.writeMany([['b', 2]]) - await writer.writeMany([['c', 3]]) - const replayed: KVChangeEvent[] = [] - watcher.on('change', (e) => replayed.push(e)) - await watcher.subscribe('') // resolves once caught up, so the missed changes have been emitted - expect(replayed.map((e) => e.changes)).toEqual([[{ key: 'b', value: 2 }], [{ key: 'c', value: 3 }]]) + const resync = watcher.waitFor('resync', () => true) + network.dropStreams() + await resync + const after = watcher.waitFor('change', () => true) + await network.client().set('a', 1) + expect((await after).changes).toEqual([{ key: 'a', value: 1 }]) }) - it('asks for a resync when the missed changes are no longer in the history', async () => { - network = new TestNetwork({ historySize: 1 }) - const writer = network.client() + it('stops relaying after unsubscribe', async () => { + network = new TestNetwork() const watcher = network.client() - await watcher.subscribe('') - const seen = watcher.waitFor('change', () => true) - await writer.writeMany([['a', 1]]) - await seen - await watcher.unsubscribe('') - await writer.writeMany([['b', 2]]) - await writer.writeMany([['c', 3]]) - const resync = watcher.waitFor('resync', () => true) - await watcher.subscribe('') - expect(await resync).toEqual({ id: expect.stringMatching(/-3$/) }) + await watcher.subscribe('a') + await watcher.subscribe('b') + await watcher.unsubscribe('a') + const received: KVChangeEvent[] = [] + watcher.on('change', (e) => received.push(e)) + const arrived = watcher.waitFor('change', () => true) + await network.client().set('a', 1) + await network.client().set('b', 2) + await arrived + expect(received.flatMap((e) => e.changes)).toEqual([{ key: 'b', value: 2 }]) }) it('reports an unreachable server as StorageUnavailableError', async () => { diff --git a/test/kv-storage/kv-storage-http.test.ts b/test/kv-storage/kv-storage-http.test.ts index 50d6410..59fff25 100644 --- a/test/kv-storage/kv-storage-http.test.ts +++ b/test/kv-storage/kv-storage-http.test.ts @@ -30,7 +30,7 @@ describe('over a real socket', () => { it('stops a long scan early', async () => { const db = fresh() - await db.writeMany(Array.from({ length: 5000 }, (_, i) => [`k${String(i).padStart(5, '0')}`, i] as [string, number])) + await db.writeMany(Array.from({ length: 5000 }, (_, i) => ({ key: `k${String(i).padStart(5, '0')}`, value: i }))) const seen: string[] = [] for await (const [key] of db.scan()) if (seen.push(key) === 3) break expect(seen).toEqual(['k00000', 'k00001', 'k00002']) diff --git a/test/kv-storage/kv-storage-in-memory.test.ts b/test/kv-storage/kv-storage-in-memory.test.ts index 20f5e2a..7754c80 100644 --- a/test/kv-storage/kv-storage-in-memory.test.ts +++ b/test/kv-storage/kv-storage-in-memory.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'vitest' import fc from 'fast-check' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' -import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js' +import type { BaseKVStorage, KVChange } from '../../src/kv-storage/base-kv-storage.js' import { collect, KVStorageCounting } from '../helpers.js' import { runKVStorageConformance } from './kv-storage.conformance.js' @@ -16,14 +16,14 @@ describe('costs', () => { await users.get('1') await users.child('1').get() await users.delete('1') - expect(db.calls).toEqual({ writeMany: 1, readMany: 2, removeMany: 1 }) + expect(db.calls).toEqual({ writeMany: 2, readMany: 2 }) }) it('keys() seeks once per child with descendants, not once per key', async () => { const db = new KVStorageCounting() - const writes: [string, number][] = [] - for (let c = 0; c < 50; c++) for (let d = 0; d < 20; d++) writes.push([`n/c${c}/d${d}`, d]) - for (let c = 0; c < 50; c++) writes.push([`n/leaf${c}`, c]) + const writes: KVChange[] = [] + for (let c = 0; c < 50; c++) for (let d = 0; d < 20; d++) writes.push({ key: `n/c${c}/d${d}`, value: d }) + for (let c = 0; c < 50; c++) writes.push({ key: `n/leaf${c}`, value: c }) await db.writeMany(writes) db.reset() expect(await collect(db.child('n').keys())).toHaveLength(100) diff --git a/test/kv-storage/kv-storage.conformance.ts b/test/kv-storage/kv-storage.conformance.ts index f5814bd..caa8fd9 100644 --- a/test/kv-storage/kv-storage.conformance.ts +++ b/test/kv-storage/kv-storage.conformance.ts @@ -15,14 +15,14 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage describe('primitives', () => { beforeEach(async () => { await db.writeMany([ - ['b', 2], - ['a/2', { n: 'a2' }], - ['c', [3]], - ['a', 1], - ['a/1', 'a1'], - ['a0', null], - ['10', 10], - ['9', 9], + { key: 'b', value: 2 }, + { key: 'a/2', value: { n: 'a2' } }, + { key: 'c', value: [3] }, + { key: 'a', value: 1 }, + { key: 'a/1', value: 'a1' }, + { key: 'a0', value: null }, + { key: '10', value: 10 }, + { key: '9', value: 9 }, ]) }) @@ -48,16 +48,9 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage expect(await db.readMany(['b', 'c'])).toEqual([undefined, [3]]) }) - it('writeIfAbsent writes only absent keys, and only one concurrent claim wins', async () => { - expect( - await db.writeIfAbsent([ - ['a', 'x'], - ['new', 'y'], - ]), - ).toEqual([false, true]) - expect(await db.readMany(['a', 'new'])).toEqual([1, 'y']) - const results = await Promise.all([1, 2, 3].map((n) => db.writeIfAbsent([['claim', n]]))) - expect(results.flat().filter(Boolean)).toHaveLength(1) + it('writes and removes in one batch', async () => { + await db.writeMany([{ key: 'b' }, { key: 'new', value: 'y' }, { key: 'a', value: 'x' }]) + expect(await db.readMany(['a', 'b', 'new'])).toEqual(['x', undefined, 'y']) }) it('keeps scanning correctly when keys change mid-scan', async () => { @@ -74,22 +67,18 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage it('emits change events for writes and removals', async () => { const events: unknown[] = [] const off = db.on('change', (event) => events.push(event)) - await db.writeMany([['x', 1]]) + await db.writeMany([{ key: 'x', value: 1 }]) await db.removeMany(['x']) - await db.writeIfAbsent([ - ['a', 'no'], - ['y', 2], - ]) + await db.writeMany([]) off() expect(events).toEqual([ { changes: [{ key: 'x', value: 1 }], origin: 'local' }, { changes: [{ key: 'x' }], origin: 'local' }, - { changes: [{ key: 'y', value: 2 }], origin: 'local' }, ]) }) it('stores bigint and Uint8Array values', async () => { - await db.writeMany([['bin', { n: 10n ** 20n, bytes: new Uint8Array([0, 255]) }]]) + await db.writeMany([{ key: 'bin', value: { n: 10n ** 20n, bytes: new Uint8Array([0, 255]) } }]) expect(await db.readMany(['bin'])).toEqual([{ n: 10n ** 20n, bytes: new Uint8Array([0, 255]) }]) }) @@ -123,6 +112,14 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage expect(await n.getMany(['a', 'b'])).toEqual([undefined, 2]) }) + it('sets and gets direct children as fields', async () => { + const user = db.child('users').child('1') + await user.setFields({ name: 'Ada', hobby: 'chess', skipped: undefined }) + await user.child('address').set('city', 'Paris') + expect(await user.get('hobby')).toBe('chess') + expect(await user.getFields()).toEqual({ hobby: 'chess', name: 'Ada' }) + }) + it('keeps own values and children independent', async () => { const user = db.child('users').child('1') await user.set('own') @@ -157,7 +154,7 @@ export function runKVStorageConformance(name: string, make: () => BaseKVStorage await n.set('c', 3) await n.child('c').child('x').set('y', 1) await n.set('a-1', 4) - await db.writeMany([['n/~e/000000000001', {}]]) + await db.writeMany([{ key: 'n/~e/000000000001', value: {} }]) expect(await collect(n.keys())).toEqual(['a', 'a-1', 'b', 'c']) expect(await collect(n.keys({ reverse: true }))).toEqual(['c', 'b', 'a-1', 'a']) expect(await collect(n.keys({ after: 'a', limit: 2 }))).toEqual(['a-1', 'b']) diff --git a/test/kv-storage/sync-helpers.ts b/test/kv-storage/sync-helpers.ts index 8b3eece..7da9805 100644 --- a/test/kv-storage/sync-helpers.ts +++ b/test/kv-storage/sync-helpers.ts @@ -7,12 +7,14 @@ import { KVStorageCounting } from '../helpers.js' /** * An in-process server with a network switch. While `online` is false, new requests fail like a * dropped connection (open SSE streams stay up, so pushes can still be simulated). + * `dropStreams()` ends every open SSE stream, as a proxy or server restart would. */ export class TestNetwork { online = true readonly store = new KVStorageCounting(new KVStorageInMemory()) - readonly server: KVStorageHttpServer + server: KVStorageHttpServer private readonly clients: KVStorageHttp[] = [] + private readonly streams = new Set<() => void>() constructor(options: Partial = {}) { this.server = new KVStorageHttpServer(this.store, { auth: 'none', ...options }) @@ -24,13 +26,43 @@ export class TestNetwork { signer, fetch: async (url, init) => { if (!this.online) throw new TypeError('fetch failed') - return this.server.handle(new Request(url, init)) + const response = await this.server.handle(new Request(url, init)) + return url.endsWith('/events') && response.body ? this.droppable(response) : response }, }) this.clients.push(client) return client } + dropStreams(): void { + for (const drop of this.streams) drop() + this.streams.clear() + } + + private droppable(response: Response): Response { + const reader = response.body!.getReader() + let drop!: () => void + const body = new ReadableStream({ + start: (controller) => { + drop = () => { + controller.close() + void reader.cancel() + } + this.streams.add(drop) + }, + pull: async (controller) => { + const { done, value } = await reader.read() + if (done) controller.close() + else controller.enqueue(value) + }, + cancel: async () => { + this.streams.delete(drop) + await reader.cancel() + }, + }) + return new Response(body, { status: response.status, headers: response.headers }) + } + async close(): Promise { await Promise.all(this.clients.map((c) => c.close())) } diff --git a/test/types.test-d.ts b/test/types.test-d.ts index 7f589f6..c397b2d 100644 --- a/test/types.test-d.ts +++ b/test/types.test-d.ts @@ -84,6 +84,6 @@ describe('event and document types', () => { new DocumentStorageInMemory(), ) expectTypeOf(docs.child('users').history('1')).toEqualTypeOf>>() - expectTypeOf(docs.child('users').getAt('1', 3)).toEqualTypeOf>() + expectTypeOf(docs.child('users').getAt('1', 'id')).toEqualTypeOf>() }) }) diff --git a/test/utils/building-blocks.test.ts b/test/utils/building-blocks.test.ts index 0c5eb21..7fbcec2 100644 --- a/test/utils/building-blocks.test.ts +++ b/test/utils/building-blocks.test.ts @@ -1,9 +1,6 @@ -import { describe, expect, it, vi } from 'vitest' +import { describe, expect, it } from 'vitest' import fc from 'fast-check' import { KeyRange } from '../../src/utils/key-range.js' -import { KeyCounter } from '../../src/utils/key-counter.js' -import { SerialQueue } from '../../src/utils/serial-queue.js' -import { RetryTimer } from '../../src/utils/retry-timer.js' import { Keys } from '../../src/utils/keys.js' describe('KeyRange', () => { @@ -36,54 +33,3 @@ describe('KeyRange', () => { expect(new KeyRange('').covers(undefined, undefined)).toBe(true) }) }) - -describe('KeyCounter', () => { - it('counts per key and in total', () => { - const counter = new KeyCounter() - counter.add(['a', 'a', 'b']) - expect(counter.size).toBe(3) - expect(counter.take('a')).toBe(true) - expect(counter.has('a')).toBe(true) - expect(counter.take('a')).toBe(true) - expect(counter.has('a')).toBe(false) - expect(counter.take('a')).toBe(false) - expect(counter.size).toBe(1) - counter.clear() - expect(counter.size).toBe(0) - }) -}) - -describe('SerialQueue', () => { - it('runs tasks in order per key, keys independently, and survives failures', async () => { - const queue = new SerialQueue() - const log: string[] = [] - const task = (name: string, ms: number) => () => new Promise((r) => setTimeout(() => (log.push(name), r()), ms)) - const a1 = queue.run(task('a1', 20), 'a') - const failed = queue.run(() => Promise.reject(new Error('boom')), 'a') - const a2 = queue.run(task('a2', 1), 'a') - const b1 = queue.run(task('b1', 1), 'b') - await expect(failed).rejects.toThrow('boom') - await Promise.all([a1, a2, b1]) - expect(log).toEqual(['b1', 'a1', 'a2']) - await queue.idle('a') - }) -}) - -describe('RetryTimer', () => { - it('replaces a pending retry and stops cleanly', () => { - vi.useFakeTimers() - const timer = new RetryTimer({ baseDelay: 10, maxDelay: 10 }) - const first = vi.fn() - const second = vi.fn() - timer.schedule(first) - timer.schedule(second) - vi.advanceTimersByTime(20) - expect(first).not.toHaveBeenCalled() - expect(second).toHaveBeenCalledOnce() - timer.schedule(first) - timer.stop() - vi.advanceTimersByTime(20) - expect(first).not.toHaveBeenCalled() - vi.useRealTimers() - }) -})