cleanup logic

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -464,6 +464,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   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.
@@ -474,13 +477,15 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       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;
@@ -510,6 +515,10 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       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!);
@@ -519,25 +528,44 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Connects to stream of events from the PowerSync instance.
    */
   async connect(connector: PowerSyncBackendConnector, options?: PowerSyncConnectionOptions) {
-    // This overrides options if present.
+    // 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();
-    await this.disconnectInternal();
 
-    const chain = (result) => {
+    // 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<void> => {
       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<PowerSyncDB
   protected async disconnectInternal() {
     if (this.disconnectingPromise) {
       // A disconnect is already in progress
-      return await this.disconnectingPromise;
+      return this.disconnectingPromise;
     }
+
     // Wait if a sync stream implementation is being created before closing it
-    // (it must be assigned before we can properly dispose it)
+    // (syncStreamImplementation must be assigned before we can properly dispose it)
     await this.syncStreamInitPromise;
 
     this.disconnectingPromise = (async () => {

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<void>((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

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<Connected
 
     it('PowerSync reconnect multiple connect calls', async () => {
       // 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<any>[] = [];
+
+      // 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<Connected
       // are in the same for loop
       expect(spy.mock.calls.length).lessThan(connectionAttempts);
 
-      // Now with random delays
+      // Now with random awaited delays between unawaited calls
       for (let i = connectionAttempts; i >= 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<Connected
         },
         { timeout: 2000, interval: 100 }
       );
+
+      console.log(`Generated stream a total of ${spy.mock.calls.length} times`);
+      expect(spy.mock.calls.length).lessThanOrEqual(2 * connectionAttempts);
+
+      // The last request should make a network request with the client params
+      await vi.waitFor(
+        () => {
+          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 () => {

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 {