Add KVStorageComposite, IndexedDB storage and SSE change streams

- Every KV storage emits change events (base class owns writeMany/removeMany/ writeIfAbsent; implementations provide protected apply* methods) - KVStorageHttpServer streams changes over SSE (GET /events) with a resumable history; KVStorageHttp.watch() subscribes and resumes via Last-Event-ID - KVStorageIndexedDB for local persistence - KVStorageComposite keeps replicas of a source in sync: local reads, resume instead of full sync, offline outbox replayed in order, echo tracking - Values may contain bigint/Uint8Array (Extended JSON over the network) - Vendor sse-session, event-emitter and extended-json (no runtime deps) - Freeze child nodes so shadowed state on children fails loudly - CI on Node 22, 24 and 26 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE
2026-10-23 08:41:42 +00:00
parent 08fb78721f
commit 4ffbc7e78d
36 changed files with 2679 additions and 99 deletions
@@ -0,0 +1,379 @@
+import { StorageUnavailableError } from '../utils/errors.js'
+import type { OffCallback } from '../utils/event-emitter.js'
+import { Json } from '../utils/json.js'
+import type { AnySpec, Spec } from '../utils/schema.js'
+import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js'
+import { BaseKVStorage, type KVChange, type KVChangeEvent, type ScanOptions } from './base-kv-storage.js'
+
+export interface KVStorageCompositeOptions {
+  /** The source of truth (e.g. `KVStorageHttp`). */
+  source: BaseKVStorage
+  /**
+   * 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.
+   */
+  replicas: BaseKVStorage[]
+  /** Delays between attempts to replay queued writes or re-sync. */
+  retry?: Partial<{ baseDelay: number; maxDelay: number }>
+}
+
+/** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */
+interface WatchableSource {
+  watch(): Promise<void>
+  unwatch(): Promise<void>
+  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
+
+/**
+ * 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.
+ *
+ * - **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 (`watch()`, 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 `watch()`, 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.
+ *
+ * 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
+  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<string, number>()
+  /**
+   * 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.
+   */
+  private readonly echoes = new Map<string, number>()
+  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. */
+    hasSnapshot: boolean
+    snapshotReady: Promise<void>
+    markSnapshot: () => void
+    /**
+     * 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<void>
+    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
+    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<void>((resolve) => (markSnapshot = resolve))
+    this.state = { hasSnapshot: false, snapshotReady, markSnapshot, applying: Promise.resolve(), outboxSeq: 0, retryAttempt: 0, closed: false }
+  }
+
+  /** Starts syncing (also happens on first use). Resolves once local data is loaded. */
+  start(): Promise<void> {
+    return (this.state.started ??= this.open())
+  }
+
+  /** 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.applying
+  }
+
+  /** Number of writes waiting to be replayed to the source. */
+  get queued(): number {
+    let total = 0
+    for (const count of this.pending.values()) total += count
+    return total
+  }
+
+  /** Stops watching the source and cancels retries. */
+  async close(): Promise<void> {
+    this.state.closed = true
+    clearTimeout(this.state.retryTimer)
+    for (const off of this.offs.splice(0)) off()
+    if (this.watchable) await (this.source as unknown as WatchableSource).unwatch()
+  }
+
+  async readMany(keys: string[]): Promise<unknown[]> {
+    await this.start()
+    return (this.state.hasSnapshot ? this.replicas[0]! : this.source).readMany(keys)
+  }
+
+  async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> {
+    await this.start()
+    if (!this.state.hasSnapshot) return yield* this.source.scan(options)
+    // 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({ ...options, 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> {
+    await this.start()
+    await this.writeThrough({ type: 'write', entries })
+  }
+
+  protected async applyRemove(keys: string[]): Promise<void> {
+    await this.start()
+    await this.writeThrough({ type: 'remove', keys })
+  }
+
+  protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
+    await this.start()
+    const written = await this.source.writeIfAbsent(entries)
+    await this.applyToReplicas(entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })))
+    return written
+  }
+
+  // ------------------------------------------------------------------ start
+
+  private get watchable(): boolean {
+    return typeof (this.source as unknown as Partial<WatchableSource>).watch === 'function'
+  }
+
+  private async open(): Promise<void> {
+    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()
+    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()
+      }),
+      this.source.on('resync', () => this.enqueueApply(() => this.fullSync())),
+    )
+
+    if (this.watchable) {
+      const source = this.source as unknown as WatchableSource
+      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.watch()
+      if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync()))
+    } else {
+      this.enqueueApply(() => this.fullSync())
+    }
+    if (this.queued > 0) void this.replay()
+  }
+
+  /** Copies the persistent replica into the faster ones. */
+  private async seedFasterReplicas(): 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.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 async applyRemote(event: KVChangeEvent): Promise<void> {
+    const candidates = event.changes.filter(({ key }) => !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX))
+    // 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]])
+    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. */
+  private async fullSync(): Promise<void> {
+    // The diff re-reads the source, so outstanding echoes no longer matter (and may never arrive).
+    this.echoes.clear()
+    try {
+      const id = this.watchable ? (this.source as unknown as WatchableSource).lastEventId : undefined
+      let changes: KVChange[] = []
+      const flush = async () => {
+        const applied = changes.filter(({ key }) => !this.pending.has(key))
+        await this.applyToReplicas(applied)
+        if (applied.length) this.emit('change', { changes: applied, origin: 'remote' })
+        changes = []
+      }
+      for await (const change of this.diff()) if (changes.push(change) >= SYNC_BATCH) await flush()
+      await flush()
+      await this.persistent.writeMany([[META_KEY, { lastEventId: id } satisfies SyncMeta]])
+      this.snapshotTaken()
+      this.state.retryAttempt = 0
+    } catch (error) {
+      if (!(error instanceof StorageUnavailableError)) throw error
+      this.scheduleRetry(() => this.enqueueApply(() => this.fullSync()))
+    }
+  }
+
+  /** Changes that turn the persistent replica into the source (both scanned in key order). */
+  private async *diff(): AsyncGenerator<KVChange> {
+    const remote = this.source.scan({ values: true })[Symbol.asyncIterator]()
+    const local = this.persistent.scan({ values: true })[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> {
+    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<void> {
+    if (this.queued === 0) {
+      try {
+        await this.send(entry)
+        return this.applyToReplicas(KVStorageComposite.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()
+  }
+
+  /** Sends queued writes to the source in order; stops (and retries later) if it's unreachable. */
+  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) : []
+    for (const key of keys) this.echoes.set(key, (this.echoes.get(key) ?? 0) + 1)
+    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)
+      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 }))
+  }
+}