Break KVStorageComposite into focused classes

- KVReplicaSet: seed, apply, local scan, diff against the source, sync meta
- KVOutbox: persisted offline queue with in-order replay (load() resumes it)
- utils: KeyRange (also used for server authorization), KeyCounter,
  SerialQueue (also used by KVStorageHttp and the event-sourced store),
  RetryTimer (separate timers for sync and replay; previously one shared
  timer could cancel the other's retry)

The composite now only orchestrates. Behaviour unchanged; new unit tests for
the building blocks and the outbox.

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

DESIGN.md

@@ -177,6 +177,18 @@ outbox `~q/o/<seq:12>`.
 - `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/<seq>`), 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`.

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<S extends Spec = AnySpec> {
  */
 export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
   private readonly maxRetries: number
-  /** Tail of the pending writes per document, so writes in one process don't conflict with each other. */
+  /** Writes to one document run one at a time, so writes in one process don't conflict with each other. */
-  private readonly locks = new Map<string, Promise<unknown>>()
+  private readonly writes = new SerialQueue()
 
   constructor(
     private readonly events: BaseEventStorage,
@@ -50,12 +51,7 @@ export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends BaseD
   }
 
   mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
-    const id = JSON.stringify(path)
+    return this.writes.run(() => this.write(path, change), 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
   }
 
   private async write(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {

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'

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/<seq:12>` 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<void> | undefined
+
+  constructor(
+    private readonly store: BaseKVStorage,
+    private readonly send: (entry: OutboxEntry) => Promise<void>,
+    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<void> {
+    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<void> {
+    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<void> {
+    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]
+    }
+  }
+}

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<SyncMeta | undefined> {
+    return (await this.persistent.readMany([META_KEY]))[0] as SyncMeta | undefined
+  }
+
+  async writeMeta(meta: SyncMeta): Promise<void> {
+    await this.persistent.writeMany([[META_KEY, meta]])
+  }
+
+  /** Copies the persistent replica into the faster ones. */
+  async seed(): Promise<void> {
+    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<void> {
+    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<KVChange[]> {
+    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<KVChange> {
+    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
+    }
+  }
+}

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 { KeyCounter } from '../utils/key-counter.js'
-import { Keys } from '../utils/keys.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
+   * Local copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be persistent:
-   * persistent: it also stores the offline outbox and sync metadata, and seeds the others on start.
+   * 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. */
+  /** Backoff for replaying queued writes and retrying a failed sync. */
-  retry?: Partial<{ baseDelay: number; maxDelay: number }>
+  retry?: RetryTimerOptions
 }
 
-/** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */
+/** Implemented by sources that push remote changes (e.g. `KVStorageHttp`). */
-interface WatchableSource {
+interface SubscribableSource {
   subscribe(path: string): Promise<void>
   unsubscribe(path: string): Promise<void>
   lastEventId: string | undefined
 }
 
-type OutboxEntry = { type: 'write'; entries: [string, unknown][] } | { type: 'remove'; keys: string[] }
+// Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates
-
+// them with Object.create, so they share this class's prototype).
-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).
 export interface KVStorageComposite<S extends Spec = AnySpec> {
   child<K extends KeyOf<S>>(key: K): KVStorageComposite<ChildOf<S, K>>
 }
 
 /**
- * Keeps local replicas of a source in sync, so reads (including `keys()`, `entries()` and document
+ * Keeps local replicas of (part of) a source in sync, so reads, including `keys()`, `entries()`
- * queries) are served locally, and the app keeps working offline.
+ * 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).
+ *   previous sync, reads are served locally straight away (possibly stale until caught up).
- * - **Staying in sync:** remote changes pushed by the source (`subscribe(prefix)`, e.g. SSE) are applied to
+ * - **Staying in sync:** remote changes pushed by the source are applied to every replica, in
- *   every replica. On restart the source resumes from the saved `lastEventId`; a full sync (diffing
+ *   order. On restart the source resumes from the saved `lastEventId`; a full sync (diff of the
- *   the source against the replicas) only runs the first time, or when the source says it can't
+ *   source against the replicas) only runs the first time, when the source can't resume
- *   resume (`resync`). Without `subscribe()`, a full sync runs on start.
+ *   (`resync`), or when it can't push changes at all.
- * - **Writes** go to the source, then the replicas. If the source is unreachable
+ * - **Writes** go to the source, then the replicas. When the source is unreachable they are
- *   (`StorageUnavailableError`), they are applied locally and queued in a persisted outbox, then
+ *   applied locally and queued in the outbox (`KVOutbox`), which replays them in order. While
- *   replayed in order when it's reachable again (last writer wins on conflict). While any write is
+ *   anything is queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source.
- *   queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source and
- *   fails offline.
  *
  * Write through the composite, not the source: changes made directly through the same source
  * instance (`origin: 'local'`) are not mirrored.
  */
 export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly source: BaseKVStorage
-  /** Synced key range: `[prefix, prefix + '0')` is the prefix's own key and its subtree. */
+  private readonly range: KeyRange
-  private readonly scope: { prefix: string; gte?: string; lt?: string }
+  private readonly replicas: KVReplicaSet
-  private readonly replicas: BaseKVStorage[]
+  private readonly outbox: KVOutbox
-  /** Persistent replica: seeds the others and holds the outbox and metadata. */
+  /** Remote changes and full syncs are applied strictly in order. */
-  private readonly persistent: BaseKVStorage
+  private readonly applying = new SerialQueue()
-  private readonly backoff: { baseDelay: number; maxDelay: number }
+  private readonly syncRetry: RetryTimer
-  /** Queued-write count per key; remote changes to these keys are ignored until replayed. */
+  private readonly outboxRetry: RetryTimer
-  private readonly pending = new Map<string, number>()
   /**
-   * Echoes still expected per key: one per write this client sent to a watchable source. A remote
+   * Echoes still expected per key: one per write sent to a subscribable source. A remote change to
-   * change to such a key is the echo or older than it (the stream is in server order), so it is
+   * such a key is the echo or older than it (the stream is in server order), so it is consumed
-   * consumed instead of applied; this avoids briefly showing a stale value.
+   * instead of applied; this avoids briefly showing a stale value.
    */
-  private readonly echoes = new Map<string, number>()
+  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<void>
-    /** 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<void>
+    snapshotTaken: () => void
-    markSnapshot: () => void
+    snapshot: Promise<void>
     /**
-     * Remote changes and full syncs are applied strictly in order through this chain. Changes that
+     * Keys written while a full sync runs. Its diff may have scanned the source before such a
-     * arrive during a full sync wait behind it, so the end state matches the source even if it
+     * write and the replica after it, so these keys are left out of the diff.
-     * changed during the scan.
-     */
-    applying: Promise<void>
-    /**
-     * 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).
      */
     writtenDuringSync?: Set<string>
-    replaying?: Promise<void>
-    outboxSeq: number
-    retryTimer?: ReturnType<typeof setTimeout>
-    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.range = new KeyRange(options.prefix ?? '')
-    this.scope = prefix === '' ? { prefix } : { prefix, gte: prefix, lt: prefix + Keys.afterSeparator }
+    this.replicas = new KVReplicaSet(options.replicas, this.range)
-    this.replicas = options.replicas
+    this.syncRetry = new RetryTimer(options.retry)
-    this.persistent = options.replicas.at(-1)!
+    this.outboxRetry = new RetryTimer(options.retry)
-    this.backoff = { baseDelay: 1000, maxDelay: 30_000, ...options.retry }
+    this.outbox = new KVOutbox(this.replicas.persistent, (entry) => this.send(entry), this.outboxRetry)
-    let markSnapshot!: () => void
+    let snapshotTaken!: () => void
-    const snapshotReady = new Promise<void>((resolve) => (markSnapshot = resolve))
+    const snapshot = new Promise<void>((resolve) => (snapshotTaken = resolve))
-    this.state = { hasSnapshot: false, snapshotReady, markSnapshot, applying: Promise.resolve(), outboxSeq: 0, retryAttempt: 0, closed: false }
+    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<S extends Spec = AnySpec> extends BaseKVStorage<
   /** Resolves once the replicas hold a complete copy and all received remote changes are applied. */
   async synced(): Promise<void> {
     await this.start()
-    await this.state.snapshotReady
+    await this.state.snapshot
-    await this.state.applying
+    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
+    return this.outbox.size
-    for (const count of this.pending.values()) total += count
-    return total
   }
 
-  /** Stops watching the source and cancels retries. */
+  /** Stops following the source and cancels retries. */
   async close(): Promise<void> {
-    this.state.closed = true
+    this.syncRetry.stop()
-    clearTimeout(this.state.retryTimer)
+    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<unknown[]> {
-    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).
+    if (this.state.hasSnapshot) return yield* this.replicas.scan(options)
-    const gte = this.scope.gte !== undefined && (options.gte === undefined || options.gte < this.scope.gte) ? this.scope.gte : options.gte
+    const scoped = this.range.clamp(options)
-    const lt = this.scope.lt !== undefined && (options.lt === undefined || options.lt > this.scope.lt) ? this.scope.lt : options.lt
+    if (scoped) yield* this.source.scan(scoped)
-    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
-    }
   }
 
   protected async applyWrite(entries: [string, unknown][]): Promise<void> {
-    this.checkScope(entries.map(([key]) => key))
+    await this.write({ type: 'write', entries })
-    await this.start()
-    await this.writeThrough({ type: 'write', entries })
   }
 
   protected async applyRemove(keys: string[]): Promise<void> {
-    this.checkScope(keys)
+    await this.write({ type: 'remove', keys })
-    await this.start()
-    await this.writeThrough({ type: 'remove', keys })
   }
 
   protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
-    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<WatchableSource>).subscribe === 'function'
-  }
-
   private async open(): Promise<void> {
-    const [meta] = (await this.persistent.readMany([META_KEY])) as (SyncMeta | undefined)[]
+    const meta = await this.replicas.readMeta()
-    for await (const [key, value] of this.persistent.scan({ gte: OUTBOX_PREFIX, lt: OUTBOX_END, values: true })) {
+    await this.outbox.load()
-      this.state.outboxSeq = Number(key.slice(OUTBOX_PREFIX.length))
+    await this.replicas.seed()
-      for (const k of KVStorageComposite.keysOf(value as OutboxEntry)) this.pending.set(k, (this.pending.get(k) ?? 0) + 1)
-    }
-    await this.seedFasterReplicas()
     if (meta) this.snapshotTaken()
 
     this.offs.push(
       this.source.on('change', (event) => {
-        if (event.origin === 'remote') this.enqueueApply(() => this.applyRemote(event))
+        if (event.origin === 'remote') this.enqueue(() => this.applyRemote(event))
-        if (this.queued > 0) void this.replay()
+        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.subscribable
-      const source = this.source as unknown as WatchableSource
+    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)
+      const subscribed = source.subscribe(this.range.prefix)
-      if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())).catch((error) => console.error(error))
+      if (!meta?.lastEventId) void subscribed.then(() => this.enqueue(() => this.fullSync())).catch((error) => console.error(error))
-      else void watching.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. */
+  /** Queues a sync task; failures are logged (an unreachable source is retried elsewhere). */
-  private async seedFasterReplicas(): Promise<void> {
+  private enqueue(task: () => Promise<void>): void {
-    const faster = this.replicas.slice(0, -1)
+    this.applying.run(task).catch((error) => {
-    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>): void {
-    this.state.applying = this.state.applying.then(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<SubscribableSource>
+    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<void> {
-    const candidates = event.changes.filter(
+    const relevant = event.changes.filter(
-      ({ key }) => this.inScope(key) && !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX),
+      ({ 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 changes = await this.replicas.unapplied(relevant)
-    const current = await this.replicas[0]!.readMany(candidates.map(({ key }) => key))
+    await this.replicas.apply(changes)
-    const changes = candidates.filter(({ value }, i) => !Json.equal(value, current[i]))
+    if (event.id) await this.replicas.writeMeta({ lastEventId: event.id })
-    await this.applyToReplicas(changes)
-    if (event.id) await this.persistent.writeMany([[META_KEY, { lastEventId: event.id } satisfies SyncMeta]])
     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<void> {
     // 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
+      const lastEventId = this.subscribable?.lastEventId
-      let changes: KVChange[] = []
+      let batch: KVChange[] = []
       const flush = async () => {
-        const applied = changes.filter(({ key }) => !this.pending.has(key) && !writtenDuringSync.has(key))
+        const changes = batch.filter((change) => !skip(change))
-        await this.applyToReplicas(applied)
+        await this.replicas.apply(changes)
-        if (applied.length) this.emit('change', { changes: applied, origin: 'remote' })
+        if (changes.length) this.emit('change', { changes, origin: 'remote' })
-        changes = []
+        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). */
+  // ------------------------------------------------------------------ writes → source
-  private async *diff(): AsyncGenerator<KVChange> {
-    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()
-      }
-    }
-  }
 
-  private async applyToReplicas(changes: KVChange[]): Promise<void> {
+  private async write(entry: OutboxEntry): Promise<void> {
-    const writes = changes.filter((c) => c.value !== undefined).map((c): [string, unknown] => [c.key, c.value])
+    const keys = KVOutbox.keysOf(entry)
-    const removes = changes.filter((c) => c.value === undefined).map((c) => c.key)
+    this.checkRange(keys)
-    await Promise.all(this.replicas.map(async (replica) => (await replica.writeMany(writes), await replica.removeMany(removes))))
+    await this.start()
-  }
+    for (const key of keys) this.state.writtenDuringSync?.add(key)
-
+    if (this.outbox.size === 0) {
-  // ------------------------------------------------------------------ local writes → source
-
-  private async writeThrough(entry: OutboxEntry): Promise<void> {
-    for (const key of KVStorageComposite.keysOf(entry)) this.state.writtenDuringSync?.add(key)
-    if (this.queued === 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
+    await this.outbox.push(entry)
-    for (const key of KVStorageComposite.keysOf(entry)) this.pending.set(key, (this.pending.get(key) ?? 0) + 1)
+    await this.replicas.apply(KVOutbox.changesOf(entry))
-    await this.persistent.writeMany([[OUTBOX_PREFIX + String(seq).padStart(12, '0'), entry]])
-    await this.applyToReplicas(KVStorageComposite.changesOf(entry))
-    void this.replay()
   }
 
-  /** Sends queued writes to the source in order; stops (and retries later) if it's unreachable. */
+  /** Sends a write to the source, expecting its echo when the source pushes changes. */
-  private replay(): Promise<void> {
-    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. */
   private async send(entry: OutboxEntry): Promise<void> {
-    const keys = this.watchable ? KVStorageComposite.keysOf(entry) : []
+    const keys = this.subscribable ? KVOutbox.keysOf(entry) : []
-    for (const key of keys) this.echoes.set(key, (this.echoes.get(key) ?? 0) + 1)
+    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 }))
-  }
 }

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). */
+  /** Keys (with their subtrees) this subscriber receives; `''` = everything. */
-  paths: string[]
+  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
+    if (owner !== undefined && !new KeyRange(owner).covers(options.gte, options.lt)) throw new AuthError(403, 'Scan outside your keys')
-    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)
   }
 
   // ------------------------------------------------------------------ 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<typeof setInterval> | undefined
     const stream = new ReadableStream<Uint8Array>({
       start: (controller) => {
-        subscriber = { paths, send: (frame) => controller.enqueue(encoder.encode(frame)) }
+        const ranges = paths.map((path) => new KeyRange(path))
-        subscriber.send(this.catchUpFrames(paths, lastEventId))
+        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. */
+  /** A `change` frame with only the changes inside `ranges`, or `''` if none match. */
-  private changeFrame(entry: HistoryEntry, paths: string[]): string {
+  private changeFrame(entry: HistoryEntry, ranges: KeyRange[]): string {
-    const changes = entry.changes.filter(({ key }) => paths.some((path) => KVStorageHttpServer.within(path, key)))
+    const changes = entry.changes.filter(({ key }) => ranges.some((range) => range.contains(key)))
     return changes.length ? this.frame('change', entry.seq, changes) : ''
   }
 

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<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly headers: Record<string, string>
   private readonly fetcher: (url: string, init: RequestInit) => Promise<Response>
   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: {
+  private readonly state: { paths: Set<string>; session?: SSESession; lastEventId?: string } = { paths: new Set() }
-    paths: Set<string>
-    session?: SSESession
-    lastEventId?: string
-    /** Subscription changes are applied one at a time. */
-    updating: Promise<void>
-  } = { paths: new Set(), updating: Promise.resolve() }
 
   constructor(options: KVStorageHttpOptions) {
     super()
@@ -141,9 +138,7 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
       this.state.session = undefined
       if (this.state.paths.size > 0) await this.connect([...this.state.paths])
     }
-    const result = this.state.updating.then(run)
+    return this.updates.run(run)
-    this.state.updating = result.catch(() => {})
-    return result
   }
 
   private async connect(paths: string[]): Promise<void> {

src/utils/key-counter.ts

@@ -0,0 +1,36 @@
+/** A count per key (a multiset). */
+export class KeyCounter {
+  private readonly counts = new Map<string, number>()
+  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<string>): 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
+  }
+}

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 }
+  }
+}

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<typeof setTimeout> | 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)
+  }
+}

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<string, Promise<unknown>>()
+
+  run<T>(task: () => Promise<T>, key: string = ''): Promise<T> {
+    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<void> {
+    while (this.tails.has(key)) await this.tails.get(key)
+  }
+}

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])
+  })
+})

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<void>((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()
+  })
+})