diff --git a/DESIGN.md b/DESIGN.md index b919ff4..fbd9a6b 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -177,6 +177,18 @@ outbox `~q/o/`. - `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) | + ## 5. Event storage `BaseEventStorage` primitives over stream paths: `appendTo`, `readFrom`, `versionOf`, `streamsOf`. diff --git a/src/document-storage/event-sourced-document-storage.ts b/src/document-storage/event-sourced-document-storage.ts index d9f27c8..1a3f957 100644 --- a/src/document-storage/event-sourced-document-storage.ts +++ b/src/document-storage/event-sourced-document-storage.ts @@ -3,6 +3,7 @@ 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' @@ -29,8 +30,8 @@ export interface EventSourcedDocumentStorage { */ export class EventSourcedDocumentStorage extends BaseDocumentStorage { private readonly maxRetries: number - /** Tail of the pending writes per document, so writes in one process don't conflict with each other. */ - private readonly locks = new Map>() + /** 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, @@ -50,12 +51,7 @@ export class EventSourcedDocumentStorage extends BaseD } mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise { - const id = JSON.stringify(path) - const result = (this.locks.get(id) ?? Promise.resolve()).then(() => this.write(path, change)) - const tail = result.catch(() => {}) - this.locks.set(id, tail) - void tail.then(() => this.locks.get(id) === tail && this.locks.delete(id)) - return result + return this.writes.run(() => this.write(path, change), JSON.stringify(path)) } private async write(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise { diff --git a/src/index.ts b/src/index.ts index 966cb34..29b67db 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,8 @@ 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' @@ -17,6 +19,10 @@ 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/kv-outbox.ts b/src/kv-storage/kv-outbox.ts new file mode 100644 index 0000000..ead5a91 --- /dev/null +++ b/src/kv-storage/kv-outbox.ts @@ -0,0 +1,91 @@ +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 new file mode 100644 index 0000000..75a1409 --- /dev/null +++ b/src/kv-storage/kv-replica-set.ts @@ -0,0 +1,105 @@ +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 c698f80..1595bfc 100644 --- a/src/kv-storage/kv-storage-composite.ts +++ b/src/kv-storage/kv-storage-composite.ts @@ -1,10 +1,13 @@ import { StorageUnavailableError } from '../utils/errors.js' import type { OffCallback } from '../utils/event-emitter.js' -import { Json } from '../utils/json.js' -import { Keys } from '../utils/keys.js' +import { KeyCounter } from '../utils/key-counter.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 { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.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' export interface KVStorageCompositeOptions { /** The source of truth (e.g. `KVStorageHttp`). */ @@ -16,115 +19,85 @@ export interface KVStorageCompositeOptions { */ prefix?: string /** - * Local full copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be - * persistent: it also stores the offline outbox and sync metadata, and seeds the others on start. + * 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. */ replicas: BaseKVStorage[] - /** Delays between attempts to replay queued writes or re-sync. */ - retry?: Partial<{ baseDelay: number; maxDelay: number }> + /** Backoff for replaying queued writes and retrying a failed sync. */ + retry?: RetryTimerOptions } -/** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */ -interface WatchableSource { +/** Implemented by sources that push remote changes (e.g. `KVStorageHttp`). */ +interface SubscribableSource { subscribe(path: string): Promise unsubscribe(path: string): Promise lastEventId: string | undefined } -type OutboxEntry = { type: 'write'; entries: [string, unknown][] } | { type: 'remove'; keys: string[] } - -interface SyncMeta { - lastEventId?: string -} - -/** Internal keys in the persistent replica; hidden from reads. */ -const INTERNAL_PREFIX = '~q/' -const INTERNAL_END = '~q0' -const META_KEY = '~q/meta' -const OUTBOX_PREFIX = '~q/o/' -const OUTBOX_END = '~q/o0' -const SYNC_BATCH = 500 - -// Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates them with -// Object.create, so they share this class's prototype). +// Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates +// them with Object.create, so they share this class's prototype). export interface KVStorageComposite { child>(key: K): KVStorageComposite> } /** - * Keeps local replicas of a source in sync, so reads (including `keys()`, `entries()` and document - * queries) are served locally, and the app keeps working offline. + * 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. * * - **Start** (lazily, on first use): the persistent replica seeds the faster ones. If it holds a - * previous sync, reads are served from it straight away (possibly stale until caught up). - * - **Staying in sync:** remote changes pushed by the source (`subscribe(prefix)`, e.g. SSE) are applied to - * every replica. On restart the source resumes from the saved `lastEventId`; a full sync (diffing - * the source against the replicas) only runs the first time, or when the source says it can't - * resume (`resync`). Without `subscribe()`, a full sync runs on start. - * - **Writes** go to the source, then the replicas. If the source is unreachable - * (`StorageUnavailableError`), they are applied locally and queued in a persisted outbox, then - * replayed in order when it's reachable again (last writer wins on conflict). While any write is - * queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source and - * fails offline. + * 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. */ export class KVStorageComposite extends BaseKVStorage { private readonly source: BaseKVStorage - /** Synced key range: `[prefix, prefix + '0')` is the prefix's own key and its subtree. */ - private readonly scope: { prefix: string; gte?: string; lt?: string } - private readonly replicas: BaseKVStorage[] - /** Persistent replica: seeds the others and holds the outbox and metadata. */ - private readonly persistent: BaseKVStorage - private readonly backoff: { baseDelay: number; maxDelay: number } - /** Queued-write count per key; remote changes to these keys are ignored until replayed. */ - private readonly pending = new Map() + 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 this client sent to a watchable 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. + * 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 Map() + private readonly echoes = new KeyCounter() private readonly offs: OffCallback[] = [] /** Mutable state lives here: child nodes share this object (see `child()`). */ private readonly state: { started?: Promise - /** Replicas hold a complete (possibly stale) copy, so reads can be served locally. */ + /** The replicas hold a complete (possibly stale) copy, so reads can be served locally. */ hasSnapshot: boolean - snapshotReady: Promise - markSnapshot: () => void + snapshotTaken: () => void + snapshot: Promise /** - * Remote changes and full syncs are applied strictly in order through this chain. Changes that - * arrive during a full sync wait behind it, so the end state matches the source even if it - * changed during the scan. - */ - applying: Promise - /** - * Keys written through the composite while a full sync runs. The diff may have scanned the - * source before such a write and the replica after it; these keys are left out of the diff - * (the write already reached the source and the replicas). + * 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 - replaying?: Promise - outboxSeq: number - retryTimer?: ReturnType - retryAttempt: number - closed: boolean } constructor(options: KVStorageCompositeOptions) { super() - if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica') this.source = options.source - const prefix = options.prefix ?? '' - this.scope = prefix === '' ? { prefix } : { prefix, gte: prefix, lt: prefix + Keys.afterSeparator } - this.replicas = options.replicas - this.persistent = options.replicas.at(-1)! - this.backoff = { baseDelay: 1000, maxDelay: 30_000, ...options.retry } - let markSnapshot!: () => void - const snapshotReady = new Promise((resolve) => (markSnapshot = resolve)) - this.state = { hasSnapshot: false, snapshotReady, markSnapshot, applying: Promise.resolve(), outboxSeq: 0, retryAttempt: 0, closed: false } + 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 } } /** Starts syncing (also happens on first use). Resolves once local data is loaded. */ @@ -135,296 +108,172 @@ export class KVStorageComposite extends BaseKVStorage< /** Resolves once the replicas hold a complete copy and all received remote changes are applied. */ async synced(): Promise { await this.start() - await this.state.snapshotReady - await this.state.applying + await this.state.snapshot + await this.applying.idle() } - /** Number of writes waiting to be replayed to the source. */ + /** Number of writes waiting to be sent to the source. */ get queued(): number { - let total = 0 - for (const count of this.pending.values()) total += count - return total + return this.outbox.size } - /** Stops watching the source and cancels retries. */ + /** Stops following the source and cancels retries. */ async close(): Promise { - this.state.closed = true - clearTimeout(this.state.retryTimer) + this.syncRetry.stop() + this.outboxRetry.stop() for (const off of this.offs.splice(0)) off() - if (this.watchable) await (this.source as unknown as WatchableSource).unsubscribe(this.scope.prefix) + if (this.subscribable) await this.subscribable.unsubscribe(this.range.prefix) } async readMany(keys: string[]): Promise { - this.checkScope(keys) + this.checkRange(keys) await this.start() - return (this.state.hasSnapshot ? this.replicas[0]! : this.source).readMany(keys) + return (this.state.hasSnapshot ? this.replicas.fastest : this.source).readMany(keys) } async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { await this.start() - // Narrow the range to the synced scope (and, against an authenticated server, to what it allows). - const gte = this.scope.gte !== undefined && (options.gte === undefined || options.gte < this.scope.gte) ? this.scope.gte : options.gte - const lt = this.scope.lt !== undefined && (options.lt === undefined || options.lt > this.scope.lt) ? this.scope.lt : options.lt - if (gte !== undefined && lt !== undefined && gte >= lt) return - const scoped = { ...options, gte, lt } - if (!this.state.hasSnapshot) return yield* this.source.scan(scoped) - // Skip internal keys before applying the limit. - let remaining = options.limit ?? Infinity - if (remaining <= 0) return - for await (const entry of this.replicas[0]!.scan({ ...scoped, limit: undefined })) { - if (entry[0] >= INTERNAL_PREFIX && entry[0] < INTERNAL_END) continue - yield entry - if (--remaining <= 0) return - } + if (this.state.hasSnapshot) return yield* this.replicas.scan(options) + const scoped = this.range.clamp(options) + if (scoped) yield* this.source.scan(scoped) } protected async applyWrite(entries: [string, unknown][]): Promise { - this.checkScope(entries.map(([key]) => key)) - await this.start() - await this.writeThrough({ type: 'write', entries }) + await this.write({ type: 'write', entries }) } protected async applyRemove(keys: string[]): Promise { - this.checkScope(keys) - await this.start() - await this.writeThrough({ type: 'remove', keys }) + await this.write({ type: 'remove', keys }) } protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { - this.checkScope(entries.map(([key]) => key)) + this.checkRange(entries.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.applyToReplicas(changes) + await this.replicas.apply(changes) return written } // ------------------------------------------------------------------ start - private inScope(key: string): boolean { - const { prefix } = this.scope - return prefix === '' || key === prefix || key.startsWith(prefix + Keys.separator) - } - - private checkScope(keys: string[]): void { - for (const key of keys) if (!this.inScope(key)) throw new RangeError(`"${key}" is outside this composite's prefix "${this.scope.prefix}"`) - } - - private get watchable(): boolean { - return typeof (this.source as unknown as Partial).subscribe === 'function' - } - private async open(): Promise { - const [meta] = (await this.persistent.readMany([META_KEY])) as (SyncMeta | undefined)[] - for await (const [key, value] of this.persistent.scan({ gte: OUTBOX_PREFIX, lt: OUTBOX_END, values: true })) { - this.state.outboxSeq = Number(key.slice(OUTBOX_PREFIX.length)) - for (const k of KVStorageComposite.keysOf(value as OutboxEntry)) this.pending.set(k, (this.pending.get(k) ?? 0) + 1) - } - await this.seedFasterReplicas() + const meta = await this.replicas.readMeta() + await this.outbox.load() + await this.replicas.seed() if (meta) this.snapshotTaken() this.offs.push( this.source.on('change', (event) => { - if (event.origin === 'remote') this.enqueueApply(() => this.applyRemote(event)) - if (this.queued > 0) void this.replay() + if (event.origin === 'remote') this.enqueue(() => this.applyRemote(event)) + if (this.outbox.size > 0) void this.outbox.replay() // the source is reachable again }), - this.source.on('resync', () => this.enqueueApply(() => this.fullSync())), + this.source.on('resync', () => this.enqueue(() => this.fullSync())), ) - if (this.watchable) { - const source = this.source as unknown as WatchableSource + 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 watching = source.subscribe(this.scope.prefix) - if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())).catch((error) => console.error(error)) - else void watching.catch((error) => console.error(error)) + 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)) } else { - this.enqueueApply(() => this.fullSync()) + this.enqueue(() => this.fullSync()) } - if (this.queued > 0) void this.replay() } - /** Copies the persistent replica into the faster ones. */ - private async seedFasterReplicas(): 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.persistent.scan({ values: true })) { - if (!entry[0].startsWith(INTERNAL_PREFIX) && batch.push(entry) >= SYNC_BATCH) await flush() - } - await flush() - } - - private snapshotTaken(): void { - this.state.hasSnapshot = true - this.state.markSnapshot() - } - - // ------------------------------------------------------------------ remote → replicas - - private enqueueApply(task: () => Promise): void { - this.state.applying = this.state.applying.then(task).catch((error) => { + /** Queues a sync task; failures are logged (an unreachable source is retried elsewhere). */ + private enqueue(task: () => Promise): void { + this.applying.run(task).catch((error) => { if (!(error instanceof StorageUnavailableError)) console.error(error) }) } + private snapshotTaken(): void { + this.state.hasSnapshot = true + this.state.snapshotTaken() + } + + private get subscribable(): SubscribableSource | undefined { + const source = this.source as unknown as Partial + return typeof source.subscribe === 'function' ? (source as SubscribableSource) : undefined + } + + 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 candidates = event.changes.filter( - ({ key }) => this.inScope(key) && !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX), + const relevant = event.changes.filter( + ({ key }) => this.range.contains(key) && !SYNC_INTERNAL.contains(key) && !this.echoes.take(key) && !this.outbox.pending.has(key), ) - // Skip changes the replicas already reflect. - const current = await this.replicas[0]!.readMany(candidates.map(({ key }) => key)) - const changes = candidates.filter(({ value }, i) => !Json.equal(value, current[i])) - await this.applyToReplicas(changes) - if (event.id) await this.persistent.writeMany([[META_KEY, { lastEventId: event.id } satisfies SyncMeta]]) + 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 }) } - /** Diffs the source against the persistent replica and applies the differences to every replica. */ + /** 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 id = this.watchable ? (this.source as unknown as WatchableSource).lastEventId : undefined - let changes: KVChange[] = [] + const lastEventId = this.subscribable?.lastEventId + let batch: KVChange[] = [] const flush = async () => { - const applied = changes.filter(({ key }) => !this.pending.has(key) && !writtenDuringSync.has(key)) - await this.applyToReplicas(applied) - if (applied.length) this.emit('change', { changes: applied, origin: 'remote' }) - changes = [] + 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.diff()) if (changes.push(change) >= SYNC_BATCH) await flush() + for await (const change of this.replicas.diff(this.source)) if (batch.push(change) >= 500) await flush() await flush() - await this.persistent.writeMany([[META_KEY, { lastEventId: id } satisfies SyncMeta]]) + await this.replicas.writeMeta({ lastEventId }) this.snapshotTaken() - this.state.retryAttempt = 0 + this.syncRetry.succeeded() } catch (error) { if (!(error instanceof StorageUnavailableError)) throw error - this.scheduleRetry(() => this.enqueueApply(() => this.fullSync())) + this.syncRetry.schedule(() => this.enqueue(() => this.fullSync())) } finally { this.state.writtenDuringSync = undefined } } - /** Changes that turn the persistent replica into the source (both scanned in key order). */ - private async *diff(): AsyncGenerator { - const range = { gte: this.scope.gte, lt: this.scope.lt, values: true } - const remote = this.source.scan(range)[Symbol.asyncIterator]() - const local = this.persistent.scan(range)[Symbol.asyncIterator]() - const nextLocal = async () => { - for (let next = await local.next(); ; next = await local.next()) if (next.done || !next.value[0].startsWith(INTERNAL_PREFIX)) return next - } - let r = await remote.next() - let l = await nextLocal() - 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 nextLocal() - } 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 nextLocal() - } - } - } + // ------------------------------------------------------------------ writes → source - private async applyToReplicas(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)))) - } - - // ------------------------------------------------------------------ local writes → source - - private async writeThrough(entry: OutboxEntry): Promise { - for (const key of KVStorageComposite.keysOf(entry)) this.state.writtenDuringSync?.add(key) - if (this.queued === 0) { + 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.applyToReplicas(KVStorageComposite.changesOf(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. - const seq = ++this.state.outboxSeq - for (const key of KVStorageComposite.keysOf(entry)) this.pending.set(key, (this.pending.get(key) ?? 0) + 1) - await this.persistent.writeMany([[OUTBOX_PREFIX + String(seq).padStart(12, '0'), entry]]) - await this.applyToReplicas(KVStorageComposite.changesOf(entry)) - void this.replay() + await this.outbox.push(entry) + await this.replicas.apply(KVOutbox.changesOf(entry)) } - /** Sends queued writes to the source in order; stops (and retries later) if it's unreachable. */ - private replay(): Promise { - const run = async () => { - try { - for (;;) { - let next: [string, unknown] | undefined - for await (const entry of this.persistent.scan({ gte: OUTBOX_PREFIX, lt: OUTBOX_END, limit: 1, values: true })) next = entry - if (!next) return - const [key, entry] = next as [string, OutboxEntry] - await this.send(entry) - await this.persistent.removeMany([key]) - for (const k of KVStorageComposite.keysOf(entry)) { - const count = this.pending.get(k)! - 1 - if (count > 0) this.pending.set(k, count) - else this.pending.delete(k) - } - this.state.retryAttempt = 0 - } - } catch (error) { - if (!(error instanceof StorageUnavailableError)) console.error(error) - this.scheduleRetry(() => void this.replay()) - } finally { - this.state.replaying = undefined - } - } - return (this.state.replaying ??= run()) - } - - /** Sends a write to the source, expecting its echo when the source is watched. */ + /** Sends a write to the source, expecting its echo when the source pushes changes. */ private async send(entry: OutboxEntry): Promise { - const keys = this.watchable ? KVStorageComposite.keysOf(entry) : [] - for (const key of keys) this.echoes.set(key, (this.echoes.get(key) ?? 0) + 1) + 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.consumeEcho(key) + for (const key of keys) this.echoes.take(key) throw error } } - - /** Whether `key` had an outstanding echo (now consumed). */ - private consumeEcho(key: string): boolean { - const count = this.echoes.get(key) - if (!count) return false - if (count > 1) this.echoes.set(key, count - 1) - else this.echoes.delete(key) - return true - } - - private scheduleRetry(task: () => void): void { - if (this.state.closed) return - clearTimeout(this.state.retryTimer) - const delay = ExponentialBackoff.calculateDelay({ ...this.backoff, growthRate: 2, jitter: 0.2, maxAttempts: 0 }, this.state.retryAttempt++) - this.state.retryTimer = setTimeout(task, delay) - ;(this.state.retryTimer as { unref?: () => void }).unref?.() - } - - private static keysOf(entry: OutboxEntry): string[] { - return entry.type === 'write' ? entry.entries.map(([key]) => key) : entry.keys - } - - private static changesOf(entry: OutboxEntry): KVChange[] { - return entry.type === 'write' ? entry.entries.map(([key, value]) => ({ key, value })) : entry.keys.map((key) => ({ key })) - } } diff --git a/src/kv-storage/kv-storage-http-server.ts b/src/kv-storage/kv-storage-http-server.ts index d785a2b..32ee001 100644 --- a/src/kv-storage/kv-storage-http-server.ts +++ b/src/kv-storage/kv-storage-http-server.ts @@ -1,6 +1,6 @@ import { AuthError } from '../utils/errors.js' import { ExtendedJson } from '../utils/extended-json.js' -import { Keys } from '../utils/keys.js' +import { KeyRange } from '../utils/key-range.js' import { RequestAuth, type RequestAuthOptions, type SignatureVerifier } from '../utils/request-auth.js' import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js' @@ -25,8 +25,8 @@ interface HistoryEntry { } interface Subscriber { - /** Keys whose own value and subtree this subscriber receives (`''` = everything). */ - paths: string[] + /** Keys (with their subtrees) this subscriber receives; `''` = everything. */ + ranges: KeyRange[] send: (frame: string) => void } @@ -121,21 +121,15 @@ export class KVStorageHttpServer { /** 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 + const allowed = new KeyRange(owner) for (const key of keys) { - if (typeof key !== 'string' || !KVStorageHttpServer.within(owner, key)) throw new AuthError(403, `No access to "${key}"`) + if (typeof key !== 'string' || !allowed.contains(key)) throw new AuthError(403, `No access to "${key}"`) } } - /** A scan must stay inside `[owner, owner + '0')`: the owner's key and its subtree. */ + /** A scan must stay inside the owner's key and its subtree. */ private authorizeRange(owner: string | undefined, options: ScanOptions): void { - if (owner === undefined) return - const { gte, lt } = options - if (gte === undefined || lt === undefined || gte < owner || lt > owner + Keys.afterSeparator) throw new AuthError(403, 'Scan outside your keys') - } - - /** Whether `key` is `path` or below it (`''` covers everything). */ - private static within(path: string, key: string): boolean { - return path === '' || key === path || key.startsWith(path + Keys.separator) + if (owner !== undefined && !new KeyRange(owner).covers(options.gte, options.lt)) throw new AuthError(403, 'Scan outside your keys') } // ------------------------------------------------------------------ change stream @@ -145,7 +139,7 @@ export class KVStorageHttpServer { 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.paths) + const frame = this.changeFrame(entry, subscriber.ranges) if (frame) subscriber.send(frame) } } @@ -156,8 +150,9 @@ export class KVStorageHttpServer { let keepAlive: ReturnType | undefined const stream = new ReadableStream({ start: (controller) => { - subscriber = { paths, send: (frame) => controller.enqueue(encoder.encode(frame)) } - subscriber.send(this.catchUpFrames(paths, lastEventId)) + const ranges = paths.map((path) => new KeyRange(path)) + subscriber = { ranges, send: (frame) => controller.enqueue(encoder.encode(frame)) } + subscriber.send(this.catchUpFrames(ranges, lastEventId)) this.subscribers.add(subscriber) keepAlive = setInterval(() => subscriber!.send(': keep-alive\n\n'), this.keepAliveMs) ;(keepAlive as { unref?: () => void }).unref?.() @@ -174,7 +169,7 @@ export class KVStorageHttpServer { * 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(paths: string[], lastEventId: string | null): string { + 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('-') @@ -185,14 +180,14 @@ export class KVStorageHttpServer { return ( this.history .filter((entry) => entry.seq > seq) - .map((entry) => this.changeFrame(entry, paths)) + .map((entry) => this.changeFrame(entry, ranges)) .join('') + ready ) } - /** A `change` frame with only the changes under `paths`, or `''` if none match. */ - private changeFrame(entry: HistoryEntry, paths: string[]): string { - const changes = entry.changes.filter(({ key }) => paths.some((path) => KVStorageHttpServer.within(path, key))) + /** 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) : '' } diff --git a/src/kv-storage/kv-storage-http.ts b/src/kv-storage/kv-storage-http.ts index 9cfe6cc..fd8bb64 100644 --- a/src/kv-storage/kv-storage-http.ts +++ b/src/kv-storage/kv-storage-http.ts @@ -3,6 +3,7 @@ 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' @@ -37,14 +38,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() /** Mutable state lives here: child nodes share this object (see `child()`). */ - private readonly state: { - paths: Set - session?: SSESession - lastEventId?: string - /** Subscription changes are applied one at a time. */ - updating: Promise - } = { paths: new Set(), updating: Promise.resolve() } + private readonly state: { paths: Set; session?: SSESession; lastEventId?: string } = { paths: new Set() } constructor(options: KVStorageHttpOptions) { super() @@ -141,9 +138,7 @@ export class KVStorageHttp extends BaseKVStorage { this.state.session = undefined if (this.state.paths.size > 0) await this.connect([...this.state.paths]) } - const result = this.state.updating.then(run) - this.state.updating = result.catch(() => {}) - return result + return this.updates.run(run) } private async connect(paths: string[]): Promise { diff --git a/src/utils/key-counter.ts b/src/utils/key-counter.ts new file mode 100644 index 0000000..9e5e016 --- /dev/null +++ b/src/utils/key-counter.ts @@ -0,0 +1,36 @@ +/** 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/key-range.ts b/src/utils/key-range.ts new file mode 100644 index 0000000..642a558 --- /dev/null +++ b/src/utils/key-range.ts @@ -0,0 +1,36 @@ +import type { ScanOptions } from '../kv-storage/base-kv-storage.js' +import { Keys } from './keys.js' + +/** + * A key and its whole subtree: `[prefix, prefix + '0')`. Escaping guarantees nothing else sorts + * in that range (not even lookalikes such as `prefix-x`). The empty prefix is every key. + */ +export class KeyRange { + /** Inclusive lower bound (`undefined` = unbounded). */ + readonly gte: string | undefined + /** Exclusive upper bound (`undefined` = unbounded). */ + readonly lt: string | undefined + + constructor(readonly prefix: string) { + this.gte = prefix === '' ? undefined : prefix + this.lt = prefix === '' ? undefined : prefix + Keys.afterSeparator + } + + contains(key: string): boolean { + return this.prefix === '' || key === this.prefix || key.startsWith(this.prefix + Keys.separator) + } + + /** Whether a scan over `[gte, lt)` stays inside this range. */ + covers(gte: string | undefined, lt: string | undefined): boolean { + if (this.prefix === '') return true + return gte !== undefined && lt !== undefined && gte >= this.gte! && lt <= this.lt! + } + + /** Scan options narrowed to this range, or `undefined` if nothing is left. */ + clamp(options: ScanOptions): ScanOptions | undefined { + const gte = this.gte !== undefined && (options.gte === undefined || options.gte < this.gte) ? this.gte : options.gte + const lt = this.lt !== undefined && (options.lt === undefined || options.lt > this.lt) ? this.lt : options.lt + if (gte !== undefined && lt !== undefined && gte >= lt) return undefined + return { ...options, gte, lt } + } +} diff --git a/src/utils/retry-timer.ts b/src/utils/retry-timer.ts new file mode 100644 index 0000000..ac04b2b --- /dev/null +++ b/src/utils/retry-timer.ts @@ -0,0 +1,36 @@ +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 new file mode 100644 index 0000000..943a634 --- /dev/null +++ b/src/utils/serial-queue.ts @@ -0,0 +1,17 @@ +/** 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 new file mode 100644 index 0000000..e4dc134 --- /dev/null +++ b/test/kv-storage/kv-outbox.test.ts @@ -0,0 +1,36 @@ +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/utils/building-blocks.test.ts b/test/utils/building-blocks.test.ts new file mode 100644 index 0000000..0c5eb21 --- /dev/null +++ b/test/utils/building-blocks.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, it, vi } 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', () => { + it('contains a key and its subtree, not lookalikes', () => { + const range = new KeyRange('ab') + expect(['ab', 'ab/c', 'ab/c/d'].every((k) => range.contains(k))).toBe(true) + expect(['a', 'abc', 'ab0', 'ab~2D1', 'b'].some((k) => range.contains(k))).toBe(false) + expect(new KeyRange('').contains('anything')).toBe(true) + }) + + it('agrees with its scan bounds for escaped keys (property)', () => { + const path = fc.array(fc.string({ unit: fc.constantFrom('a', 'b', '-', '/', '~', '0', '.'), minLength: 1, maxLength: 3 }), { minLength: 1, maxLength: 3 }) + fc.assert( + fc.property(path, path, (p, q) => { + const range = new KeyRange(Keys.join(p)) + const key = Keys.join(q) + expect(range.contains(key)).toBe(key >= range.gte! && key < range.lt!) + }), + ) + }) + + it('clamps and checks scan ranges', () => { + const range = new KeyRange('ab') + expect(range.clamp({ limit: 5 })).toEqual({ limit: 5, gte: 'ab', lt: 'ab0' }) + expect(range.clamp({ gte: 'ab/x', lt: 'zz' })).toEqual({ gte: 'ab/x', lt: 'ab0' }) + expect(range.clamp({ gte: 'b' })).toBeUndefined() + expect(range.covers('ab/', 'ab0')).toBe(true) + expect(range.covers('a', 'ab0')).toBe(false) + expect(range.covers(undefined, 'ab0')).toBe(false) + 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() + }) +})