diff --git a/.changeset/tidy-pillows-brush.md b/.changeset/tidy-pillows-brush.md new file mode 100644 index 000000000..fb9530f33 --- /dev/null +++ b/.changeset/tidy-pillows-brush.md @@ -0,0 +1,9 @@ +--- +'@powersync/shared-internals': patch +--- + +Fix watched queries leaking listeners when closed during initialization. Closing a watched query +while it was still resolving tables (or waiting for the database to be ready) registered a +permanent `tablesUpdated` listener on the database adapter, because `onChangeWithCallback` relied +on an `abort` event that an already-aborted signal never emits. In `@powersync/react`, every +`useQuery` mount hit this race and leaked one adapter listener. diff --git a/packages/shared-internals/src/client/BasePowerSyncDatabase.ts b/packages/shared-internals/src/client/BasePowerSyncDatabase.ts index 9efa63072..7bd6bc5a1 100644 --- a/packages/shared-internals/src/client/BasePowerSyncDatabase.ts +++ b/packages/shared-internals/src/client/BasePowerSyncDatabase.ts @@ -774,6 +774,13 @@ SELECT * FROM crud_entries; } const resolvedOptions = options ?? {}; + + // An already-aborted signal never emits an `abort` event, so the listener registered + // below would never be disposed. Avoid registering anything in that case. + if (resolvedOptions.signal?.aborted) { + return () => {}; + } + const watchedTables = new Set( (resolvedOptions?.tables ?? []).flatMap((table) => [table, `ps_data__${table}`, `ps_data_local__${table}`]) ); diff --git a/packages/shared-internals/src/client/watched/AbstractQueryProcessor.ts b/packages/shared-internals/src/client/watched/AbstractQueryProcessor.ts index 04b27a9b3..5286fd796 100644 --- a/packages/shared-internals/src/client/watched/AbstractQueryProcessor.ts +++ b/packages/shared-internals/src/client/watched/AbstractQueryProcessor.ts @@ -166,6 +166,14 @@ export abstract class AbstractQueryProcessor< // Wait for the schema to be set before listening to changes await db.waitForReady(); + + // close() might have been called while waiting for the database to be ready. At that point + // disposeListeners was not assigned yet, so dispose the closing listener here. + if (this._closed) { + disposeCloseListener(); + return; + } + const disposeSchemaListener = db.registerListener({ schemaChanged: async () => { await this.runWithReporting(async () => { diff --git a/packages/shared-internals/src/client/watched/DifferentialQueryProcessor.ts b/packages/shared-internals/src/client/watched/DifferentialQueryProcessor.ts index e171b2671..ace52349d 100644 --- a/packages/shared-internals/src/client/watched/DifferentialQueryProcessor.ts +++ b/packages/shared-internals/src/client/watched/DifferentialQueryProcessor.ts @@ -147,6 +147,12 @@ export class DifferentialQueryProcessor tables: options.settings.triggerOnTables }); + // The query might have been closed or aborted while resolving tables. + // Registering the change listener at this point would leak it. + if (this.closed || abortSignal.aborted) { + return; + } + let currentMap: DataHashMap = new Map(); // populate the currentMap from the placeholder data diff --git a/packages/shared-internals/src/client/watched/OnChangeQueryProcessor.ts b/packages/shared-internals/src/client/watched/OnChangeQueryProcessor.ts index 0b86a6a65..4e5b09abc 100644 --- a/packages/shared-internals/src/client/watched/OnChangeQueryProcessor.ts +++ b/packages/shared-internals/src/client/watched/OnChangeQueryProcessor.ts @@ -43,6 +43,12 @@ export class OnChangeQueryProcessor extends AbstractQueryProcessor { diff --git a/packages/shared-internals/tests/client/watched/listenerLeaks.test.ts b/packages/shared-internals/tests/client/watched/listenerLeaks.test.ts new file mode 100644 index 000000000..d481955eb --- /dev/null +++ b/packages/shared-internals/tests/client/watched/listenerLeaks.test.ts @@ -0,0 +1,195 @@ +import { DBAdapter, LockContext, RawQueryResult, Schema } from '@powersync/common'; +import { describe, expect, it } from 'vitest'; +import { BasePowerSyncDatabase } from '../../../src/client/BasePowerSyncDatabase.js'; + +class MockLockContext extends LockContext { + async executeRaw(): Promise { + return { columnNames: [], rawRows: [] }; + } +} + +class MockDBAdapter extends DBAdapter { + /** + * Read locks await this promise before executing. + * Tests use this to park a query (e.g. `resolveTables`) mid-await. + */ + readGate: Promise = Promise.resolve(); + + get name() { + return 'mock-db'; + } + + /** + * The number of registered adapter listeners. Exposed for leak assertions. + */ + get registeredListenerCount() { + return this.listeners.size; + } + + async close() {} + + async refreshSchema() {} + + async readLock(fn: (tx: LockContext) => Promise): Promise { + await this.readGate; + return fn(new MockLockContext()); + } + + async writeLock(fn: (tx: LockContext) => Promise): Promise { + return fn(new MockLockContext()); + } +} + +class TestPowerSyncDatabase extends BasePowerSyncDatabase { + /** + * When set, `waitForReady` awaits this promise. + * Tests use this to park watched-query initialization mid-await. + */ + readyGate: Promise | null = null; + + /** + * The number of registered database listeners. Exposed for leak assertions. + */ + get registeredListenerCount() { + return this.listeners.size; + } + + get mockAdapter() { + return this.database as MockDBAdapter; + } + + protected openDBAdapter(): DBAdapter { + return new MockDBAdapter(); + } + + protected generateSyncStreamImplementation(): never { + throw new Error('Sync is not required for these tests'); + } + + protected generateBucketStorageAdapter(): never { + return null as never; + } + + protected async _initialize() {} + + // The real initialization runs queries against the SQLite database, which is not required here. + protected async initialize() { + this.ready = true; + } + + async waitForReady(): Promise { + await this.readyGate; + } +} + +const createTestDatabase = () => new TestPowerSyncDatabase({ schema: new Schema([]) }); + +const watchQuery = (db: TestPowerSyncDatabase) => + db + .customQuery<{ id: string }>({ + compile: () => ({ sql: 'SELECT * FROM todos', parameters: [] }), + execute: async () => [] + }) + .watch({}); + +const deferred = () => { + let resolve!: () => void; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +}; + +/** Waits for pending micro and macro tasks to complete. */ +const settle = async () => { + for (let i = 0; i < 10; i++) { + await new Promise((r) => setTimeout(r)); + } +}; + +describe('watched query listener cleanup', () => { + it('onChangeWithCallback should not register an adapter listener for an already-aborted signal', () => { + const db = createTestDatabase(); + const controller = new AbortController(); + controller.abort(); + + const dispose = db.onChangeWithCallback({ onChange: () => {} }, { signal: controller.signal, tables: ['todos'] }); + + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + // The returned dispose function should still be safe to call + dispose(); + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + }); + + it('onChangeWithCallback should dispose the adapter listener when the signal aborts', () => { + const db = createTestDatabase(); + const controller = new AbortController(); + + db.onChangeWithCallback({ onChange: () => {} }, { signal: controller.signal, tables: ['todos'] }); + + expect(db.mockAdapter.registeredListenerCount).toEqual(1); + controller.abort(); + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + }); + + it('should register and dispose listeners across a watched query lifecycle', async () => { + const db = createTestDatabase(); + const baseListenerCount = db.registeredListenerCount; + + const query = watchQuery(db); + await settle(); + + expect(db.mockAdapter.registeredListenerCount).toEqual(1); + // The closing and schemaChanged listeners + expect(db.registeredListenerCount).toEqual(baseListenerCount + 2); + + await query.close(); + await settle(); + + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + expect(db.registeredListenerCount).toEqual(baseListenerCount); + }); + + it('should not leak listeners when a watched query is closed while resolving tables', async () => { + const db = createTestDatabase(); + const baseListenerCount = db.registeredListenerCount; + + // Park the query's `resolveTables` call on its read lock + const readGate = deferred(); + db.mockAdapter.readGate = readGate.promise; + + // Mimics React's useQuery, which creates a watched query during render + // and closes it from a mount effect one frame later. + const query = watchQuery(db); + await settle(); + await query.close(); + + readGate.resolve(); + await settle(); + + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + expect(db.registeredListenerCount).toEqual(baseListenerCount); + }); + + it('should not leak listeners when a watched query is closed while waiting for ready', async () => { + const db = createTestDatabase(); + const baseListenerCount = db.registeredListenerCount; + + // Park the watched query's initialization on `waitForReady` + const readyGate = deferred(); + db.readyGate = readyGate.promise; + + const query = watchQuery(db); + await settle(); + + // The closing listener is registered before waiting for ready + expect(db.registeredListenerCount).toEqual(baseListenerCount + 1); + + await query.close(); + readyGate.resolve(); + await settle(); + + expect(db.registeredListenerCount).toEqual(baseListenerCount); + expect(db.mockAdapter.registeredListenerCount).toEqual(0); + }); +});