Plain operations: last write wins, hash-keyed events, simpler composite

- KV: one write primitive, write(changes) with { key, value? }; writeIfAbsent removed.
  The server executes exactly what clients send and never reads values, so they can be
  encrypted end to end. Adds setFields/getFields on the base class.
- HTTP: no server history or event ids; SSE relays every change (own writes included) and
  the client emits resync when a dropped stream reconnects.
- Composite: writes apply locally, then go to the source (undone locally if refused);
  pushed changes are applied and announced only where they differ from local values;
  offline writes go to a queue store (any KV storage); sync() = replay, diff, replay.
  Removes KVReplicaSet, KVOutbox, KeyCounter, SerialQueue and RetryTimer.
- Events: id = <time>_<sha256(time + canonical data)>, so concurrent appends never
  collide and retries can't duplicate; no versions or ConcurrencyError.
- Event-sourced documents: read, resolve, append, apply; catchUp folds in foreign events;
  getAt takes an event id.

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

src/document-storage/document-storage-kv.ts

@@ -26,7 +26,7 @@ export class DocumentStorageKV<S extends Spec = AnySpec> extends BaseDocumentSto
     const resolved = change(current)
     if (!resolved) return current
     const next = DocumentUpdate.apply(current, resolved)
-    if (next) await this.kv.writeMany([[key, next]])
+    if (next) await this.kv.writeMany([{ key, value: next }])
     else await this.kv.removeMany([key])
     return next
   }

src/document-storage/event-sourced-document-storage.ts

@@ -2,16 +2,9 @@ import type { BaseEventStorage, StoredEvent } from '../event-storage/base-event-
 import type { Doc } from '../utils/document-path.js'
 import type { DocumentQuery, FindResult } from '../utils/document-query.js'
 import { DocumentUpdate, type DocChange } from '../utils/document-update.js'
-import { ConcurrencyError } from '../utils/errors.js'
-import { SerialQueue } from '../utils/serial-queue.js'
 import type { AnySpec, ChildOf, KeyOf, Spec, ValueAt } from '../utils/schema.js'
 import { BaseDocumentStorage } from './base-document-storage.js'
 
-export interface EventSourcedDocumentStorageOptions {
-  /** Retries after a write conflict, each preceded by a catch-up. Default 5. */
-  maxRetries?: number
-}
-
 // Types only: children of an event-sourced store are event-sourced stores too (child() creates
 // them with Object.create, so they share this class's prototype).
 export interface EventSourcedDocumentStorage<S extends Spec = AnySpec> {
@@ -19,103 +12,66 @@ export interface EventSourcedDocumentStorage<S extends Spec = AnySpec> {
 }
 
 /**
- * Documents as event streams: each document's changes are appended to its own stream, and the
- * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all
- * reads and queries.
+ * Documents as event streams: each change to a document is appended to its own stream, and the
+ * current state is kept in a projection store (any `BaseDocumentStorage`), which serves all reads
+ * and queries.
  *
- * Projections carry a `_version` field (stripped from results) used as `expectedVersion` on the
- * next append. If a write crashes between appending and updating the projection, the next
- * write's append conflicts, the projection is rebuilt from the stream, and the write retries.
- * `_version` is therefore reserved in documents.
+ * A document's state is its events folded in order. Changes are stored as resolved values
+ * (`$inc` becomes `set: { n: 6 }`), so two devices editing the same field concurrently end with
+ * the later event's value; edits to different fields both survive.
+ *
+ * The projection is updated by this store's own writes. Events written elsewhere (another device,
+ * arriving through sync) are folded in by `catchUp(key)`, which rebuilds the document from its
+ * whole stream; call it when such events arrive.
  */
 export class EventSourcedDocumentStorage<S extends Spec = AnySpec> extends BaseDocumentStorage<S> {
-  private readonly maxRetries: number
-  /** Writes to one document run one at a time, so writes in one process don't conflict with each other. */
-  private readonly writes = new SerialQueue()
-
   constructor(
     private readonly events: BaseEventStorage,
     private readonly projections: BaseDocumentStorage,
-    options: EventSourcedDocumentStorageOptions = {},
   ) {
     super()
-    this.maxRetries = options.maxRetries ?? 5
   }
 
-  async readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
-    return (await this.projections.readMany(paths)).map((doc) => EventSourcedDocumentStorage.strip(doc))
+  readMany(paths: (readonly string[])[]): Promise<(Doc | undefined)[]> {
+    return this.projections.readMany(paths)
   }
 
-  async *query(parent: readonly string[], query: DocumentQuery): AsyncGenerator<FindResult> {
-    for await (const { key, data } of this.projections.query(parent, query)) yield { key, data: EventSourcedDocumentStorage.strip(data)! }
+  query(parent: readonly string[], query: DocumentQuery): AsyncIterable<FindResult> {
+    return this.projections.query(parent, query)
   }
 
-  mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
-    return this.writes.run(() => this.write(path, change), JSON.stringify(path))
+  async mutate(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
+    const [current] = await this.projections.readMany([path])
+    const resolved = change(current)
+    if (!resolved) return current
+    await this.events.appendTo(path, [resolved])
+    // Apply the change itself (not the state computed above), so concurrent writes in this
+    // process each land on the latest projection.
+    return this.projections.mutate(path, () => resolved)
   }
 
-  private async write(path: readonly string[], change: (current: Doc | undefined) => DocChange | null): Promise<Doc | undefined> {
-    for (let attempt = 0; ; attempt++) {
-      const [stored] = await this.projections.readMany([path])
-      const version = (stored?._version as number | undefined) ?? (await this.events.versionOf(path))
-      if (!stored && version > 0) {
-        // No projection but the stream has history (deleted, or a crashed first write): rebuild.
-        if ((await this.rebuild(path)) !== undefined) continue
-      }
-      const current = EventSourcedDocumentStorage.strip(stored)
-      const resolved = change(current)
-      if (!resolved) return current
-      const next = DocumentUpdate.apply(current, resolved)
-      try {
-        await this.events.appendTo(path, [resolved], version)
-      } catch (error) {
-        if (!(error instanceof ConcurrencyError) || attempt >= this.maxRetries) throw error
-        await this.rebuild(path)
-        continue
-      }
-      await this.project(path, next, version + 1)
-      return next
-    }
-  }
-
-  /** Events of this document (or a child). */
+  /** Events of this document (or a child), oldest first. */
   history(key?: KeyOf<S>): AsyncIterable<StoredEvent<DocChange>> {
     return this.events.readFrom(this.pathOf(arguments.length > 0, key), {}) as AsyncIterable<StoredEvent<DocChange>>
   }
 
-  /** State as of `version`: `getAt(version)` for this document, `getAt(key, version)` for a child. */
-  async getAt<K extends KeyOf<S>>(...args: [version: number] | [key: K, version: number]): Promise<ValueAt<S, K> | undefined> {
-    const path = this.pathOf(args.length > 1, args[0])
+  /** State right after event `eventId`: `getAt(eventId)` for this document, `getAt(key, eventId)` for a child. */
+  async getAt<K extends KeyOf<S>>(...args: [eventId: string] | [key: K, eventId: string]): Promise<ValueAt<S, K> | undefined> {
+    const until = args[args.length - 1] as string
     let state: Doc | undefined
-    for await (const event of this.events.readFrom(path, { to: args[args.length - 1] as number })) {
+    for await (const event of this.events.readFrom(this.pathOf(args.length > 1, args[0]), {})) {
+      if (event.id > until) break
       state = DocumentUpdate.apply(state, event.data as DocChange)
     }
     return state as ValueAt<S, K> | undefined
   }
 
-  /** Rebuilds the projection of this document (or a child) from its stream. Returns the state. */
-  catchUp(key?: KeyOf<S>): Promise<Doc | undefined> {
-    return this.rebuild(this.pathOf(arguments.length > 0, key))
-  }
-
-  private async rebuild(path: readonly string[]): Promise<Doc | undefined> {
+  /** Rebuilds the projection of this document (or a child) from all its events. Returns the state. */
+  async catchUp(key?: KeyOf<S>): Promise<Doc | undefined> {
+    const path = this.pathOf(arguments.length > 0, key)
     let state: Doc | undefined
-    let version = 0
-    for await (const event of this.events.readFrom(path, {})) {
-      state = DocumentUpdate.apply(state, event.data as DocChange)
-      version = event.version
-    }
-    await this.project(path, state, version)
-    return state
-  }
-
-  private async project(path: readonly string[], state: Doc | undefined, version: number): Promise<void> {
-    await this.projections.mutate(path, () => (state ? { op: 'set', doc: { ...state, _version: version } } : { op: 'delete' }))
-  }
-
-  private static strip(doc: Doc | undefined): Doc | undefined {
-    if (!doc) return undefined
-    const { _version, ...state } = doc
+    for await (const event of this.events.readFrom(path, {})) state = DocumentUpdate.apply(state, event.data as DocChange)
+    await this.projections.mutate(path, () => (state ? { op: 'set', doc: state } : { op: 'delete' }))
     return state
   }
 }

src/event-storage/base-event-storage.ts

@@ -1,27 +1,17 @@
 import { Keys } from '../utils/keys.js'
 import type { AnySpec, ChildOf, KeyOf, KeyString, Spec, ValueAt, ValueOf } from '../utils/schema.js'
 
-export interface AppendOptions {
-  /** Version the stream must be at (0 = empty). Default `'any'`. */
-  expectedVersion?: number | 'any'
-}
-
-export interface AppendResult {
-  /** Stream version after the append. */
-  version: number
-}
-
 export interface ReadOptions {
-  /** First version to read (inclusive). Default 1. */
-  from?: number
-  /** Last version to read (inclusive). Default: the head. */
-  to?: number
+  /** Start after this event id (exclusive), in the direction of travel. */
+  after?: string
   limit?: number
+  reverse?: boolean
 }
 
 export interface StoredEvent<E> {
-  version: number
-  /** Append timestamp (ms since epoch). */
+  /** Unique, time-ordered id (see the implementation for its format). */
+  id: string
+  /** Append time (ms since epoch). */
   at: number
   data: E
 }
@@ -34,16 +24,20 @@ export interface StreamKeysOptions {
 
 /**
  * Append-only event streams. Every node is a stream (its own events) and a parent of child
- * streams. Implementations provide four primitives over stream paths; the user-facing,
+ * streams. Implementations provide three primitives over stream paths; the user-facing,
  * self-or-child API is defined here once.
+ *
+ * Appends never conflict: concurrent writers each add their own events, and every reader sees all
+ * of them in the same (time) order. Folding events into state is where concurrent changes meet.
  */
 export abstract class BaseEventStorage<S extends Spec = AnySpec> {
   /** Unescaped path of this node (`[]` for the root, which is not a stream). */
   readonly path: readonly string[] = []
 
-  abstract appendTo(stream: readonly string[], events: unknown[], expectedVersion: number | 'any'): Promise<AppendResult>
+  /** Appends events in order; returns their ids. */
+  abstract appendTo(stream: readonly string[], events: unknown[]): Promise<string[]>
+  /** Events in id (time) order. */
   abstract readFrom(stream: readonly string[], options: ReadOptions): AsyncIterable<StoredEvent<unknown>>
-  abstract versionOf(stream: readonly string[]): Promise<number>
   abstract streamsOf(parent: readonly string[], options: StreamKeysOptions): AsyncIterable<string>
 
   child<K extends KeyOf<S>>(key: K): BaseEventStorage<ChildOf<S, K>> {
@@ -59,15 +53,12 @@ export abstract class BaseEventStorage<S extends Spec = AnySpec> {
     return this as unknown as BaseEventStorage<S2>
   }
 
-  /** `append(events, options?)` on this stream; `append(key, events, options?)` on a child stream. */
-  append<K extends KeyOf<S>>(
-    ...args: [events: ValueOf<S>[], options?: AppendOptions] | [key: K, events: ValueOf<ChildOf<S, K>>[], options?: AppendOptions]
-  ): Promise<AppendResult> {
+  /** `append(events)` to this stream; `append(key, events)` to a child stream. Returns the event ids. */
+  append<K extends KeyOf<S>>(...args: [events: ValueOf<S>[]] | [key: K, events: ValueOf<ChildOf<S, K>>[]]): Promise<string[]> {
     const hasKey = !Array.isArray(args[0])
     const events = args[hasKey ? 1 : 0]
-    const options = (args[hasKey ? 2 : 1] ?? {}) as AppendOptions
     if (!Array.isArray(events)) return Promise.reject(new TypeError('append() expects an array of events'))
-    return this.withStream(hasKey, args[0], (stream) => this.appendTo(stream, events, options.expectedVersion ?? 'any'))
+    return this.withStream(hasKey, args[0], (stream) => this.appendTo(stream, events))
   }
 
   /** `read(options?)` this stream; `read(key, options?)` a child stream. */
@@ -77,11 +68,6 @@ export abstract class BaseEventStorage<S extends Spec = AnySpec> {
     yield* this.readFrom(stream, ((hasKey ? args[1] : args[0]) ?? {}) as ReadOptions) as AsyncIterable<StoredEvent<ValueAt<S, K>>>
   }
 
-  /** Current version of this stream (or a child's); 0 when empty. */
-  version(key?: KeyOf<S>): Promise<number> {
-    return this.withStream(arguments.length > 0, key, (stream) => this.versionOf(stream))
-  }
-
   /** Keys of child streams. */
   keys(options: StreamKeysOptions = {}): AsyncIterable<KeyString<S>> {
     return this.streamsOf(this.path, options) as AsyncIterable<KeyString<S>>

src/event-storage/event-storage-kv.ts

@@ -1,151 +1,86 @@
 import type { BaseKVStorage } from '../kv-storage/base-kv-storage.js'
-import { ConcurrencyError } from '../utils/errors.js'
+import { ExtendedJson } from '../utils/extended-json.js'
+import { Hex } from '../utils/hex.js'
 import { Keys } from '../utils/keys.js'
-import type { AnySpec, Spec } from '../utils/schema.js'
-import { BaseEventStorage, type AppendResult, type ReadOptions, type StoredEvent, type StreamKeysOptions } from './base-event-storage.js'
-
-/** Stored value of one commit key. */
-interface Commit {
-  firstVersion: number
-  at: number
-  events: unknown[]
-}
-
-interface Head {
-  commit: number
-  version: number
-}
+import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js'
+import { BaseEventStorage, type ReadOptions, type StoredEvent, type StreamKeysOptions } from './base-event-storage.js'
 
 export interface EventStorageKVOptions {
-  /** Retries after losing a commit race when `expectedVersion` is `'any'`. Default 5. */
-  maxRetries?: number
-  /** Max streams whose head is cached. Default 10 000. */
-  cacheSize?: number
   now?: () => number
 }
 
+// Types only: children of an EventStorageKV are EventStorageKV nodes too (child() creates them with
+// Object.create, so they share this class's prototype).
+export interface EventStorageKV<S extends Spec = AnySpec> {
+  child<K extends KeyOf<S>>(key: K): EventStorageKV<ChildOf<S, K>>
+}
+
 /**
- * Event storage on any KV storage. Each append is one commit key, densely numbered:
- * `<stream>/~e/<commit:12>` → `{ firstVersion, at, events }`. Commits are claimed with
- * `writeIfAbsent`, so a writer with a stale view always collides with the existing key.
+ * Event storage on any KV storage. Each event is one plain write:
+ * `<stream>/~e/<time:15>_<hash>` → the event, where `hash` is the SHA-256 (hex) of the time and
+ * the event's canonical Extended JSON.
+ *
+ * - Concurrent appends from different writers get different keys, so all are kept.
+ * - Keys sort by time, so a stream reads in time order; equal times order by hash, the same way
+ *   for every reader. One instance never reuses a time, so its own appends keep their order.
+ * - The key can be recomputed from the content, and writing the same event again writes the same
+ *   key (e.g. a retried request), so it can't be duplicated.
  */
 export class EventStorageKV<S extends Spec = AnySpec> extends BaseEventStorage<S> {
-  private readonly heads = new Map<string, Head>()
-  private readonly maxRetries: number
-  private readonly cacheSize: number
   private readonly now: () => number
+  /** Mutable state lives here: child nodes share this object (see `child()`). */
+  private readonly state = { lastTime: 0 }
 
   constructor(
     private readonly kv: BaseKVStorage,
     options: EventStorageKVOptions = {},
   ) {
     super()
-    this.maxRetries = options.maxRetries ?? 5
-    this.cacheSize = options.cacheSize ?? 10_000
     this.now = options.now ?? Date.now
   }
 
-  async appendTo(stream: readonly string[], events: unknown[], expectedVersion: number | 'any'): Promise<AppendResult> {
-    const base = this.commitPrefix(stream)
-    for (let attempt = 0; ; attempt++) {
-      let head = this.heads.get(base) ?? (await this.loadHead(base))
-      // The cache may be stale; only trust a mismatch after re-reading.
-      if (expectedVersion !== 'any' && head.version !== expectedVersion) head = await this.loadHead(base)
-      if (expectedVersion !== 'any' && head.version !== expectedVersion) throw new ConcurrencyError(base, expectedVersion, head.version)
-      if (events.length === 0) return { version: head.version }
-
-      const commit: Commit = { firstVersion: head.version + 1, at: this.now(), events }
-      const [written] = await this.kv.writeIfAbsent([[base + EventStorageKV.pad(head.commit + 1), commit]])
-      if (written) {
-        const next = { commit: head.commit + 1, version: head.version + events.length }
-        this.cacheHead(base, next)
-        return { version: next.version }
-      }
-      this.heads.delete(base)
-      if (expectedVersion !== 'any' || attempt >= this.maxRetries) {
-        throw new ConcurrencyError(base, expectedVersion, (await this.loadHead(base)).version)
-      }
-    }
+  async appendTo(stream: readonly string[], events: unknown[]): Promise<string[]> {
+    const base = this.eventsPrefix(stream)
+    const entries = await Promise.all(
+      events.map(async (data) => {
+        this.state.lastTime = Math.max(this.now(), this.state.lastTime + 1)
+        const id = await EventStorageKV.id(this.state.lastTime, data)
+        return { key: base + id, value: data }
+      }),
+    )
+    await this.kv.writeMany(entries)
+    return entries.map(({ key }) => key.slice(base.length))
   }
 
   async *readFrom(stream: readonly string[], options: ReadOptions): AsyncGenerator<StoredEvent<unknown>> {
-    const from = Math.max(1, options.from ?? 1)
-    const to = options.to ?? Infinity
-    let remaining = options.limit ?? Infinity
-    if (to < from || remaining <= 0) return
-    const base = this.commitPrefix(stream)
-    const start = from === 1 ? 1 : await this.findCommit(base, from)
-    if (start === 0) return
-    for await (const [, value] of this.kv.scan({ gte: base + EventStorageKV.pad(start), lt: this.commitEnd(base), values: true })) {
-      const commit = value as Commit
-      for (let i = 0; i < commit.events.length; i++) {
-        const version = commit.firstVersion + i
-        if (version < from) continue
-        if (version > to || remaining-- <= 0) return
-        yield { version, at: commit.at, data: commit.events[i] }
-      }
+    const base = this.eventsPrefix(stream)
+    const { after, reverse = false, limit } = options
+    // `after` is exclusive: start at it and skip it.
+    const bounds = reverse
+      ? { gte: base, lt: after === undefined ? base.slice(0, -1) + Keys.afterSeparator : base + after }
+      : { gte: base + (after ?? ''), lt: base.slice(0, -1) + Keys.afterSeparator }
+    let remaining = limit ?? Infinity
+    for await (const [key, data] of this.kv.scan({ ...bounds, reverse, values: true })) {
+      if (remaining <= 0) return
+      const id = key.slice(base.length)
+      if (id === after) continue
+      remaining--
+      yield { id, at: Number(id.slice(0, 15)), data }
     }
   }
 
-  async versionOf(stream: readonly string[]): Promise<number> {
-    return (await this.loadHead(this.commitPrefix(stream))).version
-  }
-
   async *streamsOf(parent: readonly string[], options: StreamKeysOptions): AsyncGenerator<string> {
     yield* parent.reduce<BaseKVStorage>((node, segment) => node.child(segment), this.kv).keys(options)
   }
 
-  /** Last commit of a stream: one reverse scan with limit 1. Refreshes the cache. */
-  private async loadHead(base: string): Promise<Head> {
-    for await (const [key, value] of this.kv.scan({ gte: base, lt: this.commitEnd(base), reverse: true, limit: 1, values: true })) {
-      const commit = value as Commit
-      const head = { commit: Number(key.slice(base.length)), version: commit.firstVersion + commit.events.length - 1 }
-      this.cacheHead(base, head)
-      return head
-    }
-    return { commit: 0, version: 0 }
+  /** `<time:15>_<sha256(time + canonical data)>`. */
+  static async id(time: number, data: unknown): Promise<string> {
+    const stamp = String(time).padStart(15, '0')
+    const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(stamp + ExtendedJson.canonical(data)))
+    return `${stamp}_${Hex.encode(new Uint8Array(digest))}`
   }
 
-  /**
-   * Commit containing `version`, or 0 if the stream is shorter. Commit `n` starts at a version
-   * >= n, so it's at most `version`: try that first (exact when commits hold one event each),
-   * then binary-search the commits below it.
-   */
-  private async findCommit(base: string, version: number): Promise<number> {
-    const firstVersion = async (commit: number) => ((await this.kv.readMany([base + EventStorageKV.pad(commit)]))[0] as Commit | undefined)?.firstVersion
-    const guess = await firstVersion(version)
-    if (guess !== undefined && guess <= version) return version
-    let hi = version - 1
-    if (guess === undefined) {
-      const head = await this.loadHead(base)
-      if (head.version < version) return 0
-      hi = head.commit
-    }
-    let lo = 1
-    while (lo < hi) {
-      const mid = (lo + hi + 1) >>> 1
-      if ((await firstVersion(mid))! <= version) lo = mid
-      else hi = mid - 1
-    }
-    return lo
-  }
-
-  private cacheHead(base: string, head: Head): void {
-    this.heads.delete(base)
-    this.heads.set(base, head)
-    if (this.heads.size > this.cacheSize) this.heads.delete(this.heads.keys().next().value!)
-  }
-
-  private commitPrefix(stream: readonly string[]): string {
-    const prefix = this.kv.prefix + Keys.join(stream)
-    return `${prefix}/~e/`
-  }
-
-  private commitEnd(base: string): string {
-    return base.slice(0, -1) + Keys.afterSeparator
-  }
-
-  private static pad(commit: number): string {
-    return String(commit).padStart(12, '0')
+  private eventsPrefix(stream: readonly string[]): string {
+    return `${this.kv.prefix}${Keys.join(stream)}/~e/`
   }
 }

src/index.ts

@@ -4,8 +4,6 @@ export * from './kv-storage/kv-storage-http.js'
 export * from './kv-storage/kv-storage-http-server.js'
 export * from './kv-storage/kv-storage-indexeddb.js'
 export * from './kv-storage/kv-storage-composite.js'
-export * from './kv-storage/kv-replica-set.js'
-export * from './kv-storage/kv-outbox.js'
 export * from './event-storage/base-event-storage.js'
 export * from './event-storage/event-storage-kv.js'
 export * from './document-storage/base-document-storage.js'
@@ -19,10 +17,7 @@ export * from './utils/errors.js'
 export * from './utils/event-emitter.js'
 export * from './utils/extended-json.js'
 export * from './utils/hex.js'
-export * from './utils/key-counter.js'
 export * from './utils/key-range.js'
-export * from './utils/retry-timer.js'
-export * from './utils/serial-queue.js'
 export * from './utils/request-auth.js'
 export * from './utils/json.js'
 export * from './utils/keys.js'

src/kv-storage/base-kv-storage.ts

@@ -30,20 +30,22 @@ export interface KVChangeEvent {
   changes: KVChange[]
   /** `local`: made through this instance. `remote`: observed from elsewhere (e.g. pushed by a server). */
   origin: 'local' | 'remote'
-  /** Position in the source's change stream, when it has one (used to resume). */
-  id?: string
 }
 
 export type KVStorageEvents = {
   change: KVChangeEvent
-  /** Changes may have been missed (e.g. a resume point was too old): consumers should reload. */
-  resync: { id?: string }
+  /** Changes may have been missed (e.g. a dropped connection was restored): consumers should re-read. */
+  resync: void
 }
 
 /**
- * Ordered key-value storage. Implementations provide five raw primitives over absolute keys
- * (`readMany`, `scan`, and the protected `apply*` writes); everything else (navigation, self/child
- * reads and writes, child listing, change events) is defined here once.
+ * Ordered key-value storage. Implementations provide three primitives over absolute keys:
+ * `readMany`, `scan` and the protected `write`. Everything else (navigation, self/child reads and
+ * writes, child listing, change events) is defined here once.
+ *
+ * Writes are plain operations: set a key, or delete it. There are no versions or conditions;
+ * concurrent writers to one key end with whichever write arrived last. Use event storage when
+ * concurrent writes must all be kept.
  *
  * Every write emits a `change` event from the storage (shared by all its child nodes).
  *
@@ -63,31 +65,20 @@ export abstract class BaseKVStorage<S extends Spec = AnySpec> extends EventEmitt
   abstract readMany(keys: string[]): Promise<unknown[]>
   /** Entries in key order (descending when `reverse`). */
   abstract scan(options?: ScanOptions): AsyncIterable<[key: string, value: unknown]>
-  protected abstract applyWrite(entries: [key: string, value: unknown][]): Promise<void>
-  protected abstract applyRemove(keys: string[]): Promise<void>
-  /** Writes each entry only if its key is absent; `true` = written. Must be atomic per key. */
-  protected abstract applyWriteIfAbsent(entries: [key: string, value: unknown][]): Promise<boolean[]>
+  /** Applies changes in order: sets each key to its value, or deletes it when the value is `undefined`. */
+  protected abstract write(changes: KVChange[]): Promise<void>
 
   // ------------------------------------------------------------------ raw writes (absolute keys)
 
-  async writeMany(entries: [key: string, value: unknown][]): Promise<void> {
-    if (entries.length === 0) return
-    await this.applyWrite(entries)
-    this.emit('change', { changes: entries.map(([key, value]) => ({ key, value })), origin: 'local' })
+  /** Applies changes (`value: undefined` deletes) and emits them as a `change` event. */
+  async writeMany(changes: KVChange[]): Promise<void> {
+    if (changes.length === 0) return
+    await this.write(changes)
+    this.emit('change', { changes, origin: 'local' })
   }
 
   async removeMany(keys: string[]): Promise<void> {
-    if (keys.length === 0) return
-    await this.applyRemove(keys)
-    this.emit('change', { changes: keys.map((key) => ({ key })), origin: 'local' })
-  }
-
-  async writeIfAbsent(entries: [key: string, value: unknown][]): Promise<boolean[]> {
-    if (entries.length === 0) return []
-    const written = await this.applyWriteIfAbsent(entries)
-    const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value }))
-    if (changes.length) this.emit('change', { changes, origin: 'local' })
-    return written
+    await this.writeMany(keys.map((key) => ({ key })))
   }
 
   // ------------------------------------------------------------------ navigation
@@ -127,12 +118,24 @@ export abstract class BaseKVStorage<S extends Spec = AnySpec> extends EventEmitt
   async set<K extends KeyOf<S>>(...args: [value: ValueOf<S>] | [key: K, value: ValueOf<ChildOf<S, K>>]): Promise<void> {
     const value = args[args.length - 1]
     if (value === undefined) throw new TypeError('Cannot set undefined; use delete()')
-    await this.writeMany([[this.keyOf(args.length > 1, args[0]), value]])
+    await this.writeMany([{ key: this.keyOf(args.length > 1, args[0]), value }])
   }
 
   async setMany(entries: readonly EntryOf<S>[]): Promise<void> {
     if (entries.some(([, v]) => v === undefined)) throw new TypeError('Cannot set undefined; use delete()')
-    await this.writeMany(entries.map(([k, v]) => [this.keyOf(true, k), v]))
+    await this.writeMany(entries.map(([k, value]) => ({ key: this.keyOf(true, k), value })))
+  }
+
+  /** Writes each property of `fields` as a child key (one write). Nested objects stay single values. */
+  async setFields(fields: { [K in KeyString<S>]?: ValueOf<ChildOf<S, K & KeyOf<S>>> }): Promise<void> {
+    await this.setMany(Object.entries(fields).filter(([, v]) => v !== undefined) as unknown as EntryOf<S>[])
+  }
+
+  /** Reads the direct children that have a value into an object. */
+  async getFields(): Promise<{ [K in KeyString<S>]?: ValueOf<ChildOf<S, K & KeyOf<S>>> }> {
+    const fields: Record<string, unknown> = {}
+    for await (const [key, value] of this.entries()) fields[key] = value
+    return fields as { [K in KeyString<S>]?: ValueOf<ChildOf<S, K & KeyOf<S>>> }
   }
 
   /** Deletes this node's own value (or a child's), not descendants. */

src/kv-storage/kv-outbox.ts

@@ -1,91 +0,0 @@
-import { StorageUnavailableError } from '../utils/errors.js'
-import { KeyCounter } from '../utils/key-counter.js'
-import { KeyRange } from '../utils/key-range.js'
-import type { RetryTimer } from '../utils/retry-timer.js'
-import type { BaseKVStorage, KVChange } from './base-kv-storage.js'
-
-/** A write waiting to be sent to the source. */
-export type OutboxEntry = { type: 'write'; entries: [string, unknown][] } | { type: 'remove'; keys: string[] }
-
-const ENTRIES = new KeyRange('~q/o')
-
-/**
- * Writes made while the source was unreachable, persisted as `~q/o/<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

@@ -1,105 +0,0 @@
-import { Json } from '../utils/json.js'
-import { KeyRange } from '../utils/key-range.js'
-import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js'
-
-/** Sync bookkeeping kept in the persistent replica. */
-export interface SyncMeta {
-  /** Resume position in the source's change stream. */
-  lastEventId?: string
-}
-
-/** Keys reserved for sync bookkeeping (`~q/...`); never synced or shown. */
-export const SYNC_INTERNAL = new KeyRange('~q')
-const META_KEY = '~q/meta'
-const BATCH = 500
-
-/**
- * Local copies of one key range of a source, fastest first. The last replica is the persistent
- * one: it seeds the others and holds the sync metadata (and the outbox, see `KVOutbox`).
- */
-export class KVReplicaSet {
-  readonly fastest: BaseKVStorage
-  readonly persistent: BaseKVStorage
-
-  constructor(
-    private readonly replicas: BaseKVStorage[],
-    readonly range: KeyRange,
-  ) {
-    if (replicas.length === 0) throw new TypeError('At least one replica is required')
-    this.fastest = replicas[0]!
-    this.persistent = replicas.at(-1)!
-  }
-
-  async readMeta(): Promise<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,37 +1,33 @@
 import { StorageUnavailableError } from '../utils/errors.js'
 import type { OffCallback } from '../utils/event-emitter.js'
-import { KeyCounter } from '../utils/key-counter.js'
+import { Json } from '../utils/json.js'
 import { KeyRange } from '../utils/key-range.js'
-import { RetryTimer, type RetryTimerOptions } from '../utils/retry-timer.js'
 import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js'
-import { SerialQueue } from '../utils/serial-queue.js'
-import { BaseKVStorage, type KVChange, type KVChangeEvent, type ScanOptions } from './base-kv-storage.js'
-import { KVOutbox, type OutboxEntry } from './kv-outbox.js'
-import { KVReplicaSet, SYNC_INTERNAL } from './kv-replica-set.js'
+import { Ulid } from '../utils/ulid.js'
+import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js'
+import { KVStorageInMemory } from './kv-storage-in-memory.js'
 
 export interface KVStorageCompositeOptions {
-  /** The source of truth (e.g. `KVStorageHttp`). */
+  /** The source of truth, e.g. `KVStorageHttp`. */
   source: BaseKVStorage
-  /**
-   * Only this key and its subtree are synced, cached and subscribed to (e.g. the user's public
-   * key). Keys outside it are rejected. Default `''`: the whole source (only for sources the
-   * client may read entirely).
-   */
-  prefix?: string
-  /**
-   * Local copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be persistent:
-   * it seeds the others on start and holds the offline outbox and sync metadata.
-   */
+  /** Local copies, fastest first, e.g. `[inMemory, indexedDb]`. The last one seeds the others on start. */
   replicas: BaseKVStorage[]
-  /** Backoff for replaying queued writes and retrying a failed sync. */
-  retry?: RetryTimerOptions
+  /**
+   * Holds writes made while the source is unreachable (under its `writes` child) and whether a
+   * sync has completed (`synced`). Use a persistent node outside `prefix`, e.g.
+   * `indexedDb.child('sync')`, so queued writes survive a reload. Default: in memory.
+   */
+  queue?: BaseKVStorage
+  /** Only this key and its subtree are synced (e.g. the user's public key). Default: everything. */
+  prefix?: string
+  /** Delay before retrying when the source is unreachable. Default 5 000 ms. */
+  retryMs?: number
 }
 
-/** Implemented by sources that push remote changes (e.g. `KVStorageHttp`). */
-interface SubscribableSource {
+/** Sources that push remote changes (e.g. `KVStorageHttp`). */
+interface Subscribable {
   subscribe(path: string): Promise<void>
   unsubscribe(path: string): Promise<void>
-  lastEventId: string | undefined
 }
 
 // Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates
@@ -41,239 +37,236 @@ export interface KVStorageComposite<S extends Spec = AnySpec> {
 }
 
 /**
- * Keeps local replicas of (part of) a source in sync, so reads, including `keys()`, `entries()`
- * and document queries, are local, and the app keeps working offline.
+ * Combines a source with local replicas: reads come from the first replica, writes go to every
+ * replica and then the source, exactly as given (no merging: the last write wins). A write the
+ * source refuses (e.g. 403) is undone locally and rethrown.
  *
- * - **Start** (lazily, on first use): the persistent replica seeds the faster ones. If it holds a
- *   previous sync, reads are served locally straight away (possibly stale until caught up).
- * - **Staying in sync:** remote changes pushed by the source are applied to every replica, in
- *   order. On restart the source resumes from the saved `lastEventId`; a full sync (diff of the
- *   source against the replicas) only runs the first time, when the source can't resume
- *   (`resync`), or when it can't push changes at all.
- * - **Writes** go to the source, then the replicas. When the source is unreachable they are
- *   applied locally and queued in the outbox (`KVOutbox`), which replays them in order. While
- *   anything is queued, later writes queue too, to keep their order. `writeIfAbsent` needs the source.
- *
- * Write through the composite, not the source: changes made directly through the same source
- * instance (`origin: 'local'`) are not mirrored.
+ * - Changes pushed by the source (`subscribe`) are applied to the replicas, and announced as
+ *   `change` events, only where they differ from what the replicas hold. The source relaying this
+ *   client's own writes therefore changes nothing.
+ * - A write the source can't take (`StorageUnavailableError`) is applied locally and stored in
+ *   the `queue` store; later writes queue behind it, so order is kept.
+ * - `sync()` replays the queue, then copies what changed on the source into the replicas. It runs
+ *   on start, when the source reconnects (`resync`), and after a retry delay while offline.
+ * - Reads wait for the first sync (until one succeeds, they go to the source). Whether a sync has
+ *   completed is kept in the `queue` store, so with a persistent one a restart reads offline.
  */
 export class KVStorageComposite<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly source: BaseKVStorage
+  private readonly replicas: BaseKVStorage[]
+  private readonly queue: BaseKVStorage
+  private readonly writes: BaseKVStorage
   private readonly range: KeyRange
-  private readonly replicas: KVReplicaSet
-  private readonly outbox: KVOutbox
-  /** Remote changes and full syncs are applied strictly in order. */
-  private readonly applying = new SerialQueue()
-  private readonly syncRetry: RetryTimer
-  private readonly outboxRetry: RetryTimer
-  /**
-   * Echoes still expected per key: one per write sent to a subscribable source. A remote change to
-   * such a key is the echo or older than it (the stream is in server order), so it is consumed
-   * instead of applied; this avoids briefly showing a stale value.
-   */
-  private readonly echoes = new KeyCounter()
-  private readonly offs: OffCallback[] = []
+  private readonly retryMs: number
+  private readonly ids = new Ulid()
   /** Mutable state lives here: child nodes share this object (see `child()`). */
   private readonly state: {
     started?: Promise<void>
-    /** The replicas hold a complete (possibly stale) copy, so reads can be served locally. */
-    hasSnapshot: boolean
-    snapshotTaken: () => void
-    snapshot: Promise<void>
-    /**
-     * 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<string>
-  }
+    syncing?: Promise<void>
+    /** A sync has completed at some point, so the replicas hold a full copy. */
+    synced: boolean
+    /** Writes waiting in the queue. */
+    queued: number
+    /** Remote changes are applied one after another. */
+    applying: Promise<unknown>
+    retry?: ReturnType<typeof setTimeout>
+    offs: OffCallback[]
+  } = { synced: false, queued: 0, applying: Promise.resolve(), offs: [] }
 
   constructor(options: KVStorageCompositeOptions) {
     super()
+    if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica')
     this.source = options.source
+    this.replicas = options.replicas
+    this.queue = options.queue ?? new KVStorageInMemory()
+    this.writes = this.queue.child('writes')
     this.range = new KeyRange(options.prefix ?? '')
-    this.replicas = new KVReplicaSet(options.replicas, this.range)
-    this.syncRetry = new RetryTimer(options.retry)
-    this.outboxRetry = new RetryTimer(options.retry)
-    this.outbox = new KVOutbox(this.replicas.persistent, (entry) => this.send(entry), this.outboxRetry)
-    let snapshotTaken!: () => void
-    const snapshot = new Promise<void>((resolve) => (snapshotTaken = resolve))
-    this.state = { hasSnapshot: false, snapshot, snapshotTaken }
+    this.retryMs = options.retryMs ?? 5000
+    // A full sync makes the replicas match the source within the range, which would delete queued
+    // writes stored inside that range.
+    const queueRoot = KVStorageComposite.rootOf(this.queue)
+    if (this.replicas.some((replica) => KVStorageComposite.rootOf(replica) === queueRoot) && this.range.contains(this.queue.key)) {
+      throw new TypeError('The queue must be outside the synced range (e.g. a separate store, or a node outside `prefix`)')
+    }
   }
 
-  /** Starts syncing (also happens on first use). Resolves once local data is loaded. */
+  /** Loads local data and starts syncing (also happens on first use). */
   start(): Promise<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.snapshot
-    await this.applying.idle()
+  /** Replays queued writes, then brings the replicas up to date with the source (starts first if needed). */
+  sync(): Promise<void> {
+    return (this.state.syncing ??= this.runSync().finally(() => (this.state.syncing = undefined)))
   }
 
-  /** Number of writes waiting to be sent to the source. */
+  /** Number of writes waiting for the source. */
   get queued(): number {
-    return this.outbox.size
+    return this.state.queued
   }
 
-  /** Stops following the source and cancels retries. */
   async close(): Promise<void> {
-    this.syncRetry.stop()
-    this.outboxRetry.stop()
-    for (const off of this.offs.splice(0)) off()
-    if (this.subscribable) await this.subscribable.unsubscribe(this.range.prefix)
+    clearTimeout(this.state.retry)
+    for (const off of this.state.offs.splice(0)) off()
+    if (KVStorageComposite.isSubscribable(this.source)) await this.source.unsubscribe(this.range.prefix)
   }
 
   async readMany(keys: string[]): Promise<unknown[]> {
     this.checkRange(keys)
-    await this.start()
-    return (this.state.hasSnapshot ? this.replicas.fastest : this.source).readMany(keys)
+    return (await this.reader()).readMany(keys)
   }
 
   async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> {
-    await this.start()
-    if (this.state.hasSnapshot) return yield* this.replicas.scan(options)
     const scoped = this.range.clamp(options)
-    if (scoped) yield* this.source.scan(scoped)
+    if (scoped) yield* (await this.reader()).scan(scoped)
   }
 
-  protected async applyWrite(entries: [string, unknown][]): Promise<void> {
-    await this.write({ type: 'write', entries })
-  }
-
-  protected async applyRemove(keys: string[]): Promise<void> {
-    await this.write({ type: 'remove', keys })
-  }
-
-  protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
-    this.checkRange(entries.map(([key]) => key))
+  protected async write(changes: KVChange[]): Promise<void> {
+    const keys = changes.map(({ key }) => key)
+    this.checkRange(keys)
     await this.start()
-    const written = await this.source.writeIfAbsent(entries)
-    const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value }))
-    for (const { key } of changes) this.state.writtenDuringSync?.add(key)
-    await this.replicas.apply(changes)
-    return written
+    // Applied locally first, so the source relaying this write back finds nothing new.
+    const previous = await this.replicas[0]!.readMany(keys)
+    await this.apply(changes)
+    // While syncing or with writes queued, queue this one too so the source sees writes in order.
+    if (this.state.queued === 0 && !this.state.syncing) {
+      try {
+        return await this.source.writeMany(changes)
+      } catch (error) {
+        if (!(error instanceof StorageUnavailableError)) {
+          await this.apply(keys.map((key, i) => ({ key, value: previous[i] })))
+          throw error
+        }
+      }
+    }
+    await this.writes.set(this.ids.generate(), changes)
+    this.state.queued++
+    this.retryLater()
   }
 
-  // ------------------------------------------------------------------ start
+  /** The first replica once it holds a full copy; until then, a sync is attempted first and the source is the fallback. */
+  private async reader(): Promise<BaseKVStorage> {
+    if (!this.state.synced) await this.sync()
+    return this.state.synced ? this.replicas[0]! : this.source
+  }
 
   private async open(): Promise<void> {
-    const meta = await this.replicas.readMeta()
-    await this.outbox.load()
-    await this.replicas.seed()
-    if (meta) this.snapshotTaken()
-
-    this.offs.push(
+    this.state.synced = (await this.queue.get('synced')) === true
+    for await (const _ of this.writes.keys()) this.state.queued++
+    await this.seed()
+    this.state.offs.push(
       this.source.on('change', (event) => {
-        if (event.origin === 'remote') this.enqueue(() => this.applyRemote(event))
-        if (this.outbox.size > 0) void this.outbox.replay() // the source is reachable again
+        if (event.origin !== 'remote') return
+        const changes = event.changes.filter(({ key }) => this.range.contains(key))
+        this.state.applying = this.state.applying.then(() => this.applyRemote(changes)).catch((error) => console.error(error))
       }),
-      this.source.on('resync', () => this.enqueue(() => this.fullSync())),
+      this.source.on('resync', () => void this.sync()),
     )
-
-    const source = this.subscribable
-    if (source) {
-      source.lastEventId = meta?.lastEventId
-      // Without a resume point, catch up on the stream first so no change after the snapshot is missed.
-      const subscribed = source.subscribe(this.range.prefix)
-      if (!meta?.lastEventId) void subscribed.then(() => this.enqueue(() => this.fullSync())).catch((error) => console.error(error))
-      else void subscribed.catch((error) => console.error(error))
+    if (KVStorageComposite.isSubscribable(this.source)) {
+      void this.source
+        .subscribe(this.range.prefix)
+        .then(() => this.sync())
+        .catch((error) => console.error(error))
     } else {
-      this.enqueue(() => this.fullSync())
+      void this.sync()
     }
   }
 
-  /** Queues a sync task; failures are logged (an unreachable source is retried elsewhere). */
-  private enqueue(task: () => Promise<void>): void {
-    this.applying.run(task).catch((error) => {
+  private async runSync(): Promise<void> {
+    await this.start()
+    try {
+      await this.replay()
+      let batch: KVChange[] = []
+      for await (const change of this.diff()) if (batch.push(change) >= 500) await this.apply(batch.splice(0), true)
+      await this.apply(batch, true)
+      await this.replay() // writes made during the diff: send them, and re-apply them over it
+      await this.queue.set('synced', true)
+      this.state.synced = true
+    } catch (error) {
       if (!(error instanceof StorageUnavailableError)) console.error(error)
-    })
+      this.retryLater()
+    }
   }
 
-  private snapshotTaken(): void {
-    this.state.hasSnapshot = true
-    this.state.snapshotTaken()
+  /** Sends queued writes to the source in order, and re-applies each locally (a remote change may have overwritten it meanwhile). */
+  private async replay(): Promise<void> {
+    for (;;) {
+      let next: [string, unknown] | undefined
+      for await (const entry of this.writes.entries({ limit: 1 })) next = entry
+      if (!next) return
+      const changes = next[1] as KVChange[]
+      await this.source.writeMany(changes)
+      await this.apply(changes)
+      await this.writes.delete(next[0])
+      this.state.queued--
+    }
   }
 
-  private get subscribable(): SubscribableSource | undefined {
-    const source = this.source as unknown as Partial<SubscribableSource>
-    return typeof source.subscribe === 'function' ? (source as SubscribableSource) : undefined
+  /** What the source has that differs from the last replica (both scanned in key order). */
+  private async *diff(): AsyncGenerator<KVChange> {
+    const options = { gte: this.range.gte, lt: this.range.lt, values: true }
+    const remote = this.source.scan(options)[Symbol.asyncIterator]()
+    const local = this.replicas.at(-1)!.scan(options)[Symbol.asyncIterator]()
+    let r = await remote.next()
+    let l = await local.next()
+    while (!r.done || !l.done) {
+      if (l.done || (!r.done && r.value[0] < l.value[0])) {
+        yield { key: r.value![0], value: r.value![1] }
+        r = await remote.next()
+      } else if (r.done || l.value[0] < r.value[0]) {
+        yield { key: l.value[0] }
+        l = await local.next()
+      } else {
+        if (!Json.equal(r.value[1], l.value[1])) yield { key: r.value[0], value: r.value[1] }
+        ;[r, l] = await Promise.all([remote.next(), local.next()])
+      }
+    }
+  }
+
+  /** Applies pushed changes that differ from what the replicas hold. */
+  private async applyRemote(changes: KVChange[]): Promise<void> {
+    const current = await this.replicas[0]!.readMany(changes.map(({ key }) => key))
+    await this.apply(
+      changes.filter(({ value }, i) => !Json.equal(value, current[i])),
+      true,
+    )
+  }
+
+  /** Writes changes to every replica; `remote` ones are also announced as `change` events. */
+  private async apply(changes: KVChange[], remote = false): Promise<void> {
+    if (changes.length === 0) return
+    await Promise.all(this.replicas.map((replica) => replica.writeMany(changes)))
+    if (remote) this.emit('change', { changes, origin: 'remote' })
+  }
+
+  /** Copies the last (persistent) replica into the faster ones. */
+  private async seed(): Promise<void> {
+    if (this.replicas.length < 2) return
+    let batch: KVChange[] = []
+    const flush = () => Promise.all(this.replicas.slice(0, -1).map((replica) => replica.writeMany(batch.splice(0))))
+    for await (const [key, value] of this.replicas.at(-1)!.scan({ gte: this.range.gte, lt: this.range.lt, values: true })) {
+      if (batch.push({ key, value }) >= 500) await flush()
+    }
+    await flush()
+  }
+
+  private retryLater(): void {
+    clearTimeout(this.state.retry)
+    this.state.retry = setTimeout(() => void this.sync(), this.retryMs)
+    ;(this.state.retry as { unref?: () => void }).unref?.()
   }
 
   private checkRange(keys: string[]): void {
     for (const key of keys) if (!this.range.contains(key)) throw new RangeError(`"${key}" is outside this composite's prefix "${this.range.prefix}"`)
   }
 
-  // ------------------------------------------------------------------ source → replicas
-
-  private async applyRemote(event: KVChangeEvent): Promise<void> {
-    const relevant = event.changes.filter(
-      ({ key }) => this.range.contains(key) && !SYNC_INTERNAL.contains(key) && !this.echoes.take(key) && !this.outbox.pending.has(key),
-    )
-    const changes = await this.replicas.unapplied(relevant)
-    await this.replicas.apply(changes)
-    if (event.id) await this.replicas.writeMeta({ lastEventId: event.id })
-    if (changes.length) this.emit('change', { changes, origin: 'remote', id: event.id })
+  /** The storage instance a node belongs to (child nodes are frozen views over it). */
+  private static rootOf(node: BaseKVStorage): BaseKVStorage {
+    while (Object.isFrozen(node)) node = Object.getPrototypeOf(node) as BaseKVStorage
+    return node
   }
 
-  /** Makes the replicas match the source (within the range). */
-  private async fullSync(): Promise<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 lastEventId = this.subscribable?.lastEventId
-      let batch: KVChange[] = []
-      const flush = async () => {
-        const changes = batch.filter((change) => !skip(change))
-        await this.replicas.apply(changes)
-        if (changes.length) this.emit('change', { changes, origin: 'remote' })
-        batch = []
-      }
-      for await (const change of this.replicas.diff(this.source)) if (batch.push(change) >= 500) await flush()
-      await flush()
-      await this.replicas.writeMeta({ lastEventId })
-      this.snapshotTaken()
-      this.syncRetry.succeeded()
-    } catch (error) {
-      if (!(error instanceof StorageUnavailableError)) throw error
-      this.syncRetry.schedule(() => this.enqueue(() => this.fullSync()))
-    } finally {
-      this.state.writtenDuringSync = undefined
-    }
-  }
-
-  // ------------------------------------------------------------------ writes → source
-
-  private async write(entry: OutboxEntry): Promise<void> {
-    const keys = KVOutbox.keysOf(entry)
-    this.checkRange(keys)
-    await this.start()
-    for (const key of keys) this.state.writtenDuringSync?.add(key)
-    if (this.outbox.size === 0) {
-      try {
-        await this.send(entry)
-        return this.replicas.apply(KVOutbox.changesOf(entry))
-      } catch (error) {
-        if (!(error instanceof StorageUnavailableError)) throw error
-      }
-    }
-    // Offline, or earlier writes are still queued: apply locally and queue for the source.
-    await this.outbox.push(entry)
-    await this.replicas.apply(KVOutbox.changesOf(entry))
-  }
-
-  /** Sends a write to the source, expecting its echo when the source pushes changes. */
-  private async send(entry: OutboxEntry): Promise<void> {
-    const keys = this.subscribable ? KVOutbox.keysOf(entry) : []
-    this.echoes.add(keys)
-    try {
-      await (entry.type === 'write' ? this.source.writeMany(entry.entries) : this.source.removeMany(entry.keys))
-    } catch (error) {
-      for (const key of keys) this.echoes.take(key)
-      throw error
-    }
+  private static isSubscribable(source: BaseKVStorage): source is BaseKVStorage & Subscribable {
+    return typeof (source as Partial<Subscribable>).subscribe === 'function'
   }
 }

src/kv-storage/kv-storage-http-server.ts

@@ -13,50 +13,33 @@ export interface KVStorageHttpServerOptions {
   auth: 'none' | ({ verifiers: SignatureVerifier[] } & RequestAuthOptions)
   /** Path prefix to strip, e.g. `/kv`. Default `''`. */
   basePath?: string
-  /** Recent change batches kept for clients resuming with `Last-Event-ID`. Default 10 000. */
-  historySize?: number
   /** Interval of SSE keep-alive comments. Default 15 000 ms. */
   keepAliveMs?: number
 }
 
-interface HistoryEntry {
-  seq: number
-  changes: KVChange[]
-}
-
-interface Subscriber {
-  /** Keys (with their subtrees) this subscriber receives; `''` = everything. */
-  ranges: KeyRange[]
-  send: (frame: string) => void
-}
-
-type RequestBody = { keys: string[]; entries: [string, unknown][]; paths: string[] } & ScanOptions
+type RequestBody = { keys: string[]; changes: KVChange[]; paths: string[] } & ScanOptions
 
 /**
  * Serves any KV storage over HTTP for `KVStorageHttp`. Framework-agnostic: `handle(Request)`
- * returns a `Response`. Every route is `POST` with an Extended JSON body:
+ * returns a `Response`. It executes exactly the operations clients send; it doesn't know about
+ * versions or conflicts. Every route is `POST` with an Extended JSON body:
  *
- * - `/readMany`, `/writeMany`, `/removeMany`, `/writeIfAbsent`, `/scan` (streams NDJSON).
- * - `/events` with `{ paths }`: Server-Sent Events of changes to those keys and their subtrees.
- *   Event ids are `<epoch>-<seq>` (epoch = this server instance). A client sending
- *   `Last-Event-ID` gets the matching changes it missed if they are still in the history,
- *   otherwise a `resync` event; either way a `ready` event ends the catch-up.
+ * - `/readMany { keys }`, `/write { changes }`, `/scan { gte, lt, limit, reverse, values }` (NDJSON).
+ * - `/events { paths }`: Server-Sent Events relaying each change to those keys (and their
+ *   subtrees) as it happens, including the subscriber's own writes. A `ready` event follows the
+ *   connection. Nothing is replayed after a reconnect; clients catch up by reading.
  *
  * With authentication, each request is checked first (401), then every key, scan range and
  * subscription path must be the caller's public key or under it (403).
  *
- * Only changes made through the wrapped storage instance are seen; several server processes over
- * one database each see only their own writes.
+ * Only changes made through the wrapped storage instance are relayed; several server processes
+ * over one database each relay only their own writes.
  */
 export class KVStorageHttpServer {
   private readonly auth: RequestAuth | undefined
   private readonly basePath: string
-  private readonly historySize: number
   private readonly keepAliveMs: number
-  private readonly epoch = Math.random().toString(36).slice(2, 10)
-  private seq = 0
-  private readonly history: HistoryEntry[] = []
-  private readonly subscribers = new Set<Subscriber>()
+  private readonly subscribers = new Set<{ ranges: KeyRange[]; send: (frame: string) => void }>()
 
   constructor(
     private readonly storage: BaseKVStorage,
@@ -64,9 +47,13 @@ export class KVStorageHttpServer {
   ) {
     this.auth = options.auth === 'none' ? undefined : new RequestAuth(options.auth.verifiers, options.auth)
     this.basePath = options.basePath ?? ''
-    this.historySize = options.historySize ?? 10_000
     this.keepAliveMs = options.keepAliveMs ?? 15_000
-    storage.on('change', ({ changes }) => this.publish(changes))
+    storage.on('change', ({ changes }) => {
+      for (const subscriber of this.subscribers) {
+        const matching = changes.filter(({ key }) => subscriber.ranges.some((range) => range.contains(key)))
+        if (matching.length) subscriber.send(`event: change\ndata: ${ExtendedJson.stringify(matching)}\n\n`)
+      }
+    })
   }
 
   async handle(request: Request): Promise<Response> {
@@ -81,32 +68,22 @@ export class KVStorageHttpServer {
           this.authorize(owner, body.keys)
           const values = await this.storage.readMany(body.keys)
           const missing = values.flatMap((v, i) => (v === undefined ? [i] : []))
-          return KVStorageHttpServer.json({ values: values.map((v) => v ?? null), missing })
+          return new Response(ExtendedJson.stringify({ values: values.map((v) => v ?? null), missing }), { headers: { 'content-type': 'application/json' } })
         }
-        case '/writeMany':
+        case '/write':
           this.authorize(
             owner,
-            body.entries.map(([key]) => key),
+            body.changes.map(({ key }) => key),
           )
-          await this.storage.writeMany(body.entries)
+          await this.storage.writeMany(body.changes)
           return new Response(null, { status: 204 })
-        case '/removeMany':
-          this.authorize(owner, body.keys)
-          await this.storage.removeMany(body.keys)
-          return new Response(null, { status: 204 })
-        case '/writeIfAbsent':
-          this.authorize(
-            owner,
-            body.entries.map(([key]) => key),
-          )
-          return KVStorageHttpServer.json(await this.storage.writeIfAbsent(body.entries))
         case '/scan':
-          this.authorizeRange(owner, body)
+          if (owner !== undefined && !new KeyRange(owner).covers(body.gte, body.lt)) throw new AuthError(403, 'Scan outside your keys')
           return this.scan(body)
         case '/events':
           if (!Array.isArray(body.paths) || body.paths.length === 0) return KVStorageHttpServer.error(400, 'Subscribe to at least one path')
           this.authorize(owner, body.paths)
-          return this.events(body.paths, request.headers.get('last-event-id'))
+          return this.events(body.paths)
         default:
           return KVStorageHttpServer.error(404, 'Not found')
       }
@@ -116,8 +93,6 @@ export class KVStorageHttpServer {
     }
   }
 
-  // ------------------------------------------------------------------ authorization
-
   /** Every key must be the owner's key or under it (no-op without authentication). */
   private authorize(owner: string | undefined, keys: string[]): void {
     if (owner === undefined) return
@@ -127,33 +102,15 @@ export class KVStorageHttpServer {
     }
   }
 
-  /** A scan must stay inside the owner's key and its subtree. */
-  private authorizeRange(owner: string | undefined, options: ScanOptions): void {
-    if (owner !== undefined && !new KeyRange(owner).covers(options.gte, options.lt)) throw new AuthError(403, 'Scan outside your keys')
-  }
-
-  // ------------------------------------------------------------------ change stream
-
-  private publish(changes: KVChange[]): void {
-    const entry = { seq: ++this.seq, changes }
-    this.history.push(entry)
-    if (this.history.length > this.historySize) this.history.splice(0, this.history.length - this.historySize)
-    for (const subscriber of this.subscribers) {
-      const frame = this.changeFrame(entry, subscriber.ranges)
-      if (frame) subscriber.send(frame)
-    }
-  }
-
-  private events(paths: string[], lastEventId: string | null): Response {
+  private events(paths: string[]): Response {
     const encoder = new TextEncoder()
-    let subscriber: Subscriber | undefined
+    let subscriber: { ranges: KeyRange[]; send: (frame: string) => void } | undefined
     let keepAlive: ReturnType<typeof setInterval> | undefined
     const stream = new ReadableStream<Uint8Array>({
       start: (controller) => {
-        const ranges = paths.map((path) => new KeyRange(path))
-        subscriber = { ranges, send: (frame) => controller.enqueue(encoder.encode(frame)) }
-        subscriber.send(this.catchUpFrames(ranges, lastEventId))
+        subscriber = { ranges: paths.map((path) => new KeyRange(path)), send: (frame) => controller.enqueue(encoder.encode(frame)) }
         this.subscribers.add(subscriber)
+        subscriber.send('event: ready\ndata: null\n\n')
         keepAlive = setInterval(() => subscriber!.send(': keep-alive\n\n'), this.keepAliveMs)
         ;(keepAlive as { unref?: () => void }).unref?.()
       },
@@ -165,38 +122,6 @@ export class KVStorageHttpServer {
     return new Response(stream, { headers: { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' } })
   }
 
-  /**
-   * What a (re)connecting client gets before live changes: the matching changes it missed, or
-   * `resync` when they can't be replayed; then always `ready` (caught up as of that id).
-   */
-  private catchUpFrames(ranges: KeyRange[], lastEventId: string | null): string {
-    const ready = this.frame('ready', this.seq, null)
-    if (lastEventId === null) return ready
-    const [epoch, seqText] = lastEventId.split('-')
-    const seq = Number(seqText)
-    const oldest = this.history[0]?.seq ?? this.seq + 1
-    // Resumable only if it's our epoch and nothing after `seq` has been dropped from the history.
-    if (epoch !== this.epoch || !Number.isInteger(seq) || seq > this.seq || seq < oldest - 1) return this.frame('resync', this.seq, null) + ready
-    return (
-      this.history
-        .filter((entry) => entry.seq > seq)
-        .map((entry) => this.changeFrame(entry, ranges))
-        .join('') + ready
-    )
-  }
-
-  /** A `change` frame with only the changes inside `ranges`, or `''` if none match. */
-  private changeFrame(entry: HistoryEntry, ranges: KeyRange[]): string {
-    const changes = entry.changes.filter(({ key }) => ranges.some((range) => range.contains(key)))
-    return changes.length ? this.frame('change', entry.seq, changes) : ''
-  }
-
-  private frame(event: string, seq: number, data: unknown): string {
-    return `id: ${this.epoch}-${seq}\nevent: ${event}\ndata: ${ExtendedJson.stringify(data)}\n\n`
-  }
-
-  // ------------------------------------------------------------------ scan
-
   private scan(options: ScanOptions): Response {
     const iterator = this.storage.scan(options)[Symbol.asyncIterator]()
     const encoder = new TextEncoder()
@@ -221,10 +146,6 @@ export class KVStorageHttpServer {
     return new Response(stream, { headers: { 'content-type': 'application/x-ndjson' } })
   }
 
-  private static json(body: unknown): Response {
-    return new Response(ExtendedJson.stringify(body), { headers: { 'content-type': 'application/json' } })
-  }
-
   private static error(status: number, message: string): Response {
     return Response.json({ error: message }, { status })
   }

src/kv-storage/kv-storage-http.ts

@@ -3,7 +3,6 @@ import { ExtendedJson } from '../utils/extended-json.js'
 import { RequestAuth, type RequestSigner } from '../utils/request-auth.js'
 import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js'
 import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js'
-import { SerialQueue } from '../utils/serial-queue.js'
 import { SSESession } from '../utils/sse-session/sse-session.js'
 import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js'
 
@@ -27,10 +26,10 @@ export interface KVStorageHttp<S extends Spec = AnySpec> {
 /**
  * KV storage client for a `KVStorageHttpServer`. Same API as any other KV storage.
  *
- * `subscribe(path)` receives changes under a key (pushed over SSE) as `change` events with
- * `origin: 'remote'`; `resync` is emitted when the server can't resume from `lastEventId`. With a
- * `signer`, every request is signed, and an authenticated server only allows keys under the
- * signer's public key. Values travel as Extended JSON (`bigint` and `Uint8Array` survive).
+ * `subscribe(path)` relays every change the server pushes for a key and its subtree (including
+ * this client's own writes) as `change` events with `origin: 'remote'`. When a dropped stream reconnects, `resync` is emitted: changes may have been
+ * missed, so re-read what you need. With a `signer`, every request is signed. Values travel as
+ * Extended JSON (`bigint` and `Uint8Array` survive).
  */
 export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   private readonly url: string
@@ -38,10 +37,8 @@ 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: { paths: Set<string>; session?: SSESession; lastEventId?: string } = { paths: new Set() }
+  private readonly state: { paths: Set<string>; session?: SSESession; updating: Promise<unknown> } = { paths: new Set(), updating: Promise.resolve() }
 
   constructor(options: KVStorageHttpOptions) {
     super()
@@ -52,23 +49,9 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
     this.onSubscriptionError = options.onSubscriptionError ?? (() => {})
   }
 
-  /** Position in the server's change stream; sent as `Last-Event-ID` when (re)connecting. */
-  get lastEventId(): string | undefined {
-    return this.state.lastEventId
-  }
-
-  set lastEventId(id: string | undefined) {
-    this.state.lastEventId = id
-  }
-
-  /** Keys currently subscribed to. */
-  get subscriptions(): ReadonlySet<string> {
-    return this.state.paths
-  }
-
   async readMany(keys: string[]): Promise<unknown[]> {
     if (keys.length === 0) return []
-    const { values, missing } = (await this.post('readMany', { keys })) as { values: unknown[]; missing: number[] }
+    const { values, missing } = ExtendedJson.parse(await (await this.request('readMany', { keys })).text()) as { values: unknown[]; missing: number[] }
     for (const i of missing) values[i] = undefined
     return values
   }
@@ -96,15 +79,15 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
   }
 
   /**
-   * Receives changes to `path` (default: this node's key) and everything under it. Resolves once
-   * the stream is connected and caught up (missed changes replayed, or `resync` emitted). While
-   * the server is unreachable this stays pending, so don't await it on a path that must work offline.
+   * Relays changes to `path` (default: this node's key) and everything under it. Resolves once
+   * the stream is connected. While the server is unreachable this stays pending, so don't await it
+   * on a path that must work offline. A 401/403 rejects instead of retrying.
    */
   subscribe(path: string = this.key): Promise<void> {
     return this.updateSubscriptions(() => this.state.paths.add(path))
   }
 
-  /** Stops receiving changes to `path` (default: this node's key). */
+  /** Stops relaying changes to `path` (default: this node's key). */
   unsubscribe(path: string = this.key): Promise<void> {
     return this.updateSubscriptions(() => this.state.paths.delete(path))
   }
@@ -114,21 +97,11 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
     return this.updateSubscriptions(() => this.state.paths.clear())
   }
 
-  protected async applyWrite(entries: [string, unknown][]): Promise<void> {
-    await this.post('writeMany', { entries })
+  protected async write(changes: KVChange[]): Promise<void> {
+    await this.request('write', { changes })
   }
 
-  protected async applyRemove(keys: string[]): Promise<void> {
-    await this.post('removeMany', { keys })
-  }
-
-  protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
-    return (await this.post('writeIfAbsent', { entries })) as boolean[]
-  }
-
-  // ------------------------------------------------------------------ subscriptions
-
-  /** Applies a subscription change, then reconnects the stream with the new path list. */
+  /** Applies a subscription change (one at a time), then reconnects with the new path list. */
   private updateSubscriptions(change: () => unknown): Promise<void> {
     const run = async () => {
       const before = JSON.stringify([...this.state.paths])
@@ -138,7 +111,9 @@ 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])
     }
-    return this.updates.run(run)
+    const result = this.state.updating.then(run)
+    this.state.updating = result.catch(() => {})
+    return result
   }
 
   private async connect(paths: string[]): Promise<void> {
@@ -153,39 +128,23 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
       onError: this.onSubscriptionError,
       retry: KVStorageHttp.retryUnlessDenied(),
       // Signed on every (re)connect, so the timestamp is always fresh.
-      onRequest: async (request) => ({
-        ...request,
-        headers: {
-          ...(request.headers as Record<string, string>),
-          ...(this.state.lastEventId ? { 'Last-Event-ID': this.state.lastEventId } : {}),
-          ...(await this.authHeaders(url, body)),
-        },
-      }),
+      onRequest: async (request) => ({ ...request, headers: { ...(request.headers as Record<string, string>), ...(await this.authHeaders(url, body)) } }),
     })
+    let readies = 0
     session.on('message', (message) => {
-      if (message.id) this.state.lastEventId = message.id
-      if (message.event === 'change') {
-        this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote', id: message.id })
-      } else if (message.event === 'resync') {
-        this.emit('resync', { id: message.id })
-      }
+      if (message.event === 'change') this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote' })
+      // Every `ready` after the first is a reconnect: changes may have been missed meanwhile.
+      if (message.event === 'ready' && readies++ > 0) this.emit('resync', undefined)
     })
     this.state.session = session
-    const caughtUp = session.waitFor('message', (message) => message.event === 'ready')
+    const ready = session.waitFor('message', (message) => message.event === 'ready')
     try {
       await session.connect()
     } catch (error) {
       this.state.session = undefined
       throw (error as Error).cause instanceof AuthError ? (error as Error).cause : error
     }
-    await caughtUp
-  }
-
-  // ------------------------------------------------------------------ requests
-
-  private async post(path: string, body: unknown): Promise<unknown> {
-    const response = await this.request(path, body)
-    return response.status === 204 ? undefined : ExtendedJson.parse(await response.text())
+    await ready
   }
 
   private async request(path: string, payload: unknown): Promise<Response> {
@@ -207,6 +166,10 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
     return response
   }
 
+  private async authHeaders(url: string, body: string): Promise<Record<string, string>> {
+    return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {}
+  }
+
   /** Retries connecting with backoff, except when the server denies access (401/403): that won't fix itself. */
   private static retryUnlessDenied(): { run<T>(fn: () => Promise<T>): Promise<T> } {
     const backoff = new ExponentialBackoff({ baseDelay: 1000, maxDelay: 10_000, maxAttempts: 0, growthRate: 1.3, jitter: 0.3 })
@@ -218,8 +181,4 @@ export class KVStorageHttp<S extends Spec = AnySpec> extends BaseKVStorage<S> {
         }),
     }
   }
-
-  private async authHeaders(url: string, body: string): Promise<Record<string, string>> {
-    return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {}
-  }
 }

src/kv-storage/kv-storage-in-memory.ts

@@ -1,7 +1,7 @@
 import { Json } from '../utils/json.js'
 import { SortedMap } from '../utils/sorted-map.js'
 import type { AnySpec, Spec } from '../utils/schema.js'
-import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js'
+import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js'
 
 /** In-memory KV storage. Values are deep-copied on write and read. */
 export class KVStorageInMemory<S extends Spec = AnySpec> extends BaseKVStorage<S> {
@@ -11,18 +11,6 @@ export class KVStorageInMemory<S extends Spec = AnySpec> extends BaseKVStorage<S
     return keys.map((k) => Json.clone(this.data.get(k)))
   }
 
-  protected async applyWrite(entries: [string, unknown][]): Promise<void> {
-    for (const [k, v] of entries) this.data.set(k, Json.clone(v))
-  }
-
-  protected async applyRemove(keys: string[]): Promise<void> {
-    for (const k of keys) this.data.delete(k)
-  }
-
-  protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
-    return entries.map(([k, v]) => !this.data.has(k) && (this.data.set(k, Json.clone(v)), true))
-  }
-
   async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> {
     let remaining = options.limit ?? Infinity
     if (remaining <= 0) return
@@ -31,4 +19,11 @@ export class KVStorageInMemory<S extends Spec = AnySpec> extends BaseKVStorage<S
       if (--remaining <= 0) return
     }
   }
+
+  protected async write(changes: KVChange[]): Promise<void> {
+    for (const { key, value } of changes) {
+      if (value === undefined) this.data.delete(key)
+      else this.data.set(key, Json.clone(value))
+    }
+  }
 }

src/kv-storage/kv-storage-indexeddb.ts

@@ -1,5 +1,5 @@
 import type { AnySpec, Spec } from '../utils/schema.js'
-import { BaseKVStorage, type ScanOptions } from './base-kv-storage.js'
+import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js'
 
 export interface KVStorageIndexedDBOptions {
   /** Database name. */
@@ -56,32 +56,15 @@ export class KVStorageIndexedDB<S extends Spec = AnySpec> extends BaseKVStorage<
     ;(await this.db).close()
   }
 
-  protected async applyWrite(entries: [string, unknown][]): Promise<void> {
+  protected async write(changes: KVChange[]): Promise<void> {
     const store = await this.store('readwrite')
-    for (const [key, value] of entries) store.put(value, key)
+    for (const { key, value } of changes) {
+      if (value === undefined) store.delete(key)
+      else store.put(value, key)
+    }
     await KVStorageIndexedDB.committed(store.transaction)
   }
 
-  protected async applyRemove(keys: string[]): Promise<void> {
-    const store = await this.store('readwrite')
-    for (const key of keys) store.delete(key)
-    await KVStorageIndexedDB.committed(store.transaction)
-  }
-
-  protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise<boolean[]> {
-    // One readwrite transaction: checks and writes can't interleave with other writers.
-    const store = await this.store('readwrite')
-    const written = await Promise.all(
-      entries.map(async ([key, value]) => {
-        if ((await KVStorageIndexedDB.done(store.getKey(key))) !== undefined) return false
-        store.add(value, key)
-        return true
-      }),
-    )
-    await KVStorageIndexedDB.committed(store.transaction)
-    return written
-  }
-
   /** Up to `count` entries after `last` (exclusive), in scan order, within one transaction. */
   private async readBatch(options: ScanOptions, last: string | undefined, count: number): Promise<[string, unknown][]> {
     const reverse = options.reverse ?? false

src/utils/errors.ts

@@ -1,15 +1,3 @@
-/** An optimistic-concurrency check failed (stale `expectedVersion` or a lost commit race). */
-export class ConcurrencyError extends Error {
-  constructor(
-    readonly stream: string,
-    readonly expectedVersion: number | 'any',
-    readonly actualVersion: number,
-  ) {
-    super(`Concurrency conflict on "${stream}": expected version ${expectedVersion}, actual ${actualVersion}`)
-    this.name = 'ConcurrencyError'
-  }
-}
-
 /** A key argument was passed but is `undefined`/`null`, empty, or not a string/number. */
 export class InvalidKeyError extends TypeError {
   constructor(message: string) {

src/utils/extended-json.ts

@@ -18,6 +18,19 @@ export class ExtendedJson {
     return JSON.stringify(value, ExtendedJson.replacer)
   }
 
+  /** Like `stringify`, with object keys sorted, so equal values always give the same text (for hashing). */
+  static canonical(value: unknown): string {
+    return JSON.stringify(value, (key, v) => {
+      const replaced = ExtendedJson.replacer(key, v)
+      if (typeof replaced !== 'object' || replaced === null || Array.isArray(replaced)) return replaced
+      return Object.fromEntries(
+        Object.keys(replaced)
+          .sort()
+          .map((k) => [k, (replaced as Record<string, unknown>)[k]]),
+      )
+    })
+  }
+
   static parse(text: string): unknown {
     return JSON.parse(text, ExtendedJson.reviver)
   }

src/utils/key-counter.ts

@@ -1,36 +0,0 @@
-/** 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/retry-timer.ts

@@ -1,36 +0,0 @@
-import { ExponentialBackoff } from './sse-session/exponential-backoff.js'
-
-export interface RetryTimerOptions {
-  baseDelay?: number
-  maxDelay?: number
-}
-
-/** Schedules one pending retry at a time, with exponential backoff that resets on success. */
-export class RetryTimer {
-  private readonly options: { baseDelay: number; maxDelay: number }
-  private timer: ReturnType<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

@@ -1,17 +0,0 @@
-/** Runs tasks one at a time, in the order given; independently per key when a key is passed. */
-export class SerialQueue {
-  private readonly tails = new Map<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)
-  }
-}