From df7de27e20b1e0fb836b69d6c1fddf73385873e4 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 08:47:18 +0000 Subject: [PATCH] Composite: don't undo writes made while a full sync is scanning A write made during a full sync could be scanned in the replica but not in the (read-ahead) source scan, so the diff removed it, and echo tracking then consumed the server's echo. Keys written during a sync are now left out of the diff. Deterministic regression test with a read-ahead source. Also exclude -0 from the Extended JSON property test (JSON has no -0). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017Qxbkp4tA7G5SCWjzWuxmE --- DESIGN.md | 2 +- src/kv-storage/kv-storage-composite.ts | 16 +++++++-- test/kv-storage/kv-storage-composite.test.ts | 37 +++++++++++++++++++- test/utils/extended-json.test.ts | 2 +- 4 files changed, 52 insertions(+), 5 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index bca1a8c..9725b38 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -69,7 +69,7 @@ emit a `change` event after each write, so every store emits change events. | `applyRemove(keys)` → public `removeMany` | | | `applyWriteIfAbsent(entries)` → public `writeIfAbsent` | `true` = written; atomic per key (needed by events) | -**Values** are JSON plus `bigint` and `Uint8Array`. Anything crossing the network is encoded as +**Values** are JSON plus `bigint` and `Uint8Array` (as in JSON, `-0` becomes `0` and `undefined` properties are dropped). Anything crossing the network is encoded as Extended JSON (``, ``); IndexedDB stores them natively. **Events** (`EventEmitter`, shared by all nodes of a storage): `change` → diff --git a/src/kv-storage/kv-storage-composite.ts b/src/kv-storage/kv-storage-composite.ts index a61ab3d..3b7175e 100644 --- a/src/kv-storage/kv-storage-composite.ts +++ b/src/kv-storage/kv-storage-composite.ts @@ -85,6 +85,12 @@ export class KVStorageComposite extends BaseKVStorage< * changed during the scan. */ applying: Promise + /** + * Keys written through the composite while a full sync runs. The diff may have scanned the + * source before such a write and the replica after it; these keys are left out of the diff + * (the write already reached the source and the replicas). + */ + writtenDuringSync?: Set replaying?: Promise outboxSeq: number retryTimer?: ReturnType @@ -162,7 +168,9 @@ export class KVStorageComposite extends BaseKVStorage< protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { await this.start() const written = await this.source.writeIfAbsent(entries) - await this.applyToReplicas(entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value }))) + const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })) + for (const { key } of changes) this.state.writtenDuringSync?.add(key) + await this.applyToReplicas(changes) return written } @@ -243,11 +251,12 @@ export class KVStorageComposite extends BaseKVStorage< private async fullSync(): Promise { // The diff re-reads the source, so outstanding echoes no longer matter (and may never arrive). this.echoes.clear() + const writtenDuringSync = (this.state.writtenDuringSync = new Set()) 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)) + const applied = changes.filter(({ key }) => !this.pending.has(key) && !writtenDuringSync.has(key)) await this.applyToReplicas(applied) if (applied.length) this.emit('change', { changes: applied, origin: 'remote' }) changes = [] @@ -260,6 +269,8 @@ export class KVStorageComposite extends BaseKVStorage< } catch (error) { if (!(error instanceof StorageUnavailableError)) throw error this.scheduleRetry(() => this.enqueueApply(() => this.fullSync())) + } finally { + this.state.writtenDuringSync = undefined } } @@ -296,6 +307,7 @@ export class KVStorageComposite extends BaseKVStorage< // ------------------------------------------------------------------ local writes → source private async writeThrough(entry: OutboxEntry): Promise { + for (const key of KVStorageComposite.keysOf(entry)) this.state.writtenDuringSync?.add(key) if (this.queued === 0) { try { await this.send(entry) diff --git a/test/kv-storage/kv-storage-composite.test.ts b/test/kv-storage/kv-storage-composite.test.ts index c6bbd22..aef18cb 100644 --- a/test/kv-storage/kv-storage-composite.test.ts +++ b/test/kv-storage/kv-storage-composite.test.ts @@ -3,7 +3,7 @@ import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js' import { DocumentStorageKV } from '../../src/document-storage/document-storage-kv.js' -import { collect } from '../helpers.js' +import { collect, KVStorageCounting } from '../helpers.js' import { runKVStorageConformance } from './kv-storage.conformance.js' import { TestNetwork } from './sync-helpers.js' @@ -141,4 +141,39 @@ describe('KVStorageComposite sync', () => { network.online = false expect((await collect(users.find({ age: { $gt: 35 } }))).map((r) => r.key)).toEqual(['2']) }) + + it('does not undo writes made while a full sync is scanning', async () => { + // A source whose scan reads ahead (like the HTTP stream) and pauses after its first entry. + let release!: () => void + let reached!: () => void + const gate = new Promise((r) => (release = r)) + const paused = new Promise((r) => (reached = r)) + const inner = new KVStorageInMemory() + await inner.writeMany([ + ['a', 1], + ['c', 3], + ]) + const source = new KVStorageCounting(inner) + const scan = source.scan.bind(source) + source.scan = async function* (options) { + let first = true + for (const entry of await collect(scan(options))) { + yield entry + if (first && options?.values) ((first = false), reached(), await gate) + } + } + const persistent = new KVStorageInMemory() + await persistent.writeMany([['a0', 'stale']]) // a replica copy from before, without sync metadata + const db = new KVStorageComposite({ source, replicas: [persistent] }) + void db.start() + await paused // the diff has passed 'b' in the source but not in the replica + await db.writeMany([['b', 2]]) + release() + await db.synced() + expect(await collect(db.scan({ values: true }))).toEqual([ + ['a', 1], + ['b', 2], + ['c', 3], + ]) + }) }) diff --git a/test/utils/extended-json.test.ts b/test/utils/extended-json.test.ts index c4dd21f..e85dc28 100644 --- a/test/utils/extended-json.test.ts +++ b/test/utils/extended-json.test.ts @@ -16,7 +16,7 @@ describe('ExtendedJson', () => { fc.bigInt(), fc.uint8Array(), fc.string(), - fc.double({ noNaN: true, noDefaultInfinity: true }), + fc.double({ noNaN: true, noDefaultInfinity: true }).map((x) => (x === 0 ? 0 : x)) /* JSON has no -0 */, fc.boolean(), fc.constant(null), fc.array(tie('value')),