From fe751f246e6f7584795ceed79bf07e915786fcf9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 13:18:15 +0000 Subject: [PATCH] WIP: plain operations, simpler composite (does not build or pass tests yet) In progress, not for review: - KV adapters reduced to readMany/scan/write; writeIfAbsent removed - server executes operations verbatim; SSE relays changes, no history - composite rewritten as a combination of stores with a queue store - client-id echo tagging (to be removed per review) Events, event-sourced documents, tests and docs still to be updated. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE --- src/kv-storage/base-kv-storage.ts | 61 ++-- src/kv-storage/kv-outbox.ts | 91 ------ src/kv-storage/kv-replica-set.ts | 105 ------- src/kv-storage/kv-storage-composite.ts | 352 ++++++++++------------- src/kv-storage/kv-storage-http-server.ts | 132 ++------- src/kv-storage/kv-storage-http.ts | 97 ++----- src/kv-storage/kv-storage-in-memory.ts | 21 +- src/kv-storage/kv-storage-indexeddb.ts | 29 +- src/utils/key-counter.ts | 36 --- src/utils/retry-timer.ts | 36 --- src/utils/serial-queue.ts | 17 -- test/kv-storage/kv-outbox.test.ts | 36 --- 12 files changed, 260 insertions(+), 753 deletions(-) delete mode 100644 src/kv-storage/kv-outbox.ts delete mode 100644 src/kv-storage/kv-replica-set.ts delete mode 100644 src/utils/key-counter.ts delete mode 100644 src/utils/retry-timer.ts delete mode 100644 src/utils/serial-queue.ts delete mode 100644 test/kv-storage/kv-outbox.test.ts diff --git a/src/kv-storage/base-kv-storage.ts b/src/kv-storage/base-kv-storage.ts index 08dc823..9e12446 100644 --- a/src/kv-storage/base-kv-storage.ts +++ b/src/kv-storage/base-kv-storage.ts @@ -30,20 +30,24 @@ 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 + /** Who made the change, when known (e.g. the id of the HTTP client that sent it). */ + source?: 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 +67,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[], source?: string): Promise { + if (changes.length === 0) return + await this.write(changes) + this.emit('change', source === undefined ? { changes, origin: 'local' } : { changes, origin: 'local', source }) } 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 +120,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..c6f0623 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,201 @@ 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 the + * source and then every replica, exactly as given (no merging: the last write wins). * - * - **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 others make arrive from the source (`subscribe`) and are applied to the replicas. + * - 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. + * - Before the first successful sync, reads go to the source. */ 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 } - /** 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. */ + 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 (this.state.synced ? this.replicas[0]! : this.source).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* (this.state.synced ? this.replicas[0]! : this.source).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 { + this.checkRange(changes.map(({ key }) => key)) 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 + // 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 { + await this.source.writeMany(changes) + return this.apply(changes) + } catch (error) { + if (!(error instanceof StorageUnavailableError)) throw error + } + } + await this.writes.set(this.ids.generate(), changes) + this.state.queued++ + await this.apply(changes) + this.retryLater() } - // ------------------------------------------------------------------ start - 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.apply(changes, true)).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 { + 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 them locally. */ + 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()]) + } + } + } + + /** 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 }) - } - - /** 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..12b535d 100644 --- a/src/kv-storage/kv-storage-http-server.ts +++ b/src/kv-storage/kv-storage-http-server.ts @@ -13,50 +13,34 @@ 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[]; client?: string; 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, with the id of the client that wrote it (`/write { client }`), so a + * client can ignore its 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 +48,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, source }) => { + 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({ changes: matching, source })}\n\n`) + } + }) } async handle(request: Request): Promise { @@ -81,32 +69,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, typeof body.client === 'string' ? body.client : undefined) 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 +94,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 +103,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 +123,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 +147,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..668cd21 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 changes under a key made by others, pushed over SSE, as `change` events + * with `origin: 'remote'` (this client's own writes aren't echoed back). 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,10 @@ 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() + /** Sent with writes; the server relays it with each change so this client can skip its own. */ + private readonly clientId = crypto.randomUUID() /** 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 +51,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 +81,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 +99,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, client: this.clientId }) } - 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 +113,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 +130,27 @@ 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 }) + const { changes, source } = ExtendedJson.parse(message.data) as { changes: KVChange[]; source?: string } + // Our own writes are already applied locally; a late echo could only undo a newer write. + if (source !== this.clientId) this.emit('change', { changes, origin: 'remote', ...(source === undefined ? {} : { source }) }) } + // 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 +172,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 +187,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/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/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]) - }) -})