diff --git a/AGENTS.md b/AGENTS.md index 56f5d5f..f4bc5d5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -193,6 +193,14 @@ Preserve these unless the task explicitly changes them. - Unmounting the last subscriber clears the shared autosave timer and unregisters its sync state. Pending debounced edits are not automatically flushed on `stop()`. +- Listener snapshots update state (`getState()`/`getHandle()`) at once, but + the subscriber fan-out is batched per task (`src/core/snapshot-batcher.ts`): + the first snapshot of a task notifies synchronously, and later snapshots in + the same task fold into one notify on the next task. The batching window is + shared by every subscription on a store, so one snapshot each across many + document subscriptions folds the same way. A burst of server + writes otherwise drives one synchronous React commit per snapshot and throws + React #185 (maximum update depth). Local writes still notify synchronously. - Undo actions are client-local. Grouped undo actions undo newest to oldest and redo oldest to newest. - Firestore updates use flattened diffs for `updateDoc`; full document @@ -230,6 +238,7 @@ Add or update focused tests near the behavior being changed: - Diff behavior: `src/utils/diff.test.ts`. - Structural sharing and dev-mode freezing (identity across edits, snapshots, and rebases): `src/__tests__/structural-sharing.test.ts`. +- Snapshot burst batching: `src/__tests__/snapshot-batching.test.ts`. - Undo behavior: `src/utils/undo.test.ts`. - Store/global sync behavior: `src/core/store.test.ts`. - React hook surface — selectors, shared subscriptions, `queryConstraints` diff --git a/docs/architecture.md b/docs/architecture.md index de091e7..d559fa0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -258,6 +258,11 @@ Important details: - Subscription state and handles are both cached until state changes (`getState()` and `getHandle()` return identity-stable references, rebuilt only on `notify()`), so React sees stable snapshots between commits. +- Listener snapshots publish state at once but batch the subscriber fan-out per + task: the first snapshot of a task notifies, and later snapshots in the same + task fold into one notify on the next task. The window is shared per store, + across subscriptions. This keeps a burst of server + writes from driving 50+ synchronous React commits (React #185). ## Document Subscriptions diff --git a/src/__tests__/snapshot-batching.test.ts b/src/__tests__/snapshot-batching.test.ts new file mode 100644 index 0000000..dfcd829 --- /dev/null +++ b/src/__tests__/snapshot-batching.test.ts @@ -0,0 +1,353 @@ +/** + * Snapshot batching: a burst of listener snapshots in one task updates state + * at once but notifies subscribers at most twice (the first snapshot, then one + * trailing notify on the next task). Without it, each snapshot drove one + * synchronous React commit, and 50+ of them in a row threw React #185 + * ("Maximum update depth exceeded"). + */ +import { vi, describe, it, expect, beforeEach } from 'vitest' + +vi.mock('firebase/firestore', async () => { + const actual = + await vi.importActual( + 'firebase/firestore' + ) + const { buildFirestoreMock } = await import('./test-harness') + return buildFirestoreMock(actual as unknown as Record) +}) + +import { createHarness, type Harness } from './test-harness' +import { createCollectionSubscription } from '../core/collection' +import { createDocumentSubscription } from '../core/document' +import { defineCollection, defineDocument } from '../registry/schema' +import { createStore, type FirestateStore } from '../core/store' + +interface Space { + id?: string + processed?: boolean +} + +interface Project { + revision?: number +} + +const spaces = defineCollection({ path: 'spaces' }) +const projectDef = defineDocument({ collection: 'projects', id: 'p1' }) + +const nextTask = () => new Promise((resolve) => setTimeout(resolve, 0)) + +const BURST_SIZE = 100 + +const spacesProcessedUpTo = (processedCount: number) => + Object.fromEntries( + Array.from({ length: BURST_SIZE }, (_, index) => [ + `space${index}`, + { processed: index < processedCount }, + ]) + ) + +describe('snapshot batching', () => { + let store: FirestateStore + let h: Harness + + beforeEach(() => { + vi.clearAllMocks() + store = createStore({ firestore: {} as never, autosave: 0 }) + h = createHarness() + }) + + const makeColl = async () => { + const sub = createCollectionSubscription({ + store, + definition: spaces, + collectionPath: 'spaces', + }) + sub.load() + h.fireCollectionSnapshot(spacesProcessedUpTo(0)) + await nextTask() + return sub + } + + const makeDoc = async () => { + const sub = createDocumentSubscription({ + store, + definition: projectDef, + docId: 'p1', + collectionPath: 'projects', + }) + sub.load() + h.fireDocSnapshot({ revision: 0 }) + await nextTask() + return sub + } + + describe('collection', () => { + it('notifies a single snapshot synchronously', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + h.fireCollectionSnapshot(spacesProcessedUpTo(1)) + + expect(subscriber).toHaveBeenCalledTimes(1) + await nextTask() + expect(subscriber).toHaveBeenCalledTimes(1) + }) + + it('folds a burst of snapshots into one trailing notify', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + for (let count = 1; count <= BURST_SIZE; count++) { + h.fireCollectionSnapshot(spacesProcessedUpTo(count)) + } + + expect(subscriber).toHaveBeenCalledTimes(1) + expect(sub.getState().data.space99?.processed).toBe(true) + + await nextTask() + expect(subscriber).toHaveBeenCalledTimes(2) + expect(subscriber).toHaveBeenLastCalledWith(sub.getState()) + }) + + it('folds a burst delivered through a microtask chain', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + for (let count = 1; count <= BURST_SIZE; count++) { + await Promise.resolve() + h.fireCollectionSnapshot(spacesProcessedUpTo(count)) + } + await nextTask() + + expect(subscriber).toHaveBeenCalledTimes(2) + expect(sub.getState().data.space99?.processed).toBe(true) + }) + + it('folds a backlog of snapshots queued as separate tasks', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + for (let count = 1; count <= BURST_SIZE; count++) { + setTimeout(() => { + h.fireCollectionSnapshot(spacesProcessedUpTo(count)) + }, 0) + } + await nextTask() + await nextTask() + + expect(subscriber).toHaveBeenCalledTimes(2) + expect(subscriber).toHaveBeenLastCalledWith(sub.getState()) + expect(sub.getState().data.space99?.processed).toBe(true) + }) + + // React #185 is thrown from inside the first synchronous notify; the + // rest of the burst must still be batched, not notified one by one. + it('keeps batching when the first notify throws', async () => { + const sub = await makeColl() + const thrown = new Error('subscriber failed') + const subscriber = vi.fn(() => { + throw thrown + }) + sub.subscribe(subscriber) + + for (let count = 1; count <= BURST_SIZE; count++) { + try { + h.fireCollectionSnapshot(spacesProcessedUpTo(count)) + } catch (error) { + expect(error).toBe(thrown) + } + } + expect(subscriber).toHaveBeenCalledTimes(1) + + const onUncaught = () => {} + process.on('uncaughtException', onUncaught) + try { + await nextTask() + await nextTask() + } finally { + process.off('uncaughtException', onUncaught) + } + expect(subscriber).toHaveBeenCalledTimes(2) + }) + + it('drops the trailing notify after stop()', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + h.fireCollectionSnapshot(spacesProcessedUpTo(1)) + h.fireCollectionSnapshot(spacesProcessedUpTo(2)) + sub.stop() + await nextTask() + + expect(subscriber).toHaveBeenCalledTimes(1) + }) + + it('keeps local updates synchronous during a burst', async () => { + const sub = await makeColl() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + h.fireCollectionSnapshot(spacesProcessedUpTo(1)) + h.fireCollectionSnapshot(spacesProcessedUpTo(2)) + sub.getHandle().update({ space50: { processed: true } }) + + expect(subscriber).toHaveBeenCalledTimes(2) + expect(subscriber).toHaveBeenLastCalledWith(sub.getState()) + expect(sub.getState().data.space1?.processed).toBe(true) + expect(sub.getState().data.space50?.processed).toBe(true) + + await nextTask() + expect(subscriber).toHaveBeenCalledTimes(2) + }) + }) + + describe('document', () => { + it('folds a burst of snapshots into one trailing notify', async () => { + const sub = await makeDoc() + const subscriber = vi.fn() + sub.subscribe(subscriber) + + for (let revision = 1; revision <= BURST_SIZE; revision++) { + h.fireDocSnapshot({ revision }) + } + + expect(subscriber).toHaveBeenCalledTimes(1) + expect(sub.getState().data?.revision).toBe(BURST_SIZE) + + await nextTask() + expect(subscriber).toHaveBeenCalledTimes(2) + expect(subscriber).toHaveBeenLastCalledWith(sub.getState()) + }) + }) + + describe('across subscriptions', () => { + const docSnap = (data: object) => ({ + exists: () => true, + data: () => data, + metadata: { fromCache: false, hasPendingWrites: false }, + }) + + // One snapshot per document subscription commits as often as a burst + // on one subscription, so the batching window is shared per store. + it('folds one snapshot each across many document subscriptions', async () => { + const subs = Array.from({ length: BURST_SIZE }, (_, index) => + createDocumentSubscription({ + store, + definition: projectDef, + docId: `p${index}`, + collectionPath: 'projects', + }) + ) + subs.forEach((sub) => sub.load()) + const listeners = h.listeners() + listeners.forEach((l) => l.onNext(docSnap({ revision: 0 }))) + await nextTask() + + const subscribers = subs.map((sub) => { + const subscriber = vi.fn() + sub.subscribe(subscriber) + return subscriber + }) + const totalCalls = () => + subscribers.reduce((n, fn) => n + fn.mock.calls.length, 0) + + for (const listener of listeners) { + await Promise.resolve() + listener.onNext(docSnap({ revision: 1 })) + } + + expect(totalCalls()).toBe(1) + expect( + subs.every((sub) => sub.getState().data?.revision === 1) + ).toBe(true) + + await nextTask() + subscribers.forEach((subscriber, index) => { + expect(subscriber).toHaveBeenCalledTimes(1) + expect(subscriber).toHaveBeenLastCalledWith( + subs[index]!.getState() + ) + }) + }) + + // React throws #185 from inside the subscriber callback, so one + // subscription's throw must not strand the rest of the shared flush. + it('notifies every subscription when one subscriber throws', async () => { + const a = await makeDoc() + const b = createDocumentSubscription({ + store, + definition: projectDef, + docId: 'p2', + collectionPath: 'projects', + }) + b.load() + h.fireDocSnapshot({ revision: 0 }) + await nextTask() + + // Open the window with a third subscription so a and b both defer. + const c = createDocumentSubscription({ + store, + definition: projectDef, + docId: 'p3', + collectionPath: 'projects', + }) + c.load() + h.fireDocSnapshot({ revision: 0 }) + + const thrown = new Error('subscriber failed') + a.subscribe(() => { + throw thrown + }) + const bSubscriber = vi.fn() + b.subscribe(bSubscriber) + + h.listeners()[0]!.onNext(docSnap({ revision: 1 })) + h.listeners()[1]!.onNext(docSnap({ revision: 1 })) + + const uncaught: unknown[] = [] + const onUncaught = (error: unknown) => uncaught.push(error) + process.on('uncaughtException', onUncaught) + try { + await nextTask() + await nextTask() + } finally { + process.off('uncaughtException', onUncaught) + } + + expect(bSubscriber).toHaveBeenCalledTimes(1) + expect(bSubscriber).toHaveBeenLastCalledWith(b.getState()) + expect(uncaught).toEqual([thrown]) + }) + + it('drops a stopped subscription from the shared trailing notify', async () => { + const a = await makeDoc() + const b = createDocumentSubscription({ + store, + definition: projectDef, + docId: 'p2', + collectionPath: 'projects', + }) + b.load() + h.fireDocSnapshot({ revision: 0 }) + await nextTask() + + const aSubscriber = vi.fn() + const bSubscriber = vi.fn() + a.subscribe(aSubscriber) + b.subscribe(bSubscriber) + + h.listeners()[0]!.onNext(docSnap({ revision: 1 })) + h.fireDocSnapshot({ revision: 1 }) + b.stop() + await nextTask() + + expect(aSubscriber).toHaveBeenCalledTimes(1) + expect(bSubscriber).not.toHaveBeenCalled() + }) + }) +}) diff --git a/src/core/collection.ts b/src/core/collection.ts index a062027..ec052c9 100644 --- a/src/core/collection.ts +++ b/src/core/collection.ts @@ -22,6 +22,7 @@ import type { UpdateOptions, } from '../types' import type { FirestateStore } from './store' +import { getSnapshotBatcher } from './snapshot-batcher' import { applyDiff, applyOverridesAtPaths, @@ -257,7 +258,9 @@ export const createCollectionSubscription = ( prev.isActive !== next.isActive || observableStateChanged(prev, next) - const notify = () => { + // Makes the current state visible to getState()/getHandle(). Returns + // whether anything observable changed. + const publishState = (): boolean => { // State is structurally shared between syncState, localState, and // every published snapshot, so it must never be mutated in place. freezeDeepInDevelopment(state.syncState) @@ -272,14 +275,34 @@ export const createCollectionSubscription = ( // Snapshot-side no-op collapse: nothing observable changed → publish // nothing, keeping the cached handle identity stable. if (lastPublished !== null && !publicStateChanged(lastPublished, publicState)) { - return + return false } lastPublished = publicState cachedHandle = null // Reuse the published state as the cached snapshot (see document.ts). cachedState = publicState - subscribers.forEach((fn) => fn(publicState)) store.reportSyncState(syncKey, publicState.isSynced) + return true + } + + const notifySubscribers = () => { + const publicState = getState() + subscribers.forEach((fn) => fn(publicState)) + } + + const notify = () => { + if (!publishState()) return + snapshotBatcher.discardDeferred(notifySubscribers) + notifySubscribers() + } + + // Snapshots update state at once but batch the subscriber fan-out across + // the store, so a burst of snapshots in one task re-renders consumers at + // most twice. + const snapshotBatcher = getSnapshotBatcher(store) + + const notifySnapshot = () => { + if (publishState()) snapshotBatcher.publishSnapshot(notifySubscribers) } // Pre-snapshot mutations are unsafe because computing a partial-update @@ -663,7 +686,7 @@ export const createCollectionSubscription = ( scheduleAutosave() } - notify() + notifySnapshot() } const handleError = (error: Error) => { @@ -752,6 +775,7 @@ export const createCollectionSubscription = ( clearTimeout(minLoadTimeout) minLoadTimeout = null } + snapshotBatcher.discardDeferred(notifySubscribers) // Drop this subscription's entry from the global sync-state map so // an unmounted hook does not leave useIsSynced stuck at false. store.unregisterSyncState(syncKey) diff --git a/src/core/document.ts b/src/core/document.ts index 40fbf61..9ccec5a 100644 --- a/src/core/document.ts +++ b/src/core/document.ts @@ -20,6 +20,7 @@ import type { UpdateOptions, } from '../types' import type { FirestateStore } from './store' +import { getSnapshotBatcher } from './snapshot-batcher' import { applyDiff, applyOverridesAtPaths, @@ -240,7 +241,9 @@ export const createDocumentSubscription = ( const publicStateChanged = observableStateChanged - const notify = () => { + // Makes the current state visible to getState()/getHandle(). Returns + // whether anything observable changed. + const publishState = (): boolean => { // State is structurally shared between syncState, localState, and // every published snapshot, so it must never be mutated in place. freezeDeepInDevelopment(state.syncState) @@ -260,15 +263,35 @@ export const createDocumentSubscription = ( // changed → publish nothing. Keeps the cached handle identity stable // so useSyncExternalStore does not re-render. if (lastPublished !== null && !publicStateChanged(lastPublished, publicState)) { - return + return false } lastPublished = publicState cachedHandle = null // Reuse the just-built state as the cached snapshot so getState() and // the published value share one identity-stable reference. cachedState = publicState - subscribers.forEach((fn) => fn(publicState)) store.reportSyncState(syncKey, publicState.isSynced) + return true + } + + const notifySubscribers = () => { + const publicState = getState() + subscribers.forEach((fn) => fn(publicState)) + } + + const notify = () => { + if (!publishState()) return + snapshotBatcher.discardDeferred(notifySubscribers) + notifySubscribers() + } + + // Snapshots update state at once but batch the subscriber fan-out across + // the store, so a burst of snapshots in one task re-renders consumers at + // most twice. + const snapshotBatcher = getSnapshotBatcher(store) + + const notifySnapshot = () => { + if (publishState()) snapshotBatcher.publishSnapshot(notifySubscribers) } const updateState = ( @@ -623,7 +646,7 @@ export const createDocumentSubscription = ( scheduleAutosave() } - notify() + notifySnapshot() } // A document that does not exist is not an error condition — consumers @@ -658,7 +681,7 @@ export const createDocumentSubscription = ( state.isLoading = false } loaded = true - notify() + notifySnapshot() } const handleError = (error: Error) => { @@ -731,6 +754,7 @@ export const createDocumentSubscription = ( clearTimeout(minLoadTimeout) minLoadTimeout = null } + snapshotBatcher.discardDeferred(notifySubscribers) // Drop this subscription's entry from the global sync-state map so // an unmounted hook does not leave useIsSynced stuck at false. store.unregisterSyncState(syncKey) diff --git a/src/core/snapshot-batcher.ts b/src/core/snapshot-batcher.ts new file mode 100644 index 0000000..905fab9 --- /dev/null +++ b/src/core/snapshot-batcher.ts @@ -0,0 +1,70 @@ +import type { FirestateStore } from './store' + +/** + * Firestore can deliver many listener snapshots in one task: its queue runs as + * a promise chain, so a burst of server writes (e.g. a Cloud Function that + * updates 100+ documents one by one) arrives back to back. Publishing each one + * makes React run one synchronous `useSyncExternalStore` commit per snapshot. + * When a consumer's effect schedules more work after each commit, React counts + * the commits as nested updates and throws "Maximum update depth exceeded" + * (#185) after 50 of them. + * + * The batcher publishes the first snapshot of a task right away, so a single + * snapshot stays synchronous. It folds every later snapshot of the same task + * into one publish per subscription on the next task. One batcher is shared by + * every subscription on a store, because a burst spread across many document + * subscriptions (one snapshot each) commits just as often as a burst on one. + */ +export const createSnapshotBatcher = () => { + let windowTimeout: ReturnType | null = null + const deferred = new Set<() => void>() + + const publishSnapshot = (publish: () => void) => { + if (windowTimeout !== null) { + deferred.add(publish) + return + } + // Open the window before publishing: if publish throws (React #185), + // the rest of the burst must still be batched. + windowTimeout = setTimeout(() => { + windowTimeout = null + const pending = [...deferred] + deferred.clear() + // Publishing in one synchronous loop lets React batch them into a + // single commit. A throw (e.g. React #185 from inside a subscriber) + // must not strand the other subscriptions, so it is rethrown last. + let firstError: { error: unknown } | null = null + for (const fn of pending) { + try { + fn() + } catch (error) { + firstError ??= { error } + } + } + if (firstError) throw firstError.error + }, 0) + publish() + } + + // Called when a synchronous notify already delivered the latest state, or + // when the subscription stops, so its deferred snapshot has nothing to do. + const discardDeferred = (publish: () => void) => { + deferred.delete(publish) + } + + return { publishSnapshot, discardDeferred } +} + +export type SnapshotBatcher = ReturnType + +const batchers = new WeakMap() + +/** The snapshot batcher shared by every subscription on `store`. */ +export const getSnapshotBatcher = (store: FirestateStore): SnapshotBatcher => { + let batcher = batchers.get(store) + if (!batcher) { + batcher = createSnapshotBatcher() + batchers.set(store, batcher) + } + return batcher +}