simplify logic

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -462,8 +462,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   protected abstract runExclusive<T>(callback: () => Promise<T>): Promise<T>;
 
   protected async connectInternal() {
-    await this.disconnectInternal();
-
     let appliedOptions: PowerSyncConnectionOptions | null = null;
 
     /**
@@ -521,42 +519,25 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Connects to stream of events from the PowerSync instance.
    */
   async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
-    await this.waitForReady();
-
-    // 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;
-
     // This overrides options if present.
     this.pendingConnectionOptions = {
       connector,
       options: options ?? {}
     };
 
-    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();
-        }
+    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<PowerSyncDB
       // 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;

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -1,13 +1,15 @@
 import Logger, { ILogger } from 'js-logger';
 
+import { InternalProgressInformation } from 'src/db/crud/SyncProgress.js';
+import { DataStream } from 'src/utils/DataStream.js';
 import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
 import { AbortOperation } from '../../../utils/AbortOperation.js';
 import { BaseListener, BaseObserver, Disposable } from '../../../utils/BaseObserver.js';
 import { onAbortPromise, throttleLeadingTrailing } from '../../../utils/async.js';
 import { BucketChecksum, BucketDescription, BucketStorageAdapter, Checkpoint } from '../bucket/BucketStorageAdapter.js';
 import { CrudEntry } from '../bucket/CrudEntry.js';
 import { SyncDataBucket } from '../bucket/SyncDataBucket.js';
-import { AbstractRemote, SyncStreamOptions, FetchStrategy } from './AbstractRemote.js';
+import { AbstractRemote, FetchStrategy, SyncStreamOptions } from './AbstractRemote.js';
 import {
   BucketRequest,
   StreamingSyncLine,
@@ -19,8 +21,6 @@ import {
   isStreamingSyncCheckpointPartiallyComplete,
   isStreamingSyncData
 } from './streaming-sync-types.js';
-import { DataStream } from 'src/utils/DataStream.js';
-import { InternalProgressInformation } from 'src/db/crud/SyncProgress.js';
 
 export enum LockType {
   CRUD = 'crud',
@@ -341,12 +341,22 @@ The next upload iteration will be delayed.`);
       await this.disconnect();
     }
 
-    this.abortController = new AbortController();
+    const controller = new AbortController();
+    this.abortController = controller;
     this.streamingSyncPromise = this.streamingSync(this.abortController.signal, options);
 
     // Return a promise that resolves when the connection status is updated
     return new Promise<void>((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();
         }
       });
     });

packages/web/tests/stream.test.ts

@@ -182,6 +182,20 @@ function describeStreamingTests(createConnectedDatabase: () => Promise<Connected
       // 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);
+
+      // Now with random delays
+      for (let i = connectionAttempts; i >= 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 () => {

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();

packages/web/vitest.config.ts

@@ -50,7 +50,7 @@ const config: UserConfigExport = {
        */
       isolate: true,
       provider: 'playwright',
-      headless: true,
+      headless: false,
       instances: [
         {
           browser: 'chromium'