From 567c56cbf19b7bc70245de8b8ca42b7b9d235608 Mon Sep 17 00:00:00 2001 From: Marcus Pousette Date: Fri, 24 Jul 2026 11:30:55 +0200 Subject: [PATCH] Harden TreeCRDT runtime teardown --- .../treecrdt/__tests__/runtimeLifecycle.ts | 208 ++++++++++++++++++ src/data-providers/treecrdt/runtime.ts | 125 +++++++++-- src/data-providers/treecrdt/thoughtspace.ts | 7 +- src/e2e/puppeteer/__tests__/startup.ts | 14 +- 4 files changed, 335 insertions(+), 19 deletions(-) create mode 100644 src/data-providers/treecrdt/__tests__/runtimeLifecycle.ts diff --git a/src/data-providers/treecrdt/__tests__/runtimeLifecycle.ts b/src/data-providers/treecrdt/__tests__/runtimeLifecycle.ts new file mode 100644 index 00000000000..40a5413f0be --- /dev/null +++ b/src/data-providers/treecrdt/__tests__/runtimeLifecycle.ts @@ -0,0 +1,208 @@ +import type { TreecrdtClient } from '@treecrdt/wa-sqlite' +import { EM_TOKEN } from '../../../constants' +import createTreecrdtThoughtspace from '../runtime' +import { enqueueMaterializedThoughtsToStoreWork } from '../sync/materializationQueue' +import { withTreecrdtWriteBarrier } from '../writeBarrier' + +const { createTreecrdtClient } = vi.hoisted(() => ({ + createTreecrdtClient: vi.fn(), +})) + +vi.mock('@treecrdt/wa-sqlite', async importOriginal => { + const actual = await importOriginal() + return { ...actual, createTreecrdtClient } +}) + +let createActualTreecrdtClient: (typeof import('@treecrdt/wa-sqlite'))['createTreecrdtClient'] +let createdClients: TreecrdtClient[] = [] + +beforeAll(async () => { + const actual = await vi.importActual('@treecrdt/wa-sqlite') + createActualTreecrdtClient = actual.createTreecrdtClient +}) + +beforeEach(() => { + createdClients = [] + createTreecrdtClient.mockImplementation(async options => { + const client = await createActualTreecrdtClient(options) + createdClients.push(client) + return client + }) +}) + +afterEach(() => { + createTreecrdtClient.mockReset() +}) + +/** Creates an isolated in-memory thoughtspace for lifecycle tests. */ +const createTestThoughtspace = () => + createTreecrdtThoughtspace({ + client: { storage: 'memory', runtime: 'direct' }, + tabPolicy: 'multiple', + }) + +const emptyUpdate = { + thoughtIndexUpdates: {}, + lexemeIndexUpdates: {}, + lexemeIndexUpdatesOld: {}, + schemaVersion: 0, +} + +it('rejects persistence started during a failing init with the same error', async () => { + const initError = new Error('client init failed') + let markClientCreationStarted!: () => void + let rejectClientCreation!: () => void + const clientCreationStarted = new Promise(resolve => { + markClientCreationStarted = resolve + }) + createTreecrdtClient.mockImplementationOnce( + () => + new Promise((_, reject) => { + markClientCreationStarted() + rejectClientCreation = () => reject(initError) + }), + ) + const thoughtspace = createTestThoughtspace() + + const initializing = thoughtspace.init() + await clientCreationStarted + const persistence = thoughtspace.persistPushQueueBatches([emptyUpdate]) + const initResult = expect(initializing).rejects.toBe(initError) + const persistenceResult = expect(persistence).rejects.toBe(initError) + + rejectClientCreation() + + await initResult + await persistenceResult + await expect(thoughtspace.waitForIdle()).rejects.toBe(initError) +}) + +it('binds persistence called between drop and init to the next session', async () => { + const thoughtspace = createTestThoughtspace() + await thoughtspace.init() + + const dropping = thoughtspace.drop() + const persistence = thoughtspace.persistPushQueueBatches([emptyUpdate]) + const initializing = thoughtspace.init() + + await Promise.all([dropping, persistence, initializing]) + await expect(thoughtspace.db.getThoughtById(EM_TOKEN)).resolves.toMatchObject({ id: EM_TOKEN }) + + await thoughtspace.drop() +}) + +it('includes lifecycle-bound persistence in the idle barrier', async () => { + const thoughtspace = createTestThoughtspace() + await thoughtspace.init() + + let markClientCreationStarted!: () => void + let releaseClientCreation!: () => void + const clientCreationStarted = new Promise(resolve => { + markClientCreationStarted = resolve + }) + const clientCreationGate = new Promise(resolve => { + releaseClientCreation = resolve + }) + createTreecrdtClient.mockImplementationOnce(async options => { + markClientCreationStarted() + await clientCreationGate + const client = await createActualTreecrdtClient(options) + createdClients.push(client) + return client + }) + + const dropping = thoughtspace.drop() + const initializing = thoughtspace.init() + const persistence = thoughtspace.persistPushQueueBatches([emptyUpdate]) + const idle = thoughtspace.waitForIdle() + let idleSettled = false + void idle + .finally(() => { + idleSettled = true + }) + .catch(() => undefined) + + await clientCreationStarted + await Promise.resolve() + expect(idleSettled).toBe(false) + + releaseClientCreation() + await Promise.all([dropping, initializing, persistence, idle]) + expect(idleSettled).toBe(true) + + await thoughtspace.drop() +}) + +it('drains persistence and materialization before dropping the client', async () => { + const thoughtspace = createTestThoughtspace() + await thoughtspace.init() + + const writeError = new Error('write failed') + let markWriteStarted!: () => void + let rejectWrite!: () => void + let markMaterializationStarted!: () => void + let releaseMaterialization!: () => void + const writeStarted = new Promise(resolve => { + markWriteStarted = resolve + }) + const materializationStarted = new Promise(resolve => { + markMaterializationStarted = resolve + }) + const blockingWrite = withTreecrdtWriteBarrier( + () => + new Promise((_, reject) => { + rejectWrite = () => reject(writeError) + markWriteStarted() + }), + ) + const blockingWriteResult = expect(blockingWrite).rejects.toBe(writeError) + const blockingMaterialization = enqueueMaterializedThoughtsToStoreWork( + () => + new Promise(resolve => { + releaseMaterialization = resolve + markMaterializationStarted() + }), + ) + await Promise.all([writeStarted, materializationStarted]) + + const persistence = thoughtspace.persistPushQueueBatches([emptyUpdate]) + await Promise.resolve() + const dropClient = vi.spyOn(createdClients[0], 'drop') + const dropping = thoughtspace.drop() + const droppingResult = expect(dropping).rejects.toBe(writeError) + await Promise.resolve() + await Promise.resolve() + expect(dropClient).not.toHaveBeenCalled() + + rejectWrite() + await blockingWriteResult + await persistence + await Promise.resolve() + expect(dropClient).not.toHaveBeenCalled() + + releaseMaterialization() + await blockingMaterialization + await droppingResult + expect(dropClient).toHaveBeenCalledTimes(1) +}) + +it('retains ownership when both drop and close fail and blocks unsafe reuse', async () => { + const thoughtspace = createTestThoughtspace() + await thoughtspace.init() + + const dropError = new Error('drop failed') + const closeError = new Error('close failed') + vi.spyOn(createdClients[0], 'drop').mockRejectedValue(dropError) + vi.spyOn(createdClients[0], 'close').mockRejectedValueOnce(closeError) + + await expect(thoughtspace.drop()).rejects.toBe(dropError) + await expect(thoughtspace.persistPushQueueBatches([emptyUpdate])).rejects.toBe(dropError) + await expect(thoughtspace.init()).rejects.toThrow( + 'TreeCRDT client cleanup is incomplete. Retry drop before initialization.', + ) + + // drop still reports deletion failure, but the second close succeeds and releases ownership. + await expect(thoughtspace.drop()).rejects.toBe(dropError) + await expect(thoughtspace.init()).resolves.toEqual({ clientId: expect.any(String) }) + await thoughtspace.drop() +}) diff --git a/src/data-providers/treecrdt/runtime.ts b/src/data-providers/treecrdt/runtime.ts index b0228e85f6a..ec49a902f20 100644 --- a/src/data-providers/treecrdt/runtime.ts +++ b/src/data-providers/treecrdt/runtime.ts @@ -95,19 +95,28 @@ const getTreecrdtClientOptions = (config?: TreecrdtClientConfig): ClientOptions } } -/** Waits until both local writes and materialization refreshes are stable. */ +/** + * Waits until both local writes and materialization refreshes are stable. + * + * The app creates one active thoughtspace per browser realm, so these shared queues intentionally serialize its one + * Redux integration pipeline. Client, session, persistence, and WebSocket ownership remain factory-local. + */ const waitForStableIdle = async (): Promise => { let writeVersion: number let materializationVersion: number + let firstError: unknown + do { writeVersion = getTreecrdtWriteBarrierVersion() materializationVersion = getMaterializedThoughtsToStoreVersion() - await waitForTreecrdtWriteBarrier() - await waitForMaterializedThoughtsToStore() + const results = await Promise.allSettled([waitForTreecrdtWriteBarrier(), waitForMaterializedThoughtsToStore()]) + firstError ??= results.find(result => result.status === 'rejected')?.reason } while ( writeVersion !== getTreecrdtWriteBarrierVersion() || materializationVersion !== getMaterializedThoughtsToStoreVersion() ) + + if (firstError) throw firstError } /** Creates one TreeCRDT client owner and its bound app thoughtspace. */ @@ -127,8 +136,12 @@ const createTreecrdtThoughtspace = ({ let client: TreecrdtClient | null = null let unsubscribeMaterialization: (() => void) | null = null let lifecycleTail: Promise = Promise.resolve() + let latestLifecycle: Promise = lifecycleTail let initPromise: Promise | null = null let dropPromise: Promise | null = null + let persistenceVersion = 0 + const pendingPersistence = new Map>() + const persistenceErrors = new Map() const provider = createTreecrdtDataProvider() const websocketSync = createTreecrdtWebSocketSync() @@ -146,8 +159,57 @@ const createTreecrdtThoughtspace = ({ } } - /** Detaches the provider and releases all resources owned by this thoughtspace. */ - const dropClient = async (): Promise => { + /** Waits for factory-local persistence calls registered through the requested version. */ + const waitForPersistenceThrough = async (version: number): Promise => { + const pending = [...pendingPersistence] + .filter(([persistenceId]) => persistenceId <= version) + .map(([, promise]) => promise) + await Promise.allSettled(pending) + + const failure = [...persistenceErrors].find(([persistenceId]) => persistenceId <= version) + for (const persistenceId of persistenceErrors.keys()) { + if (persistenceId <= version) persistenceErrors.delete(persistenceId) + } + if (failure) throw failure[1] + } + + /** Drains local persistence first, then the shared write/materialization pipeline even if either reports an error. */ + const waitForRuntimeWorkThrough = async (version: number): Promise => { + let firstError: unknown + + try { + await waitForPersistenceThrough(version) + } catch (error) { + firstError = error + } + try { + await waitForStableIdle() + } catch (error) { + firstError ??= error + } + + if (firstError) throw firstError + } + + /** Waits until no newer factory-local persistence was registered while the current snapshot drained. */ + const waitForRuntimeIdle = async (): Promise => { + let firstError: unknown + let version: number + + do { + version = persistenceVersion + try { + await waitForRuntimeWorkThrough(version) + } catch (error) { + firstError ??= error + } + } while (version !== persistenceVersion) + + if (firstError) throw firstError + } + + /** Stops ingress, drains captured work, and releases all resources owned by this thoughtspace. */ + const dropClient = async (persistenceVersionAtDrop: number): Promise => { const errors: unknown[] = [] /** Records cleanup failures without skipping later owned resources. */ const captureError = async (work: () => void | Promise): Promise => { @@ -166,6 +228,9 @@ const createTreecrdtThoughtspace = ({ await captureError(websocketSync.stop) await captureError(() => unsubscribe?.()) + await captureError(() => withIdleTimeout(waitForRuntimeWorkThrough(persistenceVersionAtDrop))) + // Writes already running during the first stop may have produced buffered local ops. + await captureError(websocketSync.stop) let clientReleased = clientToDrop === null if (clientToDrop) { @@ -192,8 +257,10 @@ const createTreecrdtThoughtspace = ({ if (dropPromise) return dropPromise initPromise = null - const promise = lifecycleTail.then(dropClient) + const persistenceVersionAtDrop = persistenceVersion + const promise = lifecycleTail.then(() => dropClient(persistenceVersionAtDrop)) dropPromise = promise + latestLifecycle = promise lifecycleTail = promise.then( () => { if (dropPromise === promise) dropPromise = null @@ -207,17 +274,40 @@ const createTreecrdtThoughtspace = ({ const db: DataProvider = { ...provider.db, clear: drop } - /** Persists push queue batches through the bound provider and forwards local ops to remote sync. */ - const persistPushQueueBatches = (batches: readonly PersistTreecrdtBatch[]): Promise => - withTreecrdtWriteBarrier(async () => { - for (const batch of batches) { - const { local: isLocal, ...updates } = batch - const maybeOps = await db.updateThoughts(updates) - if (isLocal && Array.isArray(maybeOps) && maybeOps.length > 0) { - void websocketSync.pushLocalOps(maybeOps as readonly Operation[]) + /** Binds each persistence call to the exact session ordered before it and tracks it before any async wait. */ + const persistPushQueueBatches = (batches: readonly PersistTreecrdtBatch[]): Promise => { + persistenceVersion += 1 + const persistenceId = persistenceVersion + const precedingLifecycle = latestLifecycle + const persistence = (async () => { + await precedingLifecycle + const sessionDb = await provider.waitForSession() + + return withTreecrdtWriteBarrier(async () => { + for (const batch of batches) { + const { local: isLocal, ...updates } = batch + const maybeOps = await sessionDb.updateThoughts(updates) + if (isLocal && Array.isArray(maybeOps) && maybeOps.length > 0) { + void websocketSync.pushLocalOps(maybeOps as readonly Operation[]) + } } - } - }) + }) + })() + + const trackedPersistence = persistence.then( + () => { + pendingPersistence.delete(persistenceId) + }, + error => { + pendingPersistence.delete(persistenceId) + persistenceErrors.set(persistenceId, error) + throw error + }, + ) + pendingPersistence.set(persistenceId, trackedPersistence) + void trackedPersistence.catch(() => undefined) + return trackedPersistence + } /** Opens and binds one client. Lifecycle serialization provides retryable single-flight behavior. */ const initializeClient = async (options?: ThoughtspaceRuntimeInitOptions): Promise => { @@ -254,6 +344,7 @@ const createTreecrdtThoughtspace = ({ dropPromise = null const promise = lifecycleTail.then(() => initializeClient(options)) initPromise = promise + latestLifecycle = promise lifecycleTail = promise.then( () => undefined, () => undefined, @@ -269,7 +360,7 @@ const createTreecrdtThoughtspace = ({ acquireAccess, init, drop, - waitForIdle: (): Promise => withIdleTimeout(waitForStableIdle()), + waitForIdle: (): Promise => withIdleTimeout(waitForRuntimeIdle()), persistPushQueueBatches, } } diff --git a/src/data-providers/treecrdt/thoughtspace.ts b/src/data-providers/treecrdt/thoughtspace.ts index c241c21e392..fdde8610a89 100644 --- a/src/data-providers/treecrdt/thoughtspace.ts +++ b/src/data-providers/treecrdt/thoughtspace.ts @@ -414,9 +414,13 @@ const createTreecrdtDataProvider = () => { return activeSession } + /** Captures the exact session provider that is active now or becomes active after delayed initialization. */ + const waitForSession = async (): Promise => + (activeSession ?? (await sessionGate.promise)).db + /** Dispatches public writes through the startup gate, retaining whichever session releases that write. */ const updateThoughts: DataProvider['updateThoughts'] = async updates => - (await sessionGate.promise).db.updateThoughts(updates) + (await waitForSession()).updateThoughts(updates) /** Detaches the current session, rejects startup writes, and rotates to a fresh gate. */ const resetSession = (reason: unknown): void => { @@ -477,6 +481,7 @@ const createTreecrdtDataProvider = () => { db, bindSession, resetSession, + waitForSession, } } diff --git a/src/e2e/puppeteer/__tests__/startup.ts b/src/e2e/puppeteer/__tests__/startup.ts index 83327b0fa10..088c1ad9f9a 100644 --- a/src/e2e/puppeteer/__tests__/startup.ts +++ b/src/e2e/puppeteer/__tests__/startup.ts @@ -5,6 +5,8 @@ import { page } from '../session' vi.setConfig({ testTimeout: 30000, hookTimeout: 20000 }) it('handles keyboard commands while thoughtspace initialization is delayed', async () => { + const queuedValue = 'queued before initialization' + await page.evaluateOnNewDocument(() => { const preloadedWindow = window as unknown as PreloadedEmWindow preloadedWindow.em = { @@ -27,7 +29,8 @@ it('handles keyboard commands while thoughtspace initialization is delayed', asy await page.keyboard.press('Enter') await page.waitForSelector('[data-editing=true] [data-editable]') - expect(await getEditingText()).toBe('') + await page.keyboard.type(queuedValue) + expect(await getEditingText()).toBe(queuedValue) } finally { await page.evaluate(async () => { const em = window.em @@ -35,4 +38,13 @@ it('handles keyboard commands while thoughtspace initialization is delayed', asy em.testFlags.preventInitialize = false }) } + + await page.waitForFunction( + async value => { + await window.em.testHelpers.waitForThoughtspaceRuntimeIdle() + return (await window.em.testHelpers.getLexemeFromThoughtspace(value))?.contexts.length + }, + {}, + queuedValue, + ) })