cleanup connection manager

Author: stevensJourney

packages/common/src/client/AbstractPowerSyncDatabase.ts

@@ -478,7 +478,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Use {@link connect} to connect again.
    */
   async disconnect() {
-    await this.waitForReady();
     return this.connectionManager.disconnect();
   }
 

packages/common/src/client/ConnectionManager.ts

@@ -100,7 +100,8 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
     // 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<ConnectionManagerListener> {
      * 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');
   }
 
   /**

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<Connected
           const call = spy.mock.lastCall![1] as PowerSyncConnectionOptions;
           expect(call.params!['count']).eq(0);
         },
-        { timeout: 2000, interval: 100 }
+        { timeout: 4000, interval: 100 }
       );
 
       expect(