From 2e9df4be915c175f8d84220ca3265a104b4e378d Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 22 May 2025 13:43:16 +0200 Subject: [PATCH 01/26] improve queue behaviour --- .changeset/empty-pants-give.md | 8 + packages/common/rollup.config.mjs | 5 +- .../src/client/AbstractPowerSyncDatabase.ts | 177 +++++++++++++++--- packages/node/src/db/PowerSyncDatabase.ts | 12 +- .../react-native/src/db/PowerSyncDatabase.ts | 7 + packages/web/src/db/PowerSyncDatabase.ts | 13 +- .../src/db/AbstractPowerSyncDatabase.test.ts | 10 +- packages/web/tests/stream.test.ts | 25 ++- 8 files changed, 207 insertions(+), 50 deletions(-) create mode 100644 .changeset/empty-pants-give.md diff --git a/.changeset/empty-pants-give.md b/.changeset/empty-pants-give.md new file mode 100644 index 000000000..c343f0bcb --- /dev/null +++ b/.changeset/empty-pants-give.md @@ -0,0 +1,8 @@ +--- +'@powersync/react-native': minor +'@powersync/common': minor +'@powersync/node': minor +'@powersync/web': minor +--- + +Improved behvaiour when connect is called multiple times in quick succession. Updating client parameters should now be more responsive. diff --git a/packages/common/rollup.config.mjs b/packages/common/rollup.config.mjs index ad05faf6a..4e64e9212 100644 --- a/packages/common/rollup.config.mjs +++ b/packages/common/rollup.config.mjs @@ -2,7 +2,6 @@ import commonjs from '@rollup/plugin-commonjs'; import inject from '@rollup/plugin-inject'; import json from '@rollup/plugin-json'; import nodeResolve from '@rollup/plugin-node-resolve'; -import terser from '@rollup/plugin-terser'; export default (commandLineArgs) => { const sourcemap = (commandLineArgs.sourceMap || 'true') == 'true'; @@ -26,8 +25,8 @@ export default (commandLineArgs) => { ReadableStream: ['web-streams-polyfill/ponyfill', 'ReadableStream'], // Used by can-ndjson-stream TextDecoder: ['text-encoding', 'TextDecoder'] - }), - terser() + }) + // terser() ], // This makes life easier external: [ diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index c78912c99..0df4850e6 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -9,13 +9,14 @@ import { UpdateNotification, isBatchedUpdateNotification } from '../db/DBAdapter.js'; +import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js'; import { SyncPriorityStatus, SyncStatus } from '../db/crud/SyncStatus.js'; import { UploadQueueStats } from '../db/crud/UploadQueueStatus.js'; import { Schema } from '../db/schema/Schema.js'; import { BaseObserver } from '../utils/BaseObserver.js'; import { ControlledExecutor } from '../utils/ControlledExecutor.js'; -import { mutexRunExclusive } from '../utils/mutex.js'; import { throttleTrailing } from '../utils/async.js'; +import { mutexRunExclusive } from '../utils/mutex.js'; import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js'; import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js'; import { runOnSchemaChange } from './runOnSchemaChange.js'; @@ -32,7 +33,6 @@ import { type PowerSyncConnectionOptions, type RequiredAdditionalConnectionOptions } from './sync/stream/AbstractStreamingSyncImplementation.js'; -import { FULL_SYNC_PRIORITY } from '../db/crud/SyncProgress.js'; export interface DisconnectAndClearOptions { /** When set to false, data in local-only tables is preserved. */ @@ -112,6 +112,11 @@ export interface PowerSyncCloseOptions { disconnect?: boolean; } +type StoredConnectionOptions = { + connector: PowerSyncBackendConnector; + options: PowerSyncConnectionOptions; +}; + const POWERSYNC_TABLE_MATCH = /(^ps_data__|^ps_data_local__)/; const DEFAULT_DISCONNECT_CLEAR_OPTIONS: DisconnectAndClearOptions = { @@ -176,6 +181,29 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver | null; + /** + * Tracks actively instantiating a streaming sync implementation. + */ + protected syncStreamInitPromise: Promise | null; + /** + * Active disconnect operation. Call disconnect multiple times + * will resolve to the same operation. + */ + protected disconnectingPromise: Promise | null; + /** + * Tracks the last parameters supplied to `connect` calls. + * Calling `connect` multiple times in succession will result in: + * - 1 pending connection operation which will be aborted. + * - updating the last set of parameters while waiting for the pending + * attempt to be aborted + * - internally connecting with the last set of parameters + */ + protected pendingConnectionOptions: StoredConnectionOptions | null; + constructor(options: PowerSyncDatabaseOptionsWithDBAdapter); constructor(options: PowerSyncDatabaseOptionsWithOpenFactory); constructor(options: PowerSyncDatabaseOptionsWithSettings); @@ -206,6 +234,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver Promise): Promise; + + protected async connectInternal() { + await this.disconnectInternal(); + + let appliedOptions: PowerSyncConnectionOptions | null = null; + + /** + * This portion creates a sync implementation which can be racy when disconnecting or + * if multiple tabs on web are in use. + * This is protected in an exclusive lock. + * The promise tracks the creation which is used to synchronize disconnect attempts. + */ + this.syncStreamInitPromise = this.connectExclusive(async () => { + if (this.closed) { + throw new Error('Cannot connect using a closed client'); + } + if (!this.pendingConnectionOptions) { + // A disconnect could have cleared this. + return; + } + + // get pending options and clear it in order for other connect attempts to queue other options + + const { connector, options } = this.pendingConnectionOptions; + appliedOptions = options; + this.pendingConnectionOptions = null; + + this.syncStreamImplementation = this.generateSyncStreamImplementation( + connector, + this.resolvedConnectionOptions(options) + ); + this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({ + statusChanged: (status) => { + this.currentStatus = new SyncStatus({ + ...status.toJSON(), + hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt + }); + this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus)); + } + }); + + await this.syncStreamImplementation.waitForReady(); + }); + + await this.syncStreamInitPromise; + this.syncStreamInitPromise = null; + + if (!appliedOptions) { + // A disconnect could have cleared the options which did not create a syncStreamImplementation + return; + } + + this.syncStreamImplementation?.triggerCrudUpload(); + this.options.logger?.debug('Attempting to connect to PowerSync instance'); + await this.syncStreamImplementation?.connect(appliedOptions!); + } + /** * Connects to stream of events from the PowerSync instance. */ async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) { await this.waitForReady(); - // close connection if one is open - await this.disconnect(); - if (this.closed) { - throw new Error('Cannot connect using a closed client'); - } + // A pending connection should be present if this is true + // The options also have not been used yet if this is true. + // We can update this referrence in order to update the next connection attempt. + const hadPendingConnectionOptions = !!this.pendingConnectionOptions; - const resolvedConnectOptions = this.resolvedConnectionOptions(options); + // This overrides options if present. + this.pendingConnectionOptions = { + connector, + options: options ?? {} + }; - this.syncStreamImplementation = this.generateSyncStreamImplementation(connector, resolvedConnectOptions); - this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({ - statusChanged: (status) => { - this.currentStatus = new SyncStatus({ - ...status.toJSON(), - hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt - }); - this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus)); + if (hadPendingConnectionOptions) { + // A connection attempt is already queued, but it hasn't used the options yet. + // The params have been updated and will be used when connecting. + if (!this.connectingPromise) { + throw new Error(`Pending connection options were found without a pending connect operation`); } - }); + return await this.connectingPromise; + } else if (this.connectingPromise) { + // If we didn't have pending options, we are busy with a connect. + // i.e. the pending connect used the options already and is busy proceeding. + // The call which creates the connectingPromise should check if there are pendingConnectionOptions and automatically + // schedule a connect. See below: + } else { + // No pending options or pending operation. Start one + this.connectingPromise = this.connectInternal().finally(() => { + if (this.pendingConnectionOptions) { + return this.connectInternal(); + } + this.connectingPromise = null; + return; + }); + return await this.connectingPromise; + } + } - await this.syncStreamImplementation.waitForReady(); - this.syncStreamImplementation.triggerCrudUpload(); - await this.syncStreamImplementation.connect(options); + /** + * Close the sync connection. + * + * Use {@link connect} to connect again. + */ + protected async disconnectInternal() { + if (this.disconnectingPromise) { + // A disconnect is already in progress + return await this.disconnectingPromise; + } + + // Wait if a sync stream implementation is being created before closing it + // (it must be assigned before we can properly dispose it) + await this.syncStreamInitPromise; + + this.disconnectingPromise = (async () => { + await this.syncStreamImplementation?.disconnect(); + this.syncStatusListenerDisposer?.(); + await this.syncStreamImplementation?.dispose(); + this.syncStreamImplementation = undefined; + })(); + + await this.disconnectingPromise; + this.disconnectingPromise = null; } /** @@ -462,10 +592,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { @@ -83,6 +85,10 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return super.connect(connector, options); } + protected async connectExclusive(callback: () => Promise): Promise { + await this.connectionLock.acquire(`connection-lock-${this.database.name}`, callback); + } + protected generateSyncStreamImplementation( connector: PowerSyncBackendConnector, options: NodeAdditionalConnectionOptions diff --git a/packages/react-native/src/db/PowerSyncDatabase.ts b/packages/react-native/src/db/PowerSyncDatabase.ts index 3d37a7e9b..5e8c0b676 100644 --- a/packages/react-native/src/db/PowerSyncDatabase.ts +++ b/packages/react-native/src/db/PowerSyncDatabase.ts @@ -8,6 +8,7 @@ import { type RequiredAdditionalConnectionOptions, SqliteBucketStorage } from '@powersync/common'; +import Lock from 'async-lock'; import { ReactNativeRemote } from '../sync/stream/ReactNativeRemote'; import { ReactNativeStreamingSyncImplementation } from '../sync/stream/ReactNativeStreamingSyncImplementation'; import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick-sqlite/ReactNativeQuickSQLiteOpenFactory'; @@ -27,6 +28,8 @@ import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick * ``` */ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { + protected connectionLock = new Lock(); + async _initialize(): Promise {} /** @@ -42,6 +45,10 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex); } + protected async connectExclusive(callback: () => Promise): Promise { + await this.connectionLock.acquire(`connection-lock-${this.database.name}`, callback); + } + protected generateSyncStreamImplementation( connector: PowerSyncBackendConnector, options: RequiredAdditionalConnectionOptions diff --git a/packages/web/src/db/PowerSyncDatabase.ts b/packages/web/src/db/PowerSyncDatabase.ts index 16001712e..308a3c9ae 100644 --- a/packages/web/src/db/PowerSyncDatabase.ts +++ b/packages/web/src/db/PowerSyncDatabase.ts @@ -2,7 +2,6 @@ import { type BucketStorageAdapter, type PowerSyncBackendConnector, type PowerSyncCloseOptions, - type PowerSyncConnectionOptions, type RequiredAdditionalConnectionOptions, AbstractPowerSyncDatabase, DBAdapter, @@ -172,16 +171,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { }); } - connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions): Promise { - /** - * Using React strict mode might cause calls to connect to fire multiple times - * Connect is wrapped inside a lock in order to prevent race conditions internally between multiple - * connection attempts. - */ - return this.runExclusive(() => { - this.options.logger?.debug('Attempting to connect to PowerSync instance'); - return super.connect(connector, options); - }); + protected async connectExclusive(callback: () => Promise): Promise { + await this.runExclusive(callback); } protected generateBucketStorageAdapter(): BucketStorageAdapter { diff --git a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts index 9885af547..df0a0f010 100644 --- a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts +++ b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts @@ -1,15 +1,15 @@ -import { describe, it, expect, vi } from 'vitest'; import { AbstractPowerSyncDatabase, - DEFAULT_RETRY_DELAY_MS, - DEFAULT_CRUD_UPLOAD_THROTTLE_MS, BucketStorageAdapter, DBAdapter, + DEFAULT_CRUD_UPLOAD_THROTTLE_MS, + DEFAULT_RETRY_DELAY_MS, PowerSyncBackendConnector, PowerSyncDatabaseOptionsWithSettings, RequiredAdditionalConnectionOptions, StreamingSyncImplementation } from '@powersync/common'; +import { describe, expect, it, vi } from 'vitest'; import { testSchema } from '../../utils/testDb'; class TestPowerSyncDatabase extends AbstractPowerSyncDatabase { @@ -31,6 +31,10 @@ class TestPowerSyncDatabase extends AbstractPowerSyncDatabase { return Promise.resolve(); } + protected async connectExclusive(callback: () => Promise): Promise { + await callback(); + } + get database() { return { get: vi.fn().mockResolvedValue({ version: '0.3.11' }), diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 2fc8c75a4..d0eef68d2 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -1,4 +1,4 @@ -import { BucketChecksum, WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web'; +import { BucketChecksum, PowerSyncConnectionOptions, WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web'; import { describe, expect, it, onTestFinished, vi } from 'vitest'; import { TestConnector } from './utils/MockStreamOpenFactory'; import { ConnectedDatabaseUtils, generateConnectedDatabase } from './utils/generateConnectedDatabase'; @@ -163,12 +163,25 @@ function describeStreamingTests(createConnectedDatabase: () => Promise { + const call = spy.mock.lastCall![1] as PowerSyncConnectionOptions; + expect(call.params!['count']).eq(connectionAttempts); + }, + { timeout: 2000, interval: 100 } + ); + + // In this case it should most likely be 1 attempt since all the calls + // are in the same for loop + expect(spy.mock.calls.length).lessThan(connectionAttempts); }); it('Should trigger upload connector when connected', async () => { From c4eccb8f9770bbdc9b1a7bb1612c987684ed72d2 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 22 May 2025 14:16:49 +0200 Subject: [PATCH 02/26] cleanup --- .../src/client/AbstractPowerSyncDatabase.ts | 4 +-- packages/node/src/db/PowerSyncDatabase.ts | 8 ++--- .../react-native/src/db/PowerSyncDatabase.ts | 6 ++-- packages/web/src/db/PowerSyncDatabase.ts | 2 +- .../src/db/AbstractPowerSyncDatabase.test.ts | 4 +-- .../tests/src/db/PowersyncDatabase.test.ts | 36 ++++--------------- 6 files changed, 18 insertions(+), 42 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index 0df4850e6..e4a2cdb54 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -459,7 +459,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver Promise): Promise; + protected abstract runExclusive(callback: () => Promise): Promise; protected async connectInternal() { await this.disconnectInternal(); @@ -472,7 +472,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { + this.syncStreamInitPromise = this.runExclusive(async () => { if (this.closed) { throw new Error('Cannot connect using a closed client'); } diff --git a/packages/node/src/db/PowerSyncDatabase.ts b/packages/node/src/db/PowerSyncDatabase.ts index 420a7531e..80f05b30b 100644 --- a/packages/node/src/db/PowerSyncDatabase.ts +++ b/packages/node/src/db/PowerSyncDatabase.ts @@ -56,11 +56,11 @@ export type NodePowerSyncConnectionOptions = PowerSyncConnectionOptions & NodeAd * ``` */ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { - protected connectionLock: Lock; + protected lock: Lock; constructor(options: NodePowerSyncDatabaseOptions) { super(options); - this.connectionLock = new Lock(); + this.lock = new Lock(); } async _initialize(): Promise { @@ -85,8 +85,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return super.connect(connector, options); } - protected async connectExclusive(callback: () => Promise): Promise { - await this.connectionLock.acquire(`connection-lock-${this.database.name}`, callback); + protected async runExclusive(callback: () => Promise): Promise { + return await this.lock.acquire(`lock-${this.database.name}`, callback); } protected generateSyncStreamImplementation( diff --git a/packages/react-native/src/db/PowerSyncDatabase.ts b/packages/react-native/src/db/PowerSyncDatabase.ts index 5e8c0b676..a68957a13 100644 --- a/packages/react-native/src/db/PowerSyncDatabase.ts +++ b/packages/react-native/src/db/PowerSyncDatabase.ts @@ -28,7 +28,7 @@ import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick * ``` */ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { - protected connectionLock = new Lock(); + protected lock = new Lock(); async _initialize(): Promise {} @@ -45,8 +45,8 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex); } - protected async connectExclusive(callback: () => Promise): Promise { - await this.connectionLock.acquire(`connection-lock-${this.database.name}`, callback); + protected async runExclusive(callback: () => Promise): Promise { + return await this.lock.acquire(`lock-${this.database.name}`, callback); } protected generateSyncStreamImplementation( diff --git a/packages/web/src/db/PowerSyncDatabase.ts b/packages/web/src/db/PowerSyncDatabase.ts index 308a3c9ae..072b62da0 100644 --- a/packages/web/src/db/PowerSyncDatabase.ts +++ b/packages/web/src/db/PowerSyncDatabase.ts @@ -179,7 +179,7 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex); } - protected runExclusive(cb: () => Promise) { + protected async runExclusive(cb: () => Promise) { if (this.resolvedFlags.ssrMode) { return PowerSyncDatabase.SHARED_MUTEX.runExclusive(cb); } diff --git a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts index df0a0f010..12e89f2cf 100644 --- a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts +++ b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts @@ -31,8 +31,8 @@ class TestPowerSyncDatabase extends AbstractPowerSyncDatabase { return Promise.resolve(); } - protected async connectExclusive(callback: () => Promise): Promise { - await callback(); + protected async runExclusive(callback: () => Promise): Promise { + return await callback(); } get database() { diff --git a/packages/web/tests/src/db/PowersyncDatabase.test.ts b/packages/web/tests/src/db/PowersyncDatabase.test.ts index ad44d3b9e..a7487ece1 100644 --- a/packages/web/tests/src/db/PowersyncDatabase.test.ts +++ b/packages/web/tests/src/db/PowersyncDatabase.test.ts @@ -1,5 +1,5 @@ -import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; -import { AbstractPowerSyncDatabase, PowerSyncDatabase, SyncStreamConnectionMethod } from '@powersync/web'; +import { createBaseLogger, PowerSyncDatabase } from '@powersync/web'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { testSchema } from '../../utils/testDb'; describe('PowerSyncDatabase', () => { @@ -9,9 +9,10 @@ describe('PowerSyncDatabase', () => { let mockSyncImplementation: any; beforeEach(() => { + const logger = createBaseLogger(); mockLogger = { - debug: vi.fn(), - warn: vi.fn() + debug: vi.spyOn(logger, 'debug'), + warn: vi.spyOn(logger, 'warn') }; // Initialize with minimal required options @@ -20,12 +21,8 @@ describe('PowerSyncDatabase', () => { database: { dbFilename: 'test.db' }, - logger: mockLogger + logger }); - - vi.spyOn(db as any, 'runExclusive').mockImplementation((cb: any) => cb()); - - vi.spyOn(AbstractPowerSyncDatabase.prototype, 'connect').mockResolvedValue(undefined); }); afterEach(() => { @@ -36,27 +33,6 @@ describe('PowerSyncDatabase', () => { it('should log debug message when attempting to connect', async () => { await db.connect(mockConnector); expect(mockLogger.debug).toHaveBeenCalledWith('Attempting to connect to PowerSync instance'); - expect(db['runExclusive']).toHaveBeenCalled(); - }); - - it('should use connect with correct options', async () => { - await db.connect(mockConnector, { - retryDelayMs: 1000, - crudUploadThrottleMs: 2000, - params: { - param1: 1 - }, - connectionMethod: SyncStreamConnectionMethod.HTTP - }); - - expect(AbstractPowerSyncDatabase.prototype.connect).toHaveBeenCalledWith(mockConnector, { - retryDelayMs: 1000, - crudUploadThrottleMs: 2000, - connectionMethod: 'http', - params: { - param1: 1 - } - }); }); }); }); From 5ef22b1280e01db8c5a1103d882e41ae77ca8c95 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 22 May 2025 18:08:00 +0200 Subject: [PATCH 03/26] simplify logic --- .../src/client/AbstractPowerSyncDatabase.ts | 44 +++++-------------- .../AbstractStreamingSyncImplementation.ts | 24 +++++++--- packages/web/tests/stream.test.ts | 14 ++++++ .../web/tests/utils/MockStreamOpenFactory.ts | 4 +- packages/web/vitest.config.ts | 2 +- 5 files changed, 47 insertions(+), 41 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index e4a2cdb54..e14687791 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -462,8 +462,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver(callback: () => Promise): Promise; protected async connectInternal() { - await this.disconnectInternal(); - let appliedOptions: PowerSyncConnectionOptions | null = null; /** @@ -521,42 +519,25 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { - if (this.pendingConnectionOptions) { - return this.connectInternal(); - } + await this.waitForReady(); + await this.disconnectInternal(); + + const chain = (result) => { + if (this.pendingConnectionOptions) { + return this.connectInternal().then(chain); + } else { this.connectingPromise = null; - return; - }); - return await this.connectingPromise; - } + return result; + } + }; + + return this.connectingPromise ?? this.connectInternal().then(chain); } /** @@ -569,7 +550,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver((resolve) => { - const l = this.registerListener({ + let disposer: () => void; + + const complete = () => { + disposer?.(); + resolve(); + }; + + controller.signal.addEventListener('abort', complete, { once: true }); + + disposer = this.registerListener({ statusUpdated: (update) => { // This is triggered as soon as a connection is read from if (typeof update.connected == 'undefined') { @@ -361,8 +371,8 @@ The next upload iteration will be delayed.`); this.logger.warn('Initial connect attempt did not successfully connect to server'); } - resolve(); - l(); + controller.signal.removeEventListener('abort', complete); + complete(); } }); }); diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index d0eef68d2..4de2bf857 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -182,6 +182,20 @@ function describeStreamingTests(createConnectedDatabase: () => Promise= 0; i--) { + await new Promise((r) => setTimeout(r, Math.random() * 100)); + powersync.connect(new TestConnector(), { params: { count: i } }); + } + + await vi.waitFor( + () => { + const call = spy.mock.lastCall![1] as PowerSyncConnectionOptions; + expect(call.params!['count']).eq(0); + }, + { timeout: 2000, interval: 100 } + ); }); it('Should trigger upload connector when connected', async () => { diff --git a/packages/web/tests/utils/MockStreamOpenFactory.ts b/packages/web/tests/utils/MockStreamOpenFactory.ts index e29cf313b..cc99c0bb7 100644 --- a/packages/web/tests/utils/MockStreamOpenFactory.ts +++ b/packages/web/tests/utils/MockStreamOpenFactory.ts @@ -81,7 +81,9 @@ export class MockRemote extends AbstractRemote { if (this.errorOnStreamStart) { controller.error(new Error('Mock error on stream start')); } - + if (signal?.aborted) { + controller.close(); + } signal?.addEventListener('abort', () => { try { controller.close(); diff --git a/packages/web/vitest.config.ts b/packages/web/vitest.config.ts index 42f7d5e3c..6c125b660 100644 --- a/packages/web/vitest.config.ts +++ b/packages/web/vitest.config.ts @@ -50,7 +50,7 @@ const config: UserConfigExport = { */ isolate: true, provider: 'playwright', - headless: true, + headless: false, instances: [ { browser: 'chromium' From f5df9dcbf77e1dd2a2aa6107d770ce04bc2989c2 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 26 May 2025 13:24:30 +0200 Subject: [PATCH 04/26] cleanup logic --- .../src/client/AbstractPowerSyncDatabase.ts | 49 ++++++++++++++---- .../AbstractStreamingSyncImplementation.ts | 24 ++++----- packages/web/tests/stream.test.ts | 51 +++++++++++++++++-- .../web/tests/utils/MockStreamOpenFactory.ts | 5 ++ 4 files changed, 101 insertions(+), 28 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index e14687791..a20023aee 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -464,6 +464,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { + // Disconnecting here provides aborting in progress connection attempts. + // The connectInternal method will clear pending options once it starts connecting (with the options). + // We only need to trigger a disconnect here if we have already reached the point of connecting. + // If we do already have pending options, a disconnect has already been performed. + // The connectInternal method also does a sanity disconnect to prevent straggler connections. + if (!hadPendingOptions) { + await this.disconnectInternal(); + } + + // Triggers a connect which checks if pending options are available after the connect completes. + // The completion can be for a successful, unsuccessful or aborted connection attempt. + // If pending options are available another connection will be triggered. + const checkConnection = async (): Promise => { if (this.pendingConnectionOptions) { - return this.connectInternal().then(chain); + // Pending options have been placed while connecting. + // Need to reconnect. + this.connectingPromise = this.connectInternal().finally(checkConnection); + return this.connectingPromise; } else { + // Clear the connecting promise, done. this.connectingPromise = null; - return result; + return; } }; - return this.connectingPromise ?? this.connectInternal().then(chain); + this.connectingPromise ??= this.connectInternal().finally(checkConnection); + return this.connectingPromise; } /** @@ -548,10 +576,11 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 4286a0693..488d0495d 100644 --- a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -347,16 +347,7 @@ The next upload iteration will be delayed.`); // Return a promise that resolves when the connection status is updated return new Promise((resolve) => { - let disposer: () => void; - - const complete = () => { - disposer?.(); - resolve(); - }; - - controller.signal.addEventListener('abort', complete, { once: true }); - - disposer = this.registerListener({ + const disposer = this.registerListener({ statusUpdated: (update) => { // This is triggered as soon as a connection is read from if (typeof update.connected == 'undefined') { @@ -366,13 +357,15 @@ The next upload iteration will be delayed.`); if (update.connected == false) { /** - * This function does not reject if initial connect attempt failed + * This function does not reject if initial connect attempt failed. + * Connected can be false if the connection attempt was aborted or if the initial connection + * attempt failed. */ this.logger.warn('Initial connect attempt did not successfully connect to server'); } - controller.signal.removeEventListener('abort', complete); - complete(); + disposer(); + resolve(); } }); }); @@ -530,6 +523,10 @@ The next upload iteration will be delayed.`); const clientId = await this.options.adapter.getClientId(); + if (signal.aborted) { + return; + } + this.logger.debug('Requesting stream from server'); const syncOptions: SyncStreamOptions = { @@ -557,6 +554,7 @@ The next upload iteration will be delayed.`); this.logger.debug('Stream established. Processing events'); while (!stream.closed) { + console.log('waiting for stream line'); const line = await stream.read(); if (!line) { // The stream has closed while waiting diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 4de2bf857..1132a0f08 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -1,4 +1,10 @@ -import { BucketChecksum, PowerSyncConnectionOptions, WASQLiteOpenFactory, WASQLiteVFS } from '@powersync/web'; +import { + BucketChecksum, + DataStream, + PowerSyncConnectionOptions, + WASQLiteOpenFactory, + WASQLiteVFS +} from '@powersync/web'; import { describe, expect, it, onTestFinished, vi } from 'vitest'; import { TestConnector } from './utils/MockStreamOpenFactory'; import { ConnectedDatabaseUtils, generateConnectedDatabase } from './utils/generateConnectedDatabase'; @@ -160,12 +166,25 @@ function describeStreamingTests(createConnectedDatabase: () => Promise { // This initially performs a connect call - const { powersync, waitForStream } = await createConnectedDatabase(); + const { powersync, remote } = await createConnectedDatabase(); expect(powersync.connected).toBe(true); const spy = vi.spyOn(powersync as any, 'generateSyncStreamImplementation'); - // connect many times + // Keep track of all connection stream to check if they are correctly closed later + const generatedStreams: DataStream[] = []; + + // This method is used for all mocked connections + const basePostStream = remote.postStream; + const postSpy = vi.spyOn(remote, 'postStream').mockImplementation(async (...options) => { + // Simulate a connection delay + await new Promise((r) => setTimeout(r, 100)); + const stream = await basePostStream.call(remote, ...options); + generatedStreams.push(stream); + return stream; + }); + + // Connect many times. The calls here are not awaited and have no async calls in between. const connectionAttempts = 10; for (let i = 1; i <= connectionAttempts; i++) { powersync.connect(new TestConnector(), { params: { count: i } }); @@ -183,9 +202,9 @@ function describeStreamingTests(createConnectedDatabase: () => Promise= 0; i--) { - await new Promise((r) => setTimeout(r, Math.random() * 100)); + await new Promise((r) => setTimeout(r, Math.random() * 10)); powersync.connect(new TestConnector(), { params: { count: i } }); } @@ -196,6 +215,28 @@ function describeStreamingTests(createConnectedDatabase: () => Promise { + expect(postSpy.mock.lastCall?.[0].data.parameters!['count']).equals(0); + // The async postStream call's invocation is added to the count of calls + // before the generated stream is added (there is a delay) + // expect that the stream has been generated and tracked. + expect(postSpy.mock.calls.length).equals(generatedStreams.length); + }, + { timeout: 1000, interval: 100 } + ); + + const lastConnectionStream = generatedStreams.pop(); + expect(lastConnectionStream).toBeDefined(); + expect(lastConnectionStream?.closed).false; + + // All streams except the last one (which has been popped off already) should be closed + expect(generatedStreams.every((i) => i.closed)).true; }); it('Should trigger upload connector when connected', async () => { diff --git a/packages/web/tests/utils/MockStreamOpenFactory.ts b/packages/web/tests/utils/MockStreamOpenFactory.ts index cc99c0bb7..9a02e1be8 100644 --- a/packages/web/tests/utils/MockStreamOpenFactory.ts +++ b/packages/web/tests/utils/MockStreamOpenFactory.ts @@ -108,6 +108,11 @@ export class MockRemote extends AbstractRemote { logger: this.logger }); + if (options.abortSignal?.aborted) { + stream.close(); + } + options.abortSignal?.addEventListener('abort', () => stream.close()); + const l = stream.registerListener({ lowWater: async () => { try { From be6b2b430ce798409c7041e93414b739302f300e Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 26 May 2025 13:37:12 +0200 Subject: [PATCH 05/26] cleanup --- packages/common/rollup.config.mjs | 5 +-- .../src/client/AbstractPowerSyncDatabase.ts | 34 +++++++++---------- 2 files changed, 19 insertions(+), 20 deletions(-) diff --git a/packages/common/rollup.config.mjs b/packages/common/rollup.config.mjs index 4e64e9212..ad05faf6a 100644 --- a/packages/common/rollup.config.mjs +++ b/packages/common/rollup.config.mjs @@ -2,6 +2,7 @@ import commonjs from '@rollup/plugin-commonjs'; import inject from '@rollup/plugin-inject'; import json from '@rollup/plugin-json'; import nodeResolve from '@rollup/plugin-node-resolve'; +import terser from '@rollup/plugin-terser'; export default (commandLineArgs) => { const sourcemap = (commandLineArgs.sourceMap || 'true') == 'true'; @@ -25,8 +26,8 @@ export default (commandLineArgs) => { ReadableStream: ['web-streams-polyfill/ponyfill', 'ReadableStream'], // Used by can-ndjson-stream TextDecoder: ['text-encoding', 'TextDecoder'] - }) - // terser() + }), + terser() ], // This makes life easier external: [ diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index a20023aee..fdef56d14 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -190,7 +190,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver | null; /** - * Active disconnect operation. Call disconnect multiple times + * Active disconnect operation. Calling disconnect multiple times * will resolve to the same operation. */ protected disconnectingPromise: Promise | null; @@ -457,7 +457,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver(callback: () => Promise): Promise; @@ -573,6 +574,13 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { - await this.syncStreamImplementation?.disconnect(); - this.syncStatusListenerDisposer?.(); - await this.syncStreamImplementation?.dispose(); - this.syncStreamImplementation = undefined; - })(); + this.disconnectingPromise = this.performDisconnect(); await this.disconnectingPromise; this.disconnectingPromise = null; } - /** - * Close the sync connection. - * - * Use {@link connect} to connect again. - */ - async disconnect() { - await this.waitForReady(); - // This will help abort pending connects - this.pendingConnectionOptions = null; - await this.disconnectInternal(); + protected async performDisconnect() { + await this.syncStreamImplementation?.disconnect(); + this.syncStatusListenerDisposer?.(); + await this.syncStreamImplementation?.dispose(); + this.syncStreamImplementation = undefined; } /** From 828fae9dc811f94e83ec0f32ce2054d447263d07 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 26 May 2025 13:40:00 +0200 Subject: [PATCH 06/26] cleanup --- .../client/sync/stream/AbstractStreamingSyncImplementation.ts | 1 - packages/web/tests/stream.test.ts | 2 +- packages/web/vitest.config.ts | 2 +- 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 4433a35a6..61d298c2c 100644 --- a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -554,7 +554,6 @@ The next upload iteration will be delayed.`); this.logger.debug('Stream established. Processing events'); while (!stream.closed) { - console.log('waiting for stream line'); const line = await stream.read(); if (!line) { // The stream has closed while waiting diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 1132a0f08..19d374c33 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -171,7 +171,7 @@ function describeStreamingTests(createConnectedDatabase: () => Promise[] = []; // This method is used for all mocked connections diff --git a/packages/web/vitest.config.ts b/packages/web/vitest.config.ts index 6c125b660..42f7d5e3c 100644 --- a/packages/web/vitest.config.ts +++ b/packages/web/vitest.config.ts @@ -50,7 +50,7 @@ const config: UserConfigExport = { */ isolate: true, provider: 'playwright', - headless: false, + headless: true, instances: [ { browser: 'chromium' From 8a911b7b7707b1aaf5c5dca5572fe5d7831d33d9 Mon Sep 17 00:00:00 2001 From: stevensJourney <51082125+stevensJourney@users.noreply.github.com> Date: Mon, 26 May 2025 14:31:47 +0200 Subject: [PATCH 07/26] Update packages/web/tests/stream.test.ts Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- packages/web/tests/stream.test.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 19d374c33..842bd5585 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -216,8 +216,7 @@ function describeStreamingTests(createConnectedDatabase: () => Promise Date: Mon, 26 May 2025 14:33:35 +0200 Subject: [PATCH 08/26] cleanup common rollup --- packages/common/rollup.config.mjs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/packages/common/rollup.config.mjs b/packages/common/rollup.config.mjs index ad05faf6a..1fc8fedd9 100644 --- a/packages/common/rollup.config.mjs +++ b/packages/common/rollup.config.mjs @@ -4,8 +4,11 @@ import json from '@rollup/plugin-json'; import nodeResolve from '@rollup/plugin-node-resolve'; import terser from '@rollup/plugin-terser'; +/** + * @returns {import('rollup').RollupOptions} + */ export default (commandLineArgs) => { - const sourcemap = (commandLineArgs.sourceMap || 'true') == 'true'; + const sourceMap = (commandLineArgs.sourceMap || 'true') == 'true'; // Clears rollup CLI warning https://github.com/rollup/rollup/issues/2694 delete commandLineArgs.sourceMap; @@ -15,7 +18,7 @@ export default (commandLineArgs) => { output: { file: 'dist/bundle.mjs', format: 'esm', - sourcemap: sourcemap + sourcemap: sourceMap }, plugins: [ json(), @@ -27,7 +30,7 @@ export default (commandLineArgs) => { // Used by can-ndjson-stream TextDecoder: ['text-encoding', 'TextDecoder'] }), - terser() + terser({ sourceMap }) ], // This makes life easier external: [ From b234bca9c499f51df8576388ddc80e1be34d5c20 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 26 May 2025 14:35:08 +0200 Subject: [PATCH 09/26] cleanup listener --- packages/web/tests/utils/MockStreamOpenFactory.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/web/tests/utils/MockStreamOpenFactory.ts b/packages/web/tests/utils/MockStreamOpenFactory.ts index 9a02e1be8..d4d32c39f 100644 --- a/packages/web/tests/utils/MockStreamOpenFactory.ts +++ b/packages/web/tests/utils/MockStreamOpenFactory.ts @@ -83,6 +83,7 @@ export class MockRemote extends AbstractRemote { } if (signal?.aborted) { controller.close(); + return; } signal?.addEventListener('abort', () => { try { From 308bef4b0ade3ce1e7bfb18c70c900a00766ebf3 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 26 May 2025 14:38:21 +0200 Subject: [PATCH 10/26] cleanup --- packages/common/src/client/AbstractPowerSyncDatabase.ts | 7 ++++++- packages/node/src/db/PowerSyncDatabase.ts | 8 -------- packages/react-native/src/db/PowerSyncDatabase.ts | 7 ------- 3 files changed, 6 insertions(+), 16 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index fdef56d14..ac8ed45de 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -204,6 +204,8 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver(callback: () => Promise): Promise; + protected runExclusive(callback: () => Promise): Promise { + return this.connectionMutex.runExclusive(callback); + } protected async connectInternal() { let appliedOptions: PowerSyncConnectionOptions | null = null; diff --git a/packages/node/src/db/PowerSyncDatabase.ts b/packages/node/src/db/PowerSyncDatabase.ts index 80f05b30b..0c4501678 100644 --- a/packages/node/src/db/PowerSyncDatabase.ts +++ b/packages/node/src/db/PowerSyncDatabase.ts @@ -16,7 +16,6 @@ import { import { NodeRemote } from '../sync/stream/NodeRemote.js'; import { NodeStreamingSyncImplementation } from '../sync/stream/NodeStreamingSyncImplementation.js'; -import Lock from 'async-lock'; import { Dispatcher } from 'undici'; import { BetterSQLite3DBAdapter } from './BetterSQLite3DBAdapter.js'; import { NodeSQLOpenOptions } from './options.js'; @@ -56,11 +55,8 @@ export type NodePowerSyncConnectionOptions = PowerSyncConnectionOptions & NodeAd * ``` */ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { - protected lock: Lock; - constructor(options: NodePowerSyncDatabaseOptions) { super(options); - this.lock = new Lock(); } async _initialize(): Promise { @@ -85,10 +81,6 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return super.connect(connector, options); } - protected async runExclusive(callback: () => Promise): Promise { - return await this.lock.acquire(`lock-${this.database.name}`, callback); - } - protected generateSyncStreamImplementation( connector: PowerSyncBackendConnector, options: NodeAdditionalConnectionOptions diff --git a/packages/react-native/src/db/PowerSyncDatabase.ts b/packages/react-native/src/db/PowerSyncDatabase.ts index a68957a13..3d37a7e9b 100644 --- a/packages/react-native/src/db/PowerSyncDatabase.ts +++ b/packages/react-native/src/db/PowerSyncDatabase.ts @@ -8,7 +8,6 @@ import { type RequiredAdditionalConnectionOptions, SqliteBucketStorage } from '@powersync/common'; -import Lock from 'async-lock'; import { ReactNativeRemote } from '../sync/stream/ReactNativeRemote'; import { ReactNativeStreamingSyncImplementation } from '../sync/stream/ReactNativeStreamingSyncImplementation'; import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick-sqlite/ReactNativeQuickSQLiteOpenFactory'; @@ -28,8 +27,6 @@ import { ReactNativeQuickSqliteOpenFactory } from './adapters/react-native-quick * ``` */ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { - protected lock = new Lock(); - async _initialize(): Promise {} /** @@ -45,10 +42,6 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex); } - protected async runExclusive(callback: () => Promise): Promise { - return await this.lock.acquire(`lock-${this.database.name}`, callback); - } - protected generateSyncStreamImplementation( connector: PowerSyncBackendConnector, options: RequiredAdditionalConnectionOptions From e561dda980acb116f5302430e3d73631561ef85b Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 28 May 2025 12:55:41 +0200 Subject: [PATCH 11/26] use shared ConnectionManager in order to queue connections for multipletab scenarios --- .../src/client/AbstractPowerSyncDatabase.ts | 195 +++---------- .../common/src/client/ConnectionManager.ts | 215 ++++++++++++++ packages/common/src/index.ts | 27 +- .../SharedWebStreamingSyncImplementation.ts | 26 +- .../worker/sync/SharedSyncImplementation.ts | 267 ++++++++++-------- .../sync/SharedSyncImplementation.worker.ts | 17 +- packages/web/tests/stream.test.ts | 5 +- 7 files changed, 448 insertions(+), 304 deletions(-) create mode 100644 packages/common/src/client/ConnectionManager.ts diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index ac8ed45de..408e3c85b 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -17,6 +17,7 @@ import { BaseObserver } from '../utils/BaseObserver.js'; import { ControlledExecutor } from '../utils/ControlledExecutor.js'; import { throttleTrailing } from '../utils/async.js'; import { mutexRunExclusive } from '../utils/mutex.js'; +import { ConnectionManager } from './ConnectionManager.js'; import { SQLOpenFactory, SQLOpenOptions, isDBAdapter, isSQLOpenFactory, isSQLOpenOptions } from './SQLOpenFactory.js'; import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector.js'; import { runOnSchemaChange } from './runOnSchemaChange.js'; @@ -112,11 +113,6 @@ export interface PowerSyncCloseOptions { disconnect?: boolean; } -type StoredConnectionOptions = { - connector: PowerSyncBackendConnector; - options: PowerSyncConnectionOptions; -}; - const POWERSYNC_TABLE_MATCH = /(^ps_data__|^ps_data_local__)/; const DEFAULT_DISCONNECT_CLEAR_OPTIONS: DisconnectAndClearOptions = { @@ -170,40 +166,20 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver void; protected _isReadyPromise: Promise; + protected connectionManager: ConnectionManager; + + get syncStreamImplementation() { + return this.connectionManager.syncStreamImplementation; + } protected _schema: Schema; private _database: DBAdapter; - /** - * Tracks active connection attempts - */ - protected connectingPromise: Promise | null; - /** - * Tracks actively instantiating a streaming sync implementation. - */ - protected syncStreamInitPromise: Promise | null; - /** - * Active disconnect operation. Calling disconnect multiple times - * will resolve to the same operation. - */ - protected disconnectingPromise: Promise | null; - /** - * Tracks the last parameters supplied to `connect` calls. - * Calling `connect` multiple times in succession will result in: - * - 1 pending connection operation which will be aborted. - * - updating the last set of parameters while waiting for the pending - * attempt to be aborted - * - internally connecting with the last set of parameters - */ - protected pendingConnectionOptions: StoredConnectionOptions | null; - protected connectionMutex: Mutex; constructor(options: PowerSyncDatabaseOptionsWithDBAdapter); @@ -236,11 +212,33 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { + await this.waitForReady(); + + return this.runExclusive(async () => { + const sync = this.generateSyncStreamImplementation(connector, this.resolvedConnectionOptions(options)); + const onDispose = sync.registerListener({ + statusChanged: (status) => { + this.currentStatus = new SyncStatus({ + ...status.toJSON(), + hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt + }); + this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus)); + } + }); + await sync.waitForReady(); + + return { + sync, + onDispose + }; + }); + }, + logger: this.logger + }); this._isReadyPromise = this.initialize(); } @@ -467,111 +465,11 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { - if (this.closed) { - throw new Error('Cannot connect using a closed client'); - } - - // Always await this if present since we will be populating a new sync implementation shortly - await this.disconnectingPromise; - - if (!this.pendingConnectionOptions) { - // A disconnect could have cleared this. - return; - } - // get pending options and clear it in order for other connect attempts to queue other options - const { connector, options } = this.pendingConnectionOptions; - appliedOptions = options; - this.pendingConnectionOptions = null; - - this.syncStreamImplementation = this.generateSyncStreamImplementation( - connector, - this.resolvedConnectionOptions(options) - ); - this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({ - statusChanged: (status) => { - this.currentStatus = new SyncStatus({ - ...status.toJSON(), - hasSynced: this.currentStatus?.hasSynced || !!status.lastSyncedAt - }); - this.iterateListeners((cb) => cb.statusChanged?.(this.currentStatus)); - } - }); - - await this.syncStreamImplementation.waitForReady(); - }); - - await this.syncStreamInitPromise; - this.syncStreamInitPromise = null; - - if (!appliedOptions) { - // A disconnect could have cleared the options which did not create a syncStreamImplementation - return; - } - - // It might be possible that a disconnect triggered between the last check - // and this point. Awaiting here allows the sync stream to be cleared if disconnected. - await this.disconnectingPromise; - - this.syncStreamImplementation?.triggerCrudUpload(); - this.options.logger?.debug('Attempting to connect to PowerSync instance'); - await this.syncStreamImplementation?.connect(appliedOptions!); - } - /** * Connects to stream of events from the PowerSync instance. */ async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) { - // Keep track if there were pending operations before this call - const hadPendingOptions = !!this.pendingConnectionOptions; - - // Update pending options to the latest values - this.pendingConnectionOptions = { - connector, - options: options ?? {} - }; - - await this.waitForReady(); - - // Disconnecting here provides aborting in progress connection attempts. - // The connectInternal method will clear pending options once it starts connecting (with the options). - // We only need to trigger a disconnect here if we have already reached the point of connecting. - // If we do already have pending options, a disconnect has already been performed. - // The connectInternal method also does a sanity disconnect to prevent straggler connections. - if (!hadPendingOptions) { - await this.disconnectInternal(); - } - - // Triggers a connect which checks if pending options are available after the connect completes. - // The completion can be for a successful, unsuccessful or aborted connection attempt. - // If pending options are available another connection will be triggered. - const checkConnection = async (): Promise => { - if (this.pendingConnectionOptions) { - // Pending options have been placed while connecting. - // Need to reconnect. - this.connectingPromise = this.connectInternal().finally(checkConnection); - return this.connectingPromise; - } else { - // Clear the connecting promise, done. - this.connectingPromise = null; - return; - } - }; - - this.connectingPromise ??= this.connectInternal().finally(checkConnection); - return this.connectingPromise; + return this.connectionManager.connect(connector, options); } /** @@ -581,32 +479,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver Promise | void; +} + +/** + * @internal + */ +export interface ConnectionManagerOptions { + createSyncImplementation( + connector: PowerSyncBackendConnector, + options: PowerSyncConnectionOptions + ): Promise; + logger: ILogger; +} + +type StoredConnectionOptions = { + connector: PowerSyncBackendConnector; + options: PowerSyncConnectionOptions; +}; + +/** + * @internal + */ +export interface ConnectionManagerListener extends BaseListener { + syncStreamCreated: (sync: StreamingSyncImplementation) => void; +} + +/** + * @internal + */ +export class ConnectionManager extends BaseObserver { + /** + * Tracks active connection attempts + */ + protected connectingPromise: Promise | null; + /** + * Tracks actively instantiating a streaming sync implementation. + */ + protected syncStreamInitPromise: Promise | null; + /** + * Active disconnect operation. Calling disconnect multiple times + * will resolve to the same operation. + */ + protected disconnectingPromise: Promise | null; + /** + * Tracks the last parameters supplied to `connect` calls. + * Calling `connect` multiple times in succession will result in: + * - 1 pending connection operation which will be aborted. + * - updating the last set of parameters while waiting for the pending + * attempt to be aborted + * - internally connecting with the last set of parameters + */ + protected pendingConnectionOptions: StoredConnectionOptions | null; + + syncStreamImplementation: StreamingSyncImplementation | null; + syncDisposer: (() => Promise | void) | null; + + constructor(protected options: ConnectionManagerOptions) { + super(); + this.connectingPromise = null; + this.syncStreamInitPromise = null; + this.disconnectingPromise = null; + this.pendingConnectionOptions = null; + this.syncStreamImplementation = null; + this.syncDisposer = null; + } + + get logger() { + return this.options.logger; + } + + async close() { + await this.syncDisposer?.(); + } + + async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) { + // Keep track if there were pending operations before this call + const hadPendingOptions = !!this.pendingConnectionOptions; + + // Update pending options to the latest values + this.pendingConnectionOptions = { + connector, + options: options ?? {} + }; + + // Disconnecting here provides aborting in progress connection attempts. + // The connectInternal method will clear pending options once it starts connecting (with the options). + // We only need to trigger a disconnect here if we have already reached the point of connecting. + // If we do already have pending options, a disconnect has already been performed. + // The connectInternal method also does a sanity disconnect to prevent straggler connections. + if (!hadPendingOptions) { + await this.disconnectInternal(); + } + + // Triggers a connect which checks if pending options are available after the connect completes. + // The completion can be for a successful, unsuccessful or aborted connection attempt. + // If pending options are available another connection will be triggered. + const checkConnection = async (): Promise => { + if (this.pendingConnectionOptions) { + // Pending options have been placed while connecting. + // Need to reconnect. + this.connectingPromise = this.connectInternal().finally(checkConnection); + return this.connectingPromise; + } else { + // Clear the connecting promise, done. + this.connectingPromise = null; + return; + } + }; + + this.connectingPromise ??= this.connectInternal().finally(checkConnection); + return this.connectingPromise; + } + + protected async connectInternal() { + let appliedOptions: PowerSyncConnectionOptions | null = null; + + // This method ensures a disconnect before any connection attempt + await this.disconnectInternal(); + + /** + * This portion creates a sync implementation which can be racy when disconnecting or + * if multiple tabs on web are in use. + * This is protected in an exclusive lock. + * The promise tracks the creation which is used to synchronize disconnect attempts. + */ + this.syncStreamInitPromise = (async () => { + // Always await this if present since we will be populating a new sync implementation shortly + await this.disconnectingPromise; + + if (!this.pendingConnectionOptions) { + // A disconnect could have cleared this. + return; + } + + const { connector, options } = this.pendingConnectionOptions; + appliedOptions = options; + + this.pendingConnectionOptions = null; + + const { sync, onDispose } = await this.options.createSyncImplementation(connector, options); + this.iterateListeners((l) => l.syncStreamCreated?.(sync)); + this.syncStreamImplementation = sync; + this.syncDisposer = onDispose; + await this.syncStreamImplementation.waitForReady(); + })(); + + await this.syncStreamInitPromise; + this.syncStreamInitPromise = null; + + if (!appliedOptions) { + this.logger.debug('No pending connection options found, not connecting'); + // A disconnect could have cleared the options which did not create a syncStreamImplementation + return; + } + + // It might be possible that a disconnect triggered between the last check + // and this point. Awaiting here allows the sync stream to be cleared if disconnected. + this.logger.debug('Waiting for disconnect to complete before connecting', this.disconnectingPromise); + await this.disconnectingPromise; + this.logger.debug('Disconnect completed, proceeding to connect'); + + this.syncStreamImplementation?.triggerCrudUpload(); + this.logger.debug('Attempting to connect to PowerSync instance'); + await this.syncStreamImplementation?.connect(appliedOptions!); + this.logger.debug('connected to sync stream implementation'); + } + + /** + * Close the sync connection. + * + * Use {@link connect} to connect again. + */ + async disconnect() { + // This will help abort pending connects + this.pendingConnectionOptions = null; + await this.disconnectInternal(); + } + + protected async disconnectInternal(): Promise { + if (this.disconnectingPromise) { + // A disconnect is already in progress + return this.disconnectingPromise; + } + + // Wait if a sync stream implementation is being created before closing it + // (syncStreamImplementation must be assigned before we can properly dispose it) + await this.syncStreamInitPromise; + + this.disconnectingPromise = this.performDisconnect(); + + await this.disconnectingPromise; + this.disconnectingPromise = null; + } + + protected async performDisconnect() { + await this.syncStreamImplementation?.disconnect(); + await this.syncStreamImplementation?.dispose(); + await this.syncDisposer?.(); + this.syncDisposer = null; + this.syncStreamImplementation = null; + } +} diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts index 96fdb08c1..44203a2b9 100644 --- a/packages/common/src/index.ts +++ b/packages/common/src/index.ts @@ -1,36 +1,35 @@ export * from './client/AbstractPowerSyncDatabase.js'; export * from './client/AbstractPowerSyncOpenFactory.js'; -export * from './client/SQLOpenFactory.js'; +export { compilableQueryWatch, CompilableQueryWatchHandler } from './client/compilableQueryWatch.js'; export * from './client/connection/PowerSyncBackendConnector.js'; export * from './client/connection/PowerSyncCredentials.js'; -export * from './client/sync/bucket/BucketStorageAdapter.js'; +export { MAX_OP_ID } from './client/constants.js'; export { runOnSchemaChange } from './client/runOnSchemaChange.js'; -export { CompilableQueryWatchHandler, compilableQueryWatch } from './client/compilableQueryWatch.js'; -export { UpdateType, CrudEntry, OpId } from './client/sync/bucket/CrudEntry.js'; -export * from './client/sync/bucket/SqliteBucketStorage.js'; +export * from './client/SQLOpenFactory.js'; +export * from './client/sync/bucket/BucketStorageAdapter.js'; export * from './client/sync/bucket/CrudBatch.js'; +export { CrudEntry, OpId, UpdateType } from './client/sync/bucket/CrudEntry.js'; export * from './client/sync/bucket/CrudTransaction.js'; +export * from './client/sync/bucket/OplogEntry.js'; +export * from './client/sync/bucket/OpType.js'; +export * from './client/sync/bucket/SqliteBucketStorage.js'; export * from './client/sync/bucket/SyncDataBatch.js'; export * from './client/sync/bucket/SyncDataBucket.js'; -export * from './client/sync/bucket/OpType.js'; -export * from './client/sync/bucket/OplogEntry.js'; export * from './client/sync/stream/AbstractRemote.js'; export * from './client/sync/stream/AbstractStreamingSyncImplementation.js'; export * from './client/sync/stream/streaming-sync-types.js'; -export { MAX_OP_ID } from './client/constants.js'; +export * from './client/ConnectionManager.js'; export { ProgressWithOperations, SyncProgress } from './db/crud/SyncProgress.js'; export * from './db/crud/SyncStatus.js'; export * from './db/crud/UploadQueueStatus.js'; -export * from './db/schema/Schema.js'; -export * from './db/schema/Table.js'; +export * from './db/DBAdapter.js'; +export * from './db/schema/Column.js'; export * from './db/schema/Index.js'; export * from './db/schema/IndexedColumn.js'; -export * from './db/schema/Column.js'; +export * from './db/schema/Schema.js'; +export * from './db/schema/Table.js'; export * from './db/schema/TableV2.js'; -export * from './db/crud/SyncStatus.js'; -export * from './db/crud/UploadQueueStatus.js'; -export * from './db/DBAdapter.js'; export * from './utils/AbortOperation.js'; export * from './utils/BaseObserver.js'; diff --git a/packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts b/packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts index 3dad3192b..dd059139b 100644 --- a/packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts +++ b/packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts @@ -187,9 +187,6 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem */ async connect(options?: PowerSyncConnectionOptions): Promise { await this.waitForReady(); - // This is needed since a new tab won't have any reference to the - // shared worker sync implementation since that is only created on the first call to `connect`. - await this.disconnect(); return this.syncManager.connect(options); } @@ -209,13 +206,24 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem async dispose(): Promise { await this.waitForReady(); - // Signal the shared worker that this client is closing its connection to the worker - const closeMessagePayload: ManualSharedSyncPayload = { - event: SharedSyncClientEvent.CLOSE_CLIENT, - data: {} - }; - this.messagePort.postMessage(closeMessagePayload); + await new Promise((resolve) => { + // Listen for the close acknowledgment from the worker + this.messagePort.addEventListener('message', (event) => { + const payload = event.data as ManualSharedSyncPayload; + if (payload?.event === SharedSyncClientEvent.CLOSE_ACK) { + resolve(); + } + }); + + // Signal the shared worker that this client is closing its connection to the worker + const closeMessagePayload: ManualSharedSyncPayload = { + event: SharedSyncClientEvent.CLOSE_CLIENT, + data: {} + }; + this.messagePort.postMessage(closeMessagePayload); + }); + // Release the proxy this.syncManager[Comlink.releaseProxy](); this.messagePort.close(); diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.ts b/packages/web/src/worker/sync/SharedSyncImplementation.ts index 60e2e1c0f..616319365 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.ts @@ -1,16 +1,16 @@ import { - type AbstractStreamingSyncImplementation, type ILogger, type ILogLevel, - type LockOptions, type PowerSyncConnectionOptions, type StreamingSyncImplementation, type StreamingSyncImplementationListener, type SyncStatusOptions, AbortOperation, BaseObserver, + ConnectionManager, createLogger, DBAdapter, + PowerSyncBackendConnector, SqliteBucketStorage, SyncStatus } from '@powersync/common'; @@ -26,7 +26,6 @@ import { OpenAsyncDatabaseConnection } from '../../db/adapters/AsyncDatabaseConn import { LockedAsyncDatabaseAdapter } from '../../db/adapters/LockedAsyncDatabaseAdapter'; import { ResolvedWebSQLOpenOptions } from '../../db/adapters/web-sql-flags'; import { WorkerWrappedAsyncDatabaseConnection } from '../../db/adapters/WorkerWrappedAsyncDatabaseConnection'; -import { getNavigatorLocks } from '../../shared/navigator'; import { AbstractSharedSyncClientProvider } from './AbstractSharedSyncClientProvider'; import { BroadcastLogger } from './BroadcastLogger'; @@ -38,7 +37,9 @@ export enum SharedSyncClientEvent { * This client requests the shared sync manager should * close it's connection to the client. */ - CLOSE_CLIENT = 'close-client' + CLOSE_CLIENT = 'close-client', + + CLOSE_ACK = 'close-ack' } export type ManualSharedSyncPayload = { @@ -78,6 +79,12 @@ export type RemoteOperationAbortController = { activePort: WrappedSyncPort; }; +/** + * HACK: The shared implementation wraps and provides its own + * PowerSyncBackendConnector when generating the streaming sync implementation. + * We provide this unused placeholder when connecting with the ConnectionManager. + */ +const CONNECTOR_PLACEHOLDER = {} as PowerSyncBackendConnector; /** * @internal * Shared sync implementation which runs inside a shared webworker @@ -87,7 +94,6 @@ export class SharedSyncImplementation implements StreamingSyncImplementation { protected ports: WrappedSyncPort[]; - protected syncStreamClient: AbstractStreamingSyncImplementation | null; protected isInitialized: Promise; protected statusListener?: () => void; @@ -99,7 +105,9 @@ export class SharedSyncImplementation protected syncParams: SharedSyncInitOptions | null; protected logger: ILogger; protected lastConnectOptions: PowerSyncConnectionOptions | undefined; + protected portMutex: Mutex; + protected connectionManager: ConnectionManager; syncStatus: SyncStatus; broadCastLogger: ILogger; @@ -108,9 +116,9 @@ export class SharedSyncImplementation this.ports = []; this.dbAdapter = null; this.syncParams = null; - this.syncStreamClient = null; this.logger = createLogger('shared-sync'); this.lastConnectOptions = undefined; + this.portMutex = new Mutex(); this.isInitialized = new Promise((resolve) => { const callback = this.registerListener({ @@ -123,24 +131,48 @@ export class SharedSyncImplementation this.syncStatus = new SyncStatus({}); this.broadCastLogger = new BroadcastLogger(this.ports); - } - async waitForStatus(status: SyncStatusOptions): Promise { - await this.waitForReady(); - return this.syncStreamClient!.waitForStatus(status); - } + this.connectionManager = new ConnectionManager({ + createSyncImplementation: async (connector, options) => { + await this.waitForReady(); + if (!this.dbAdapter) { + await this.openInternalDB(); + } - async waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise { - await this.waitForReady(); - return this.syncStreamClient!.waitUntilStatusMatches(predicate); + const sync = this.generateStreamingImplementation(); + const onDispose = sync.registerListener({ + statusChanged: (status) => { + this.updateAllStatuses(status.toJSON()); + } + }); + + return { + sync, + onDispose + }; + }, + logger: this.logger + }); } get lastSyncedAt(): Date | undefined { - return this.syncStreamClient?.lastSyncedAt; + return this.connectionManager.syncStreamImplementation?.lastSyncedAt; } get isConnected(): boolean { - return this.syncStreamClient?.isConnected ?? false; + return this.connectionManager.syncStreamImplementation?.isConnected ?? false; + } + + async waitForStatus(status: SyncStatusOptions): Promise { + await this.waitForReady(); + // TODO + return this.connectionManager.syncStreamImplementation!.waitForStatus(status); + } + + async waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise { + await this.waitForReady(); + // TODO + return this.connectionManager.syncStreamImplementation!.waitUntilStatusMatches(predicate); } async waitForReady() { @@ -156,30 +188,38 @@ export class SharedSyncImplementation * Configures the DBAdapter connection and a streaming sync client. */ async setParams(params: SharedSyncInitOptions) { - if (this.syncParams) { - // Cannot modify already existing sync implementation - return; - } + await this.portMutex.runExclusive(async () => { + if (this.syncParams) { + if (!this.dbAdapter) { + await this.openInternalDB(); + } + // Cannot modify already existing sync implementation + return; + } - this.syncParams = params; + this.syncParams = params; - if (params.streamOptions?.flags?.broadcastLogs) { - this.logger = this.broadCastLogger; - } + if (params.streamOptions?.flags?.broadcastLogs) { + this.logger = this.broadCastLogger; + } - self.onerror = (event) => { - // Share any uncaught events on the broadcast logger - this.logger.error('Uncaught exception in PowerSync shared sync worker', event); - }; + self.onerror = (event) => { + // Share any uncaught events on the broadcast logger + this.logger.error('Uncaught exception in PowerSync shared sync worker', event); + }; - await this.openInternalDB(); - this.iterateListeners((l) => l.initialized?.()); + if (!this.dbAdapter) { + await this.openInternalDB(); + } + + this.iterateListeners((l) => l.initialized?.()); + }); } async dispose() { await this.waitForReady(); this.statusListener?.(); - return this.syncStreamClient?.dispose(); + return this.connectionManager.close(); } /** @@ -189,49 +229,36 @@ export class SharedSyncImplementation * connects. */ async connect(options?: PowerSyncConnectionOptions) { - await this.waitForReady(); - // This effectively queues connect and disconnect calls. Ensuring multiple tabs' requests are synchronized - return getNavigatorLocks().request('shared-sync-connect', async () => { - if (!this.dbAdapter) { - await this.openInternalDB(); - } - this.syncStreamClient = this.generateStreamingImplementation(); + await this.portMutex.runExclusive(async () => { + // Keep track of the last connect options if we need to reconnect due to a lost client this.lastConnectOptions = options; - this.syncStreamClient.registerListener({ - statusChanged: (status) => { - this.updateAllStatuses(status.toJSON()); - } - }); - - await this.syncStreamClient.connect(options); + return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options); }); } async disconnect() { - await this.waitForReady(); - // This effectively queues connect and disconnect calls. Ensuring multiple tabs' requests are synchronized - return getNavigatorLocks().request('shared-sync-connect', async () => { - await this.syncStreamClient?.disconnect(); - await this.syncStreamClient?.dispose(); - this.syncStreamClient = null; + await this.portMutex.runExclusive(async () => { + await this.connectionManager.disconnect(); }); } /** * Adds a new client tab's message port to the list of connected ports */ - addPort(port: MessagePort) { - const portProvider = { - port, - clientProvider: Comlink.wrap(port) - }; - this.ports.push(portProvider); - - // Give the newly connected client the latest status - const status = this.syncStreamClient?.syncStatus; - if (status) { - portProvider.clientProvider.statusChanged(status.toJSON()); - } + async addPort(port: MessagePort) { + await this.portMutex.runExclusive(() => { + const portProvider = { + port, + clientProvider: Comlink.wrap(port) + }; + this.ports.push(portProvider); + + // Give the newly connected client the latest status + const status = this.connectionManager.syncStreamImplementation?.syncStatus; + if (status) { + portProvider.clientProvider.statusChanged(status.toJSON()); + } + }); } /** @@ -239,64 +266,86 @@ export class SharedSyncImplementation * clients. */ async removePort(port: MessagePort) { - const index = this.ports.findIndex((p) => p.port == port); - if (index < 0) { - this.logger.warn(`Could not remove port ${port} since it is not present in active ports.`); - return; - } - - const trackedPort = this.ports[index]; - // Remove from the list of active ports - this.ports.splice(index, 1); - - /** - * The port might currently be in use. Any active functions might - * not resolve. Abort them here. - */ - [this.fetchCredentialsController, this.uploadDataController].forEach((abortController) => { - if (abortController?.activePort.port == port) { - abortController!.controller.abort(new AbortOperation('Closing pending requests after client port is removed')); + return await this.portMutex.runExclusive(async () => { + const index = this.ports.findIndex((p) => p.port == port); + if (index < 0) { + this.logger.warn(`Could not remove port ${port} since it is not present in active ports.`); + return; } - }); - const shouldReconnect = !!this.syncStreamClient; - if (this.dbAdapter && this.dbAdapter == trackedPort.db) { - if (shouldReconnect) { - await this.disconnect(); - } + const trackedPort = this.ports[index]; + // Remove from the list of active ports + this.ports.splice(index, 1); + + /** + * The port might currently be in use. Any active functions might + * not resolve. Abort them here. + */ + [this.fetchCredentialsController, this.uploadDataController].forEach((abortController) => { + if (abortController?.activePort.port == port) { + abortController!.controller.abort( + new AbortOperation('Closing pending requests after client port is removed') + ); + } + }); + + const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0; + + if (this.dbAdapter && this.dbAdapter == trackedPort.db) { + if (shouldReconnect) { + await this.connectionManager.disconnect(); + } - // Clearing the adapter will result in a new one being opened in connect - this.dbAdapter = null; + // Clearing the adapter will result in a new one being opened in connect + this.dbAdapter = null; - if (shouldReconnect) { - await this.connect(this.lastConnectOptions); + if (shouldReconnect) { + await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions); + } } - } - if (trackedPort.db) { - trackedPort.db.close(); - } - // Release proxy - trackedPort.clientProvider[Comlink.releaseProxy](); + if (trackedPort.db) { + await trackedPort.db.close(); + } + this.logger.debug(`Port ${port} removed from shared sync implementation.`); + // Release proxy + return () => trackedPort.clientProvider[Comlink.releaseProxy](); + }); } triggerCrudUpload() { - this.waitForReady().then(() => this.syncStreamClient?.triggerCrudUpload()); - } - - async obtainLock(lockOptions: LockOptions): Promise { - await this.waitForReady(); - return this.syncStreamClient!.obtainLock(lockOptions); + this.waitForReady().then(() => this.connectionManager.syncStreamImplementation?.triggerCrudUpload()); } async hasCompletedSync(): Promise { - await this.waitForReady(); - return this.syncStreamClient!.hasCompletedSync(); + return this.withSyncImplementation(async (sync) => { + return sync.hasCompletedSync(); + }); } async getWriteCheckpoint(): Promise { + return this.withSyncImplementation(async (sync) => { + return sync.getWriteCheckpoint(); + }); + } + + protected async withSyncImplementation(callback: (sync: StreamingSyncImplementation) => Promise): Promise { await this.waitForReady(); - return this.syncStreamClient!.getWriteCheckpoint(); + + if (this.connectionManager.syncStreamImplementation) { + return callback(this.connectionManager.syncStreamImplementation); + } + + const sync = await new Promise((resolve) => { + const dispose = this.connectionManager.registerListener({ + syncStreamCreated: (sync) => { + resolve(sync); + dispose?.(); + } + }); + }); + + return callback(sync); } protected generateStreamingImplementation() { @@ -406,14 +455,6 @@ export class SharedSyncImplementation * sync stream client and all tab client's sync status */ private _testUpdateAllStatuses(status: SyncStatusOptions) { - if (!this.syncStreamClient) { - // This is just for testing purposes - this.syncStreamClient = this.generateStreamingImplementation(); - } - - // Only assigning, don't call listeners for this test - this.syncStreamClient!.syncStatus = new SyncStatus(status); - this.updateAllStatuses(status); } } diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.worker.ts b/packages/web/src/worker/sync/SharedSyncImplementation.worker.ts index ad7644456..11a14fd9b 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.worker.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.worker.ts @@ -1,10 +1,10 @@ +import { createBaseLogger } from '@powersync/common'; import * as Comlink from 'comlink'; import { - SharedSyncImplementation, SharedSyncClientEvent, + SharedSyncImplementation, type ManualSharedSyncPayload } from './SharedSyncImplementation'; -import { createBaseLogger } from '@powersync/common'; const _self: SharedWorkerGlobalScope = self as any; const logger = createBaseLogger(); @@ -12,20 +12,25 @@ logger.useDefaults(); const sharedSyncImplementation = new SharedSyncImplementation(); -_self.onconnect = function (event: MessageEvent) { +_self.onconnect = async function (event: MessageEvent) { const port = event.ports[0]; /** * Adds an extra listener which can remove this port * from the list of monitored ports. */ - port.addEventListener('message', (event) => { + port.addEventListener('message', async (event) => { const payload = event.data as ManualSharedSyncPayload; if (payload?.event == SharedSyncClientEvent.CLOSE_CLIENT) { - sharedSyncImplementation.removePort(port); + const release = await sharedSyncImplementation.removePort(port); + port.postMessage({ + event: SharedSyncClientEvent.CLOSE_ACK, + data: {} + } satisfies ManualSharedSyncPayload); + release?.(); } }); + await sharedSyncImplementation.addPort(port); Comlink.expose(sharedSyncImplementation, port); - sharedSyncImplementation.addPort(port); }; diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 842bd5585..647ba4d7b 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -216,7 +216,10 @@ function describeStreamingTests(createConnectedDatabase: () => Promise Date: Wed, 28 May 2025 13:55:10 +0200 Subject: [PATCH 12/26] cleanup connection manager --- .../src/client/AbstractPowerSyncDatabase.ts | 1 - .../common/src/client/ConnectionManager.ts | 50 ++++++++++--------- packages/web/tests/stream.test.ts | 22 ++++++-- 3 files changed, 43 insertions(+), 30 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index 408e3c85b..1981d0439 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -478,7 +478,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { // We only need to trigger a disconnect here if we have already reached the point of connecting. // If we do already have pending options, a disconnect has already been performed. // The connectInternal method also does a sanity disconnect to prevent straggler connections. - if (!hadPendingOptions) { + // We should also disconnect if we have already completed a connection attempt. + if (!hadPendingOptions || !this.connectingPromise) { await this.disconnectInternal(); } @@ -136,46 +137,47 @@ export class ConnectionManager extends BaseObserver { * This is protected in an exclusive lock. * The promise tracks the creation which is used to synchronize disconnect attempts. */ - this.syncStreamInitPromise = (async () => { - // Always await this if present since we will be populating a new sync implementation shortly - await this.disconnectingPromise; - - if (!this.pendingConnectionOptions) { - // A disconnect could have cleared this. - return; + this.syncStreamInitPromise = new Promise(async (resolve, reject) => { + try { + await this.disconnectingPromise; + + if (!this.pendingConnectionOptions) { + this.logger.debug('No pending connection options found, not creating sync stream implementation'); + // A disconnect could have cleared this. + return; + } + + const { connector, options } = this.pendingConnectionOptions; + appliedOptions = options; + + this.pendingConnectionOptions = null; + + const { sync, onDispose } = await this.options.createSyncImplementation(connector, options); + this.iterateListeners((l) => l.syncStreamCreated?.(sync)); + this.syncStreamImplementation = sync; + this.syncDisposer = onDispose; + await this.syncStreamImplementation.waitForReady(); + resolve(); + } catch (error) { + reject(error); } - - const { connector, options } = this.pendingConnectionOptions; - appliedOptions = options; - - this.pendingConnectionOptions = null; - - const { sync, onDispose } = await this.options.createSyncImplementation(connector, options); - this.iterateListeners((l) => l.syncStreamCreated?.(sync)); - this.syncStreamImplementation = sync; - this.syncDisposer = onDispose; - await this.syncStreamImplementation.waitForReady(); - })(); + }); await this.syncStreamInitPromise; this.syncStreamInitPromise = null; if (!appliedOptions) { - this.logger.debug('No pending connection options found, not connecting'); // A disconnect could have cleared the options which did not create a syncStreamImplementation return; } // It might be possible that a disconnect triggered between the last check // and this point. Awaiting here allows the sync stream to be cleared if disconnected. - this.logger.debug('Waiting for disconnect to complete before connecting', this.disconnectingPromise); await this.disconnectingPromise; - this.logger.debug('Disconnect completed, proceeding to connect'); this.syncStreamImplementation?.triggerCrudUpload(); this.logger.debug('Attempting to connect to PowerSync instance'); await this.syncStreamImplementation?.connect(appliedOptions!); - this.logger.debug('connected to sync stream implementation'); } /** diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 647ba4d7b..8e3fcbdb1 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -1,5 +1,6 @@ import { BucketChecksum, + createBaseLogger, DataStream, PowerSyncConnectionOptions, WASQLiteOpenFactory, @@ -11,16 +12,25 @@ import { ConnectedDatabaseUtils, generateConnectedDatabase } from './utils/gener const UPLOAD_TIMEOUT_MS = 3000; +const logger = createBaseLogger(); +logger.useDefaults(); + describe('Streaming', { sequential: true }, () => { describe( 'Streaming - With Web Workers', { sequential: true }, - describeStreamingTests(() => generateConnectedDatabase()) + describeStreamingTests(() => + generateConnectedDatabase({ + powerSyncOptions: { + logger + } + }) + ) ); - describe( + describe.only( 'Streaming - Without Web Workers', { sequential: true @@ -30,7 +40,8 @@ describe('Streaming', { sequential: true }, () => { powerSyncOptions: { flags: { useWebWorker: false - } + }, + logger } }) ) @@ -47,7 +58,8 @@ describe('Streaming', { sequential: true }, () => { database: new WASQLiteOpenFactory({ dbFilename: 'streaming-opfs.sqlite', vfs: WASQLiteVFS.OPFSCoopSyncVFS - }) + }), + logger } }) ) @@ -213,7 +225,7 @@ function describeStreamingTests(createConnectedDatabase: () => Promise Date: Wed, 28 May 2025 14:01:12 +0200 Subject: [PATCH 13/26] remove only call --- packages/web/tests/stream.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 8e3fcbdb1..691390491 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -30,7 +30,7 @@ describe('Streaming', { sequential: true }, () => { ) ); - describe.only( + describe( 'Streaming - Without Web Workers', { sequential: true From 3d57e2875600078cbc40b425ba6baae72d288548 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 28 May 2025 14:14:15 +0200 Subject: [PATCH 14/26] fix upload test --- .../worker/sync/SharedSyncImplementation.ts | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.ts b/packages/web/src/worker/sync/SharedSyncImplementation.ts index 616319365..25b1662c3 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.ts @@ -164,15 +164,15 @@ export class SharedSyncImplementation } async waitForStatus(status: SyncStatusOptions): Promise { - await this.waitForReady(); - // TODO - return this.connectionManager.syncStreamImplementation!.waitForStatus(status); + return this.withSyncImplementation(async (sync) => { + return sync.waitForStatus(status); + }); } async waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise { - await this.waitForReady(); - // TODO - return this.connectionManager.syncStreamImplementation!.waitUntilStatusMatches(predicate); + return this.withSyncImplementation(async (sync) => { + return sync.waitUntilStatusMatches(predicate); + }); } async waitForReady() { @@ -455,6 +455,13 @@ export class SharedSyncImplementation * sync stream client and all tab client's sync status */ private _testUpdateAllStatuses(status: SyncStatusOptions) { + if (!this.connectionManager.syncStreamImplementation) { + // This is just for testing purposes + this.connectionManager.syncStreamImplementation = this.generateStreamingImplementation(); + } + + // Only assigning, don't call listeners for this test + this.connectionManager.syncStreamImplementation!.syncStatus = new SyncStatus(status); this.updateAllStatuses(status); } } From a0dfaf67f24c302c027474b31a1f729df8ba2008 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 28 May 2025 14:50:08 +0200 Subject: [PATCH 15/26] test timeout on ci --- packages/web/tests/stream.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/web/tests/stream.test.ts b/packages/web/tests/stream.test.ts index 691390491..fec4a70da 100644 --- a/packages/web/tests/stream.test.ts +++ b/packages/web/tests/stream.test.ts @@ -225,7 +225,7 @@ function describeStreamingTests(createConnectedDatabase: () => Promise Date: Wed, 28 May 2025 17:09:48 +0200 Subject: [PATCH 16/26] stability improvements --- .../common/src/client/ConnectionManager.ts | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/packages/common/src/client/ConnectionManager.ts b/packages/common/src/client/ConnectionManager.ts index 0eb352611..ede731dde 100644 --- a/packages/common/src/client/ConnectionManager.ts +++ b/packages/common/src/client/ConnectionManager.ts @@ -101,7 +101,7 @@ export class ConnectionManager extends BaseObserver { // If we do already have pending options, a disconnect has already been performed. // The connectInternal method also does a sanity disconnect to prevent straggler connections. // We should also disconnect if we have already completed a connection attempt. - if (!hadPendingOptions || !this.connectingPromise) { + if (!hadPendingOptions) { await this.disconnectInternal(); } @@ -139,8 +139,6 @@ export class ConnectionManager extends BaseObserver { */ this.syncStreamInitPromise = new Promise(async (resolve, reject) => { try { - await this.disconnectingPromise; - if (!this.pendingConnectionOptions) { this.logger.debug('No pending connection options found, not creating sync stream implementation'); // A disconnect could have cleared this. @@ -175,9 +173,9 @@ export class ConnectionManager extends BaseObserver { // and this point. Awaiting here allows the sync stream to be cleared if disconnected. await this.disconnectingPromise; - this.syncStreamImplementation?.triggerCrudUpload(); this.logger.debug('Attempting to connect to PowerSync instance'); await this.syncStreamImplementation?.connect(appliedOptions!); + this.syncStreamImplementation?.triggerCrudUpload(); } /** @@ -208,10 +206,16 @@ export class ConnectionManager extends BaseObserver { } protected async performDisconnect() { - await this.syncStreamImplementation?.disconnect(); - await this.syncStreamImplementation?.dispose(); - await this.syncDisposer?.(); - this.syncDisposer = null; + // Keep reference to the sync stream implementation and disposer + // The class members will be cleared before we trigger the disconnect + // to prevent any further calls to the sync stream implementation. + const sync = this.syncStreamImplementation; this.syncStreamImplementation = null; + const disposer = this.syncDisposer; + this.syncDisposer = null; + + await sync?.disconnect(); + await sync?.dispose(); + await disposer?.(); } } From 9d37979308ec3d21db21229fb620b7d91a049987 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 10:49:06 +0200 Subject: [PATCH 17/26] Fix reconnect bugs --- .../common/src/client/ConnectionManager.ts | 27 +++-- .../AbstractStreamingSyncImplementation.ts | 7 +- .../worker/sync/SharedSyncImplementation.ts | 100 ++++++++++-------- 3 files changed, 82 insertions(+), 52 deletions(-) diff --git a/packages/common/src/client/ConnectionManager.ts b/packages/common/src/client/ConnectionManager.ts index ede731dde..ba78875f6 100644 --- a/packages/common/src/client/ConnectionManager.ts +++ b/packages/common/src/client/ConnectionManager.ts @@ -11,6 +11,10 @@ import { */ export interface ConnectionManagerSyncImplementationResult { sync: StreamingSyncImplementation; + /** + * Additional cleanup function which is called after the sync stream implementation + * is disposed. + */ onDispose: () => Promise | void; } @@ -65,7 +69,12 @@ export class ConnectionManager extends BaseObserver { protected pendingConnectionOptions: StoredConnectionOptions | null; syncStreamImplementation: StreamingSyncImplementation | null; - syncDisposer: (() => Promise | void) | null; + + /** + * Additional cleanup function which is called after the sync stream implementation + * is disposed. + */ + protected syncDisposer: (() => Promise | void) | null; constructor(protected options: ConnectionManagerOptions) { super(); @@ -82,6 +91,7 @@ export class ConnectionManager extends BaseObserver { } async close() { + await this.syncStreamImplementation?.dispose(); await this.syncDisposer?.(); } @@ -101,7 +111,7 @@ export class ConnectionManager extends BaseObserver { // If we do already have pending options, a disconnect has already been performed. // The connectInternal method also does a sanity disconnect to prevent straggler connections. // We should also disconnect if we have already completed a connection attempt. - if (!hadPendingOptions) { + if (!hadPendingOptions || this.syncStreamImplementation) { await this.disconnectInternal(); } @@ -145,11 +155,14 @@ export class ConnectionManager extends BaseObserver { return; } + if (this.disconnectingPromise) { + return; + } + const { connector, options } = this.pendingConnectionOptions; appliedOptions = options; this.pendingConnectionOptions = null; - const { sync, onDispose } = await this.options.createSyncImplementation(connector, options); this.iterateListeners((l) => l.syncStreamCreated?.(sync)); this.syncStreamImplementation = sync; @@ -195,10 +208,6 @@ export class ConnectionManager extends BaseObserver { return this.disconnectingPromise; } - // Wait if a sync stream implementation is being created before closing it - // (syncStreamImplementation must be assigned before we can properly dispose it) - await this.syncStreamInitPromise; - this.disconnectingPromise = this.performDisconnect(); await this.disconnectingPromise; @@ -206,6 +215,10 @@ export class ConnectionManager extends BaseObserver { } protected async performDisconnect() { + // Wait if a sync stream implementation is being created before closing it + // (syncStreamImplementation must be assigned before we can properly dispose it) + await this.syncStreamInitPromise; + // Keep reference to the sync stream implementation and disposer // The class members will be cleared before we trigger the disconnect // to prevent any further calls to the sync stream implementation. diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 61d298c2c..6f7d372ca 100644 --- a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -449,6 +449,7 @@ The next upload iteration will be delayed.`); await this.streamingSyncIteration(nestedAbortController.signal, options); // Continue immediately, streamingSyncIteration will wait before completing if necessary. } catch (ex) { + let delay = true; /** * Either: * - A network request failed with a failed connection or not OKAY response code. @@ -459,6 +460,8 @@ The next upload iteration will be delayed.`); */ if (ex instanceof AbortOperation) { this.logger.warn(ex); + delay = false; + // A disconnect was requested, we should not delay since there is no explicit retry } else { this.logger.error(ex); } @@ -470,7 +473,9 @@ The next upload iteration will be delayed.`); }); // On error, wait a little before retrying - await this.delayRetry(); + if (delay) { + await this.delayRetry(); + } } finally { if (!signal.aborted) { nestedAbortController.abort(new AbortOperation('Closing sync stream network requests before retry.')); diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.ts b/packages/web/src/worker/sync/SharedSyncImplementation.ts index 25b1662c3..5d6770b92 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.ts @@ -30,6 +30,7 @@ import { AbstractSharedSyncClientProvider } from './AbstractSharedSyncClientProv import { BroadcastLogger } from './BroadcastLogger'; /** + * @internal * Manual message events for shared sync clients */ export enum SharedSyncClientEvent { @@ -42,6 +43,9 @@ export enum SharedSyncClientEvent { CLOSE_ACK = 'close-ack' } +/** + * @internal + */ export type ManualSharedSyncPayload = { event: SharedSyncClientEvent; data: any; // TODO update in future @@ -85,6 +89,7 @@ export type RemoteOperationAbortController = { * We provide this unused placeholder when connecting with the ConnectionManager. */ const CONNECTOR_PLACEHOLDER = {} as PowerSyncBackendConnector; + /** * @internal * Shared sync implementation which runs inside a shared webworker @@ -133,23 +138,25 @@ export class SharedSyncImplementation this.broadCastLogger = new BroadcastLogger(this.ports); this.connectionManager = new ConnectionManager({ - createSyncImplementation: async (connector, options) => { - await this.waitForReady(); - if (!this.dbAdapter) { - await this.openInternalDB(); - } - - const sync = this.generateStreamingImplementation(); - const onDispose = sync.registerListener({ - statusChanged: (status) => { - this.updateAllStatuses(status.toJSON()); + createSyncImplementation: async () => { + return this.portMutex.runExclusive(async () => { + await this.waitForReady(); + if (!this.dbAdapter) { + await this.openInternalDB(); } - }); - return { - sync, - onDispose - }; + const sync = await this.generateStreamingImplementation(); + const onDispose = sync.registerListener({ + statusChanged: (status) => { + this.updateAllStatuses(status.toJSON()); + } + }); + + return { + sync, + onDispose + }; + }); }, logger: this.logger }); @@ -190,15 +197,17 @@ export class SharedSyncImplementation async setParams(params: SharedSyncInitOptions) { await this.portMutex.runExclusive(async () => { if (this.syncParams) { + // Cannot modify already existing sync implementation params + // But we can ask for a DB adapter, if required, at this point. + if (!this.dbAdapter) { await this.openInternalDB(); } - // Cannot modify already existing sync implementation return; } + // First time setting params this.syncParams = params; - if (params.streamOptions?.flags?.broadcastLogs) { this.logger = this.broadCastLogger; } @@ -229,17 +238,12 @@ export class SharedSyncImplementation * connects. */ async connect(options?: PowerSyncConnectionOptions) { - await this.portMutex.runExclusive(async () => { - // Keep track of the last connect options if we need to reconnect due to a lost client - this.lastConnectOptions = options; - return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options); - }); + this.lastConnectOptions = options; + return this.connectionManager.connect(CONNECTOR_PLACEHOLDER, options); } async disconnect() { - await this.portMutex.runExclusive(async () => { - await this.connectionManager.disconnect(); - }); + return this.connectionManager.disconnect(); } /** @@ -266,11 +270,11 @@ export class SharedSyncImplementation * clients. */ async removePort(port: MessagePort) { - return await this.portMutex.runExclusive(async () => { + const { trackedPort, shouldReconnect } = await this.portMutex.runExclusive(async () => { const index = this.ports.findIndex((p) => p.port == port); if (index < 0) { this.logger.warn(`Could not remove port ${port} since it is not present in active ports.`); - return; + return {}; } const trackedPort = this.ports[index]; @@ -291,26 +295,35 @@ export class SharedSyncImplementation const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0; - if (this.dbAdapter && this.dbAdapter == trackedPort.db) { - if (shouldReconnect) { - await this.connectionManager.disconnect(); - } + return { + shouldReconnect, + trackedPort + }; + }); - // Clearing the adapter will result in a new one being opened in connect - this.dbAdapter = null; + if (!trackedPort) { + // We could not find the port to remove + return () => {}; + } - if (shouldReconnect) { - await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions); - } + if (this.dbAdapter && this.dbAdapter == trackedPort.db) { + if (shouldReconnect) { + await this.connectionManager.disconnect(); } - if (trackedPort.db) { - await trackedPort.db.close(); + // Clearing the adapter will result in a new one being opened in connect + this.dbAdapter = null; + + if (shouldReconnect) { + await this.connectionManager.connect(CONNECTOR_PLACEHOLDER, this.lastConnectOptions); } - this.logger.debug(`Port ${port} removed from shared sync implementation.`); - // Release proxy - return () => trackedPort.clientProvider[Comlink.releaseProxy](); - }); + } + + if (trackedPort.db) { + await trackedPort.db.close(); + } + // Release proxy + return () => trackedPort.clientProvider[Comlink.releaseProxy](); } triggerCrudUpload() { @@ -351,7 +364,6 @@ export class SharedSyncImplementation protected generateStreamingImplementation() { // This should only be called after initialization has completed const syncParams = this.syncParams!; - // Create a new StreamingSyncImplementation for each connect call. This is usually done is all SDKs. return new WebStreamingSyncImplementation({ adapter: new SqliteBucketStorage(this.dbAdapter!, new Mutex(), this.logger), @@ -454,7 +466,7 @@ export class SharedSyncImplementation * A function only used for unit tests which updates the internal * sync stream client and all tab client's sync status */ - private _testUpdateAllStatuses(status: SyncStatusOptions) { + private async _testUpdateAllStatuses(status: SyncStatusOptions) { if (!this.connectionManager.syncStreamImplementation) { // This is just for testing purposes this.connectionManager.syncStreamImplementation = this.generateStreamingImplementation(); From bf881fc6445f96af2b693c3a2a756f36161bf9a5 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 11:01:52 +0200 Subject: [PATCH 18/26] code cleanup --- packages/common/src/client/AbstractPowerSyncDatabase.ts | 6 +++--- packages/web/src/db/PowerSyncDatabase.ts | 4 ---- packages/web/src/worker/sync/SharedSyncImplementation.ts | 4 +++- packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts | 4 ---- 4 files changed, 6 insertions(+), 12 deletions(-) diff --git a/packages/common/src/client/AbstractPowerSyncDatabase.ts b/packages/common/src/client/AbstractPowerSyncDatabase.ts index 1981d0439..590747969 100644 --- a/packages/common/src/client/AbstractPowerSyncDatabase.ts +++ b/packages/common/src/client/AbstractPowerSyncDatabase.ts @@ -180,7 +180,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver { @@ -462,7 +462,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver(callback: () => Promise): Promise { - return this.connectionMutex.runExclusive(callback); + return this.runExclusiveMutex.runExclusive(callback); } /** diff --git a/packages/web/src/db/PowerSyncDatabase.ts b/packages/web/src/db/PowerSyncDatabase.ts index 072b62da0..8351b4f1b 100644 --- a/packages/web/src/db/PowerSyncDatabase.ts +++ b/packages/web/src/db/PowerSyncDatabase.ts @@ -171,10 +171,6 @@ export class PowerSyncDatabase extends AbstractPowerSyncDatabase { }); } - protected async connectExclusive(callback: () => Promise): Promise { - await this.runExclusive(callback); - } - protected generateBucketStorageAdapter(): BucketStorageAdapter { return new SqliteBucketStorage(this.database, AbstractPowerSyncDatabase.transactionMutex); } diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.ts b/packages/web/src/worker/sync/SharedSyncImplementation.ts index 5d6770b92..e0a062511 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.ts @@ -327,7 +327,9 @@ export class SharedSyncImplementation } triggerCrudUpload() { - this.waitForReady().then(() => this.connectionManager.syncStreamImplementation?.triggerCrudUpload()); + this.withSyncImplementation(async (sync) => { + sync.triggerCrudUpload(); + }); } async hasCompletedSync(): Promise { diff --git a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts index 12e89f2cf..581d94722 100644 --- a/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts +++ b/packages/web/tests/src/db/AbstractPowerSyncDatabase.test.ts @@ -31,10 +31,6 @@ class TestPowerSyncDatabase extends AbstractPowerSyncDatabase { return Promise.resolve(); } - protected async runExclusive(callback: () => Promise): Promise { - return await callback(); - } - get database() { return { get: vi.fn().mockResolvedValue({ version: '0.3.11' }), From b054bf1668e0030e6968f11e1d125fcd835daa3e Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 11:14:01 +0200 Subject: [PATCH 19/26] Add code comments --- .../sync/stream/AbstractStreamingSyncImplementation.ts | 5 +++-- packages/web/src/worker/sync/SharedSyncImplementation.ts | 5 ++++- packages/web/tests/utils/MockStreamOpenFactory.ts | 3 +++ 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index 6f7d372ca..b83844eed 100644 --- a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -449,15 +449,16 @@ The next upload iteration will be delayed.`); await this.streamingSyncIteration(nestedAbortController.signal, options); // Continue immediately, streamingSyncIteration will wait before completing if necessary. } catch (ex) { - let delay = true; /** * Either: * - A network request failed with a failed connection or not OKAY response code. * - There was a sync processing error. - * This loop will retry. + * - The connection was aborted. + * This loop will retry after a delay if the connection was not aborted. * The nested abort controller will cleanup any open network requests and streams. * The WebRemote should only abort pending fetch requests or close active Readable streams. */ + let delay = true; if (ex instanceof AbortOperation) { this.logger.warn(ex); delay = false; diff --git a/packages/web/src/worker/sync/SharedSyncImplementation.ts b/packages/web/src/worker/sync/SharedSyncImplementation.ts index e0a062511..0af122ead 100644 --- a/packages/web/src/worker/sync/SharedSyncImplementation.ts +++ b/packages/web/src/worker/sync/SharedSyncImplementation.ts @@ -145,7 +145,7 @@ export class SharedSyncImplementation await this.openInternalDB(); } - const sync = await this.generateStreamingImplementation(); + const sync = this.generateStreamingImplementation(); const onDispose = sync.registerListener({ statusChanged: (status) => { this.updateAllStatuses(status.toJSON()); @@ -270,6 +270,9 @@ export class SharedSyncImplementation * clients. */ async removePort(port: MessagePort) { + // Remove the port within a mutex context. + // Warns if the port is not found. This should not happen in practice. + // We return early if the port is not found. const { trackedPort, shouldReconnect } = await this.portMutex.runExclusive(async () => { const index = this.ports.findIndex((p) => p.port == port); if (index < 0) { diff --git a/packages/web/tests/utils/MockStreamOpenFactory.ts b/packages/web/tests/utils/MockStreamOpenFactory.ts index d4d32c39f..84dfad789 100644 --- a/packages/web/tests/utils/MockStreamOpenFactory.ts +++ b/packages/web/tests/utils/MockStreamOpenFactory.ts @@ -81,6 +81,9 @@ export class MockRemote extends AbstractRemote { if (this.errorOnStreamStart) { controller.error(new Error('Mock error on stream start')); } + // The request could be aborted at any time. + // This checks if the signal is already aborted and closes the stream if so. + // If not, it adds an event listener to close the stream when the signal is aborted. if (signal?.aborted) { controller.close(); return; From 5f6282960f14c661a9cbcff321f813750f30acc7 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 16:08:26 +0200 Subject: [PATCH 20/26] fix react native ios reconnect --- .changeset/cyan-penguins-smell.md | 5 ++ .../src/client/sync/stream/AbstractRemote.ts | 2 +- .../sync/stream/WebsocketClientTransport.ts | 70 +++++++++++++++++++ packages/react-native/rollup.config.mjs | 2 +- 4 files changed, 77 insertions(+), 2 deletions(-) create mode 100644 .changeset/cyan-penguins-smell.md create mode 100644 packages/common/src/client/sync/stream/WebsocketClientTransport.ts diff --git a/.changeset/cyan-penguins-smell.md b/.changeset/cyan-penguins-smell.md new file mode 100644 index 000000000..ac7428965 --- /dev/null +++ b/.changeset/cyan-penguins-smell.md @@ -0,0 +1,5 @@ +--- +'@powersync/react-native': minor +--- + +Fixed issue where iOS WebSockets could fail to reconnect after a connection issue. diff --git a/packages/common/src/client/sync/stream/AbstractRemote.ts b/packages/common/src/client/sync/stream/AbstractRemote.ts index 1d180e6fe..0ccc8503d 100644 --- a/packages/common/src/client/sync/stream/AbstractRemote.ts +++ b/packages/common/src/client/sync/stream/AbstractRemote.ts @@ -4,12 +4,12 @@ import ndjsonStream from 'can-ndjson-stream'; import { type fetch } from 'cross-fetch'; import Logger, { ILogger } from 'js-logger'; import { RSocket, RSocketConnector, Requestable } from 'rsocket-core'; -import { WebsocketClientTransport } from 'rsocket-websocket-client'; import PACKAGE from '../../../../package.json' with { type: 'json' }; import { AbortOperation } from '../../../utils/AbortOperation.js'; import { DataStream } from '../../../utils/DataStream.js'; import { PowerSyncCredentials } from '../../connection/PowerSyncCredentials.js'; import { StreamingSyncLine, StreamingSyncRequest } from './streaming-sync-types.js'; +import { WebsocketClientTransport } from './WebsocketClientTransport.js'; export type BSONImplementation = typeof BSON; diff --git a/packages/common/src/client/sync/stream/WebsocketClientTransport.ts b/packages/common/src/client/sync/stream/WebsocketClientTransport.ts new file mode 100644 index 000000000..55a09c863 --- /dev/null +++ b/packages/common/src/client/sync/stream/WebsocketClientTransport.ts @@ -0,0 +1,70 @@ +/** + * Adapted from rsocket-websocket-client + * https://github.com/rsocket/rsocket-js/blob/e224cf379e747c4f1ddc4f2fa111854626cc8575/packages/rsocket-websocket-client/src/WebsocketClientTransport.ts#L17 + * This adds additional error handling for React Native iOS. + * This particularly adds a close listener to handle cases where the WebSocket + * connection closes immediately after opening without emitting an error. + */ +import { + ClientTransport, + Closeable, + Demultiplexer, + Deserializer, + DuplexConnection, + FrameHandler, + Multiplexer, + Outbound +} from 'rsocket-core'; +import { ClientOptions } from 'rsocket-websocket-client'; +import { WebsocketDuplexConnection } from 'rsocket-websocket-client/dist/WebsocketDuplexConnection.js'; + +export class WebsocketClientTransport implements ClientTransport { + private readonly url: string; + private readonly factory: (url: string) => WebSocket; + + constructor(options: ClientOptions) { + this.url = options.url; + this.factory = options.wsCreator ?? ((url: string) => new WebSocket(url)); + } + + connect( + multiplexerDemultiplexerFactory: (outbound: Outbound & Closeable) => Multiplexer & Demultiplexer & FrameHandler + ): Promise { + return new Promise((resolve, reject) => { + const websocket = this.factory(this.url); + + websocket.binaryType = 'arraybuffer'; + + let removeListeners: () => void; + + const openListener = () => { + removeListeners(); + resolve(new WebsocketDuplexConnection(websocket, new Deserializer(), multiplexerDemultiplexerFactory)); + }; + + const errorListener = (ev: ErrorEvent) => { + removeListeners(); + reject(ev.error); + }; + + /** + * In some cases, such as React Native iOS, the WebSocket connection may close immediately after opening + * without and error. In such cases, we need to handle the close event to reject the promise. + */ + const closeListener = () => { + removeListeners(); + reject(new Error('WebSocket connection closed while opening')); + }; + + removeListeners = () => { + websocket.removeEventListener('open', openListener); + websocket.removeEventListener('error', errorListener); + websocket.removeEventListener('close', closeListener); + }; + + websocket.addEventListener('open', openListener); + websocket.addEventListener('error', errorListener); + websocket.addEventListener('close', closeListener); + }); + } +} diff --git a/packages/react-native/rollup.config.mjs b/packages/react-native/rollup.config.mjs index 9a180747c..d2ee7e7fa 100644 --- a/packages/react-native/rollup.config.mjs +++ b/packages/react-native/rollup.config.mjs @@ -54,7 +54,7 @@ export default (commandLineArgs) => { } ] }), - terser() + terser({ sourceMap: sourcemap }) ], external: [ '@journeyapps/react-native-quick-sqlite', From 54196610f8d6a05bfd137e77883ca6ec1119ed86 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 17:13:07 +0200 Subject: [PATCH 21/26] fix typo --- .changeset/empty-pants-give.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/empty-pants-give.md b/.changeset/empty-pants-give.md index c343f0bcb..ccf2b5c00 100644 --- a/.changeset/empty-pants-give.md +++ b/.changeset/empty-pants-give.md @@ -5,4 +5,4 @@ '@powersync/web': minor --- -Improved behvaiour when connect is called multiple times in quick succession. Updating client parameters should now be more responsive. +Improved behaviour when connect is called multiple times in quick succession. Updating client parameters should now be more responsive. From c1657958d067658ed5174ca30f6e33c5df669b29 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Thu, 29 May 2025 17:13:25 +0200 Subject: [PATCH 22/26] add logger to RN demo --- .../library/powersync/system.ts | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/demos/react-native-supabase-todolist/library/powersync/system.ts b/demos/react-native-supabase-todolist/library/powersync/system.ts index 6ca46a861..8f1c2713e 100644 --- a/demos/react-native-supabase-todolist/library/powersync/system.ts +++ b/demos/react-native-supabase-todolist/library/powersync/system.ts @@ -5,16 +5,16 @@ import React from 'react'; import { SupabaseStorageAdapter } from '../storage/SupabaseStorageAdapter'; import { type AttachmentRecord } from '@powersync/attachments'; +import { configureFts } from '../fts/fts_setup'; import { KVStorage } from '../storage/KVStorage'; import { AppConfig } from '../supabase/AppConfig'; import { SupabaseConnector } from '../supabase/SupabaseConnector'; import { AppSchema } from './AppSchema'; import { PhotoAttachmentQueue } from './PhotoAttachmentQueue'; -import { configureFts } from '../fts/fts_setup'; const logger = createBaseLogger(); logger.useDefaults(); -logger.setLevel(LogLevel.INFO); +logger.setLevel(LogLevel.DEBUG); export class System { kvStorage: KVStorage; @@ -31,19 +31,22 @@ export class System { schema: AppSchema, database: { dbFilename: 'sqlite.db' - } + }, + logger }); /** * The snippet below uses OP-SQLite as the default database adapter. * You will have to uninstall `@journeyapps/react-native-quick-sqlite` and * install both `@powersync/op-sqlite` and `@op-engineering/op-sqlite` to use this. * + * ```typescript * import { OPSqliteOpenFactory } from '@powersync/op-sqlite'; // Add this import * * const factory = new OPSqliteOpenFactory({ - * dbFilename: 'sqlite.db' + * dbFilename: 'sqlite.db' * }); * this.powersync = new PowerSyncDatabase({ database: factory, schema: AppSchema }); + * ``` */ if (AppConfig.supabaseBucket) { From b495ca7980109b186bdad976c6da09eb0e7df9b7 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Mon, 2 Jun 2025 15:15:46 +0200 Subject: [PATCH 23/26] fix multiple tab connection contention issue --- packages/common/src/client/ConnectionManager.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/common/src/client/ConnectionManager.ts b/packages/common/src/client/ConnectionManager.ts index ba78875f6..43d1c929f 100644 --- a/packages/common/src/client/ConnectionManager.ts +++ b/packages/common/src/client/ConnectionManager.ts @@ -152,10 +152,12 @@ export class ConnectionManager extends BaseObserver { if (!this.pendingConnectionOptions) { this.logger.debug('No pending connection options found, not creating sync stream implementation'); // A disconnect could have cleared this. + resolve(); return; } if (this.disconnectingPromise) { + resolve(); return; } From 76595332d52ba76b79a1710eea46152de5a76327 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 4 Jun 2025 16:52:04 +0200 Subject: [PATCH 24/26] improve retry delay logic. Update diagnostics app error detection. --- .changeset/fuzzy-beers-occur.md | 5 ++ .../common/src/client/ConnectionManager.ts | 8 +++- .../AbstractStreamingSyncImplementation.ts | 41 +++++++++++++---- .../diagnostics-app/src/app/views/layout.tsx | 15 ++---- .../src/app/views/sync-diagnostics.tsx | 12 +---- .../library/powersync/ConnectionManager.ts | 46 ++++--------------- 6 files changed, 58 insertions(+), 69 deletions(-) create mode 100644 .changeset/fuzzy-beers-occur.md diff --git a/.changeset/fuzzy-beers-occur.md b/.changeset/fuzzy-beers-occur.md new file mode 100644 index 000000000..ab125ae07 --- /dev/null +++ b/.changeset/fuzzy-beers-occur.md @@ -0,0 +1,5 @@ +--- +'@powersync/diagnostics-app': patch +--- + +Fix bug where clicking signOut would not disconnect from the PowerSync service. Updated implementation to fetch sync errors from the SyncStatus. diff --git a/packages/common/src/client/ConnectionManager.ts b/packages/common/src/client/ConnectionManager.ts index 43d1c929f..1ade81c36 100644 --- a/packages/common/src/client/ConnectionManager.ts +++ b/packages/common/src/client/ConnectionManager.ts @@ -122,7 +122,9 @@ export class ConnectionManager extends BaseObserver { if (this.pendingConnectionOptions) { // Pending options have been placed while connecting. // Need to reconnect. - this.connectingPromise = this.connectInternal().finally(checkConnection); + this.connectingPromise = this.connectInternal() + .catch(() => {}) + .finally(checkConnection); return this.connectingPromise; } else { // Clear the connecting promise, done. @@ -131,7 +133,9 @@ export class ConnectionManager extends BaseObserver { } }; - this.connectingPromise ??= this.connectInternal().finally(checkConnection); + this.connectingPromise ??= this.connectInternal() + .catch(() => {}) + .finally(checkConnection); return this.connectingPromise; } diff --git a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts index af95abcc6..9997da2f9 100644 --- a/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts +++ b/packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts @@ -442,6 +442,7 @@ The next upload iteration will be delayed.`); */ while (true) { this.updateSyncStatus({ connecting: true }); + let shouldDelayRetry = true; try { if (signal?.aborted) { break; @@ -458,10 +459,10 @@ The next upload iteration will be delayed.`); * The nested abort controller will cleanup any open network requests and streams. * The WebRemote should only abort pending fetch requests or close active Readable streams. */ - let delay = true; + if (ex instanceof AbortOperation) { this.logger.warn(ex); - delay = false; + shouldDelayRetry = false; // A disconnect was requested, we should not delay since there is no explicit retry } else { this.logger.error(ex); @@ -472,11 +473,6 @@ The next upload iteration will be delayed.`); downloadError: ex } }); - - // On error, wait a little before retrying - if (delay) { - await this.delayRetry(); - } } finally { if (!signal.aborted) { nestedAbortController.abort(new AbortOperation('Closing sync stream network requests before retry.')); @@ -487,6 +483,11 @@ The next upload iteration will be delayed.`); connected: false, connecting: true // May be unnecessary }); + + // On error, wait a little before retrying + if (shouldDelayRetry) { + await this.delayRetry(nestedAbortController.signal); + } } } @@ -854,7 +855,29 @@ The next upload iteration will be delayed.`); this.iterateListeners((cb) => cb.statusUpdated?.(options)); } - private async delayRetry() { - return new Promise((resolve) => setTimeout(resolve, this.options.retryDelayMs)); + private async delayRetry(signal?: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal?.aborted) { + // If the signal is already aborted, resolve immediately + resolve(); + return; + } + + const { retryDelayMs } = this.options; + + let timeoutId: ReturnType | undefined; + + const endDelay = () => { + resolve(); + if (timeoutId) { + clearTimeout(timeoutId); + timeoutId = undefined; + } + signal?.removeEventListener('abort', endDelay); + }; + + signal?.addEventListener('abort', endDelay, { once: true }); + timeoutId = setTimeout(endDelay, retryDelayMs); + }); } } diff --git a/tools/diagnostics-app/src/app/views/layout.tsx b/tools/diagnostics-app/src/app/views/layout.tsx index 567f0c687..97a3b5f2b 100644 --- a/tools/diagnostics-app/src/app/views/layout.tsx +++ b/tools/diagnostics-app/src/app/views/layout.tsx @@ -6,8 +6,8 @@ import SouthIcon from '@mui/icons-material/South'; import StorageIcon from '@mui/icons-material/Storage'; import TableChartIcon from '@mui/icons-material/TableChart'; import TerminalIcon from '@mui/icons-material/Terminal'; -import WifiIcon from '@mui/icons-material/Wifi'; import UserIcon from '@mui/icons-material/VerifiedUser'; +import WifiIcon from '@mui/icons-material/Wifi'; import { AppBar, @@ -34,7 +34,7 @@ import { SYNC_DIAGNOSTICS_ROUTE } from '@/app/router'; import { useNavigationPanel } from '@/components/navigation/NavigationPanelContext'; -import { signOut, sync, syncErrorTracker } from '@/library/powersync/ConnectionManager'; +import { signOut, sync } from '@/library/powersync/ConnectionManager'; import { usePowerSync } from '@powersync/react'; import { useNavigate } from 'react-router-dom'; @@ -103,19 +103,12 @@ export default function ViewsLayout({ children }: { children: React.ReactNode }) const l = sync.registerListener({ statusChanged: (status) => { setSyncStatus(status); + setSyncError(status.dataFlowStatus.downloadError ?? null); } }); return () => l(); }, []); - React.useEffect(() => { - const l = syncErrorTracker.registerListener({ - lastErrorUpdated(error) { - setSyncError(error); - } - }); - return () => l(); - }, []); const drawerWidth = 320; const drawer = ( @@ -220,4 +213,4 @@ namespace S { object-fit: contain; padding: 20px; `; -} \ No newline at end of file +} diff --git a/tools/diagnostics-app/src/app/views/sync-diagnostics.tsx b/tools/diagnostics-app/src/app/views/sync-diagnostics.tsx index 1aa3f90d5..7772f5336 100644 --- a/tools/diagnostics-app/src/app/views/sync-diagnostics.tsx +++ b/tools/diagnostics-app/src/app/views/sync-diagnostics.tsx @@ -1,5 +1,5 @@ import { NavigationPage } from '@/components/navigation/NavigationPage'; -import { clearData, db, sync, syncErrorTracker } from '@/library/powersync/ConnectionManager'; +import { clearData, db, sync } from '@/library/powersync/ConnectionManager'; import { Box, Button, @@ -73,7 +73,6 @@ FROM local_bucket_data local`; export default function SyncDiagnosticsPage() { const [bucketRows, setBucketRows] = React.useState(null); const [tableRows, setTableRows] = React.useState(null); - const [syncError, setSyncError] = React.useState(syncErrorTracker.lastSyncError); const [lastSyncedAt, setlastSyncedAt] = React.useState(null); const bucketRowsLoading = bucketRows == null; @@ -125,15 +124,6 @@ export default function SyncDiagnosticsPage() { }; }, []); - React.useEffect(() => { - const l = syncErrorTracker.registerListener({ - lastErrorUpdated(error) { - setSyncError(error); - } - }); - return () => l(); - }, []); - const columns: GridColDef[] = [ { field: 'name', headerName: 'Name', flex: 2 }, { field: 'tables', headerName: 'Table(s)', flex: 1, type: 'text' }, diff --git a/tools/diagnostics-app/src/library/powersync/ConnectionManager.ts b/tools/diagnostics-app/src/library/powersync/ConnectionManager.ts index 61053db92..8a9a4a0ae 100644 --- a/tools/diagnostics-app/src/library/powersync/ConnectionManager.ts +++ b/tools/diagnostics-app/src/library/powersync/ConnectionManager.ts @@ -1,20 +1,19 @@ import { BaseListener, - BaseObserver, + createBaseLogger, + LogLevel, PowerSyncDatabase, - WebRemote, - WebStreamingSyncImplementation, - WebStreamingSyncImplementationOptions, - WASQLiteOpenFactory, TemporaryStorageOption, + WASQLiteOpenFactory, WASQLiteVFS, - createBaseLogger, - LogLevel + WebRemote, + WebStreamingSyncImplementation, + WebStreamingSyncImplementationOptions } from '@powersync/web'; +import { safeParse } from '../safeParse/safeParse'; import { DynamicSchemaManager } from './DynamicSchemaManager'; import { RecordingStorageAdapter } from './RecordingStorageAdapter'; import { TokenConnector } from './TokenConnector'; -import { safeParse } from '../safeParse/safeParse'; const baseLogger = createBaseLogger(); baseLogger.useDefaults(); @@ -62,31 +61,6 @@ export interface SyncErrorListener extends BaseListener { lastErrorUpdated?: ((error: Error) => void) | undefined; } -class SyncErrorTracker extends BaseObserver { - public lastSyncError: Error | null = null; - - constructor() { - super(); - // Big hack: Use the logger to get access to connection errors - const defaultHandler = baseLogger.createDefaultHandler(); - baseLogger.setHandler((messages, context) => { - defaultHandler(messages, context); - if (context.name == 'PowerSyncStream' && context.level.name == 'ERROR') { - if (messages[0] instanceof Error) { - this.lastSyncError = messages[0]; - } else { - this.lastSyncError = new Error('' + messages[0]); - } - this.iterateListeners((listener) => { - listener.lastErrorUpdated?.(this.lastSyncError!); - }); - } - }); - } -} - -export const syncErrorTracker = new SyncErrorTracker(); - if (connector.hasCredentials()) { connect(); } @@ -95,11 +69,10 @@ export async function connect() { const params = getParams(); await sync.connect({ params }); if (!sync.syncStatus.connected) { + const error = sync.syncStatus.dataFlowStatus.downloadError ?? new Error('Failed to connect'); // Disconnect but don't wait for it sync.disconnect(); - throw syncErrorTracker.lastSyncError ?? new Error('Failed to connect'); - } else { - syncErrorTracker.lastSyncError = null; + throw error; } } @@ -120,6 +93,7 @@ export async function disconnect() { export async function signOut() { connector.clearCredentials(); + await disconnect(); await db.disconnectAndClear(); await schemaManager.clear(); } From 13a58e4a22384b8e1340e63d2552269a4827c844 Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 4 Jun 2025 17:44:42 +0200 Subject: [PATCH 25/26] Report Sync errors in all pages. Fix bug where changes in sync errors were not detected. --- .changeset/fuzzy-ties-double.md | 5 +++++ packages/common/src/db/crud/SyncStatus.ts | 17 ++++++++++++++++- tools/diagnostics-app/src/app/views/layout.tsx | 2 ++ 3 files changed, 23 insertions(+), 1 deletion(-) create mode 100644 .changeset/fuzzy-ties-double.md diff --git a/.changeset/fuzzy-ties-double.md b/.changeset/fuzzy-ties-double.md new file mode 100644 index 000000000..3a0b27c1d --- /dev/null +++ b/.changeset/fuzzy-ties-double.md @@ -0,0 +1,5 @@ +--- +'@powersync/common': patch +--- + +Fixed bug where changes in SyncStatus downloadError and uploadError might not be reported. diff --git a/packages/common/src/db/crud/SyncStatus.ts b/packages/common/src/db/crud/SyncStatus.ts index f01211eaa..775c9fa36 100644 --- a/packages/common/src/db/crud/SyncStatus.ts +++ b/packages/common/src/db/crud/SyncStatus.ts @@ -171,7 +171,22 @@ export class SyncStatus { * @returns {boolean} True if the instances are considered equal, false otherwise */ isEqual(status: SyncStatus) { - return JSON.stringify(this.options) == JSON.stringify(status.options); + /** + * By default Error object are serialized to an empty object. + * This replaces Errors with more useful information before serialization. + */ + const replacer = (_: string, value: any) => { + if (value instanceof Error) { + return { + name: value.name, + message: value.message, + stack: value.stack + }; + } + return value; + }; + + return JSON.stringify(this.options, replacer) == JSON.stringify(status.options, replacer); } /** diff --git a/tools/diagnostics-app/src/app/views/layout.tsx b/tools/diagnostics-app/src/app/views/layout.tsx index 97a3b5f2b..64531a11e 100644 --- a/tools/diagnostics-app/src/app/views/layout.tsx +++ b/tools/diagnostics-app/src/app/views/layout.tsx @@ -10,6 +10,7 @@ import UserIcon from '@mui/icons-material/VerifiedUser'; import WifiIcon from '@mui/icons-material/Wifi'; import { + Alert, AppBar, Box, Divider, @@ -192,6 +193,7 @@ export default function ViewsLayout({ children }: { children: React.ReactNode }) + {syncError ? Sync error detected: {syncError.message} : null} {children} From 641c9b489add4a1de670a87fee862b775ae400bd Mon Sep 17 00:00:00 2001 From: stevensJourney Date: Wed, 4 Jun 2025 17:52:11 +0200 Subject: [PATCH 26/26] update changeset --- .changeset/fuzzy-beers-occur.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.changeset/fuzzy-beers-occur.md b/.changeset/fuzzy-beers-occur.md index ab125ae07..295b64f23 100644 --- a/.changeset/fuzzy-beers-occur.md +++ b/.changeset/fuzzy-beers-occur.md @@ -1,5 +1,7 @@ --- -'@powersync/diagnostics-app': patch +'@powersync/diagnostics-app': minor --- -Fix bug where clicking signOut would not disconnect from the PowerSync service. Updated implementation to fetch sync errors from the SyncStatus. +- Added Sync error alert banner to all views. +- Fix bug where clicking signOut would not disconnect from the PowerSync service. +- Updated implementation to fetch sync errors from the SyncStatus.