fix multiple instance unit tests

Author: stevensJourney

packages/common/src/client/sync/bucket/BucketStorageAdapter.ts

@@ -1,4 +1,4 @@
-import { BaseListener, BaseObserver, Disposable } from '../../../utils/BaseObserver.js';
+import { BaseListener, BaseObserverInterface, Disposable } from '../../../utils/BaseObserver.js';
 import { CrudBatch } from './CrudBatch.js';
 import { CrudEntry, OpId } from './CrudEntry.js';
 import { SyncDataBatch } from './SyncDataBatch.js';
@@ -72,7 +72,7 @@ export interface BucketStorageListener extends BaseListener {
   crudUpdate: () => void;
 }
 
-export interface BucketStorageAdapter extends BaseObserver<BucketStorageListener>, Disposable {
+export interface BucketStorageAdapter extends BaseObserverInterface<BucketStorageListener>, Disposable {
   init(): Promise<void>;
   saveSyncData(batch: SyncDataBatch, fixedKeyFormat?: boolean): Promise<void>;
   removeBuckets(buckets: string[]): Promise<void>;

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

@@ -1,11 +1,11 @@
 import Logger, { ILogger } from 'js-logger';
 
-import { DataStream } from '../../../utils/DataStream.js';
-import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
 import { FULL_SYNC_PRIORITY, InternalProgressInformation } from '../../../db/crud/SyncProgress.js';
 import * as sync_status from '../../../db/crud/SyncStatus.js';
+import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
 import { AbortOperation } from '../../../utils/AbortOperation.js';
-import { BaseListener, BaseObserver, Disposable } from '../../../utils/BaseObserver.js';
+import { BaseListener, BaseObserver, BaseObserverInterface, Disposable } from '../../../utils/BaseObserver.js';
+import { DataStream } from '../../../utils/DataStream.js';
 import { onAbortPromise, throttleLeadingTrailing } from '../../../utils/async.js';
 import {
   BucketChecksum,
@@ -17,6 +17,7 @@ import {
 import { CrudEntry } from '../bucket/CrudEntry.js';
 import { SyncDataBucket } from '../bucket/SyncDataBucket.js';
 import { AbstractRemote, FetchStrategy, SyncStreamOptions } from './AbstractRemote.js';
+import { EstablishSyncStream, Instruction, SyncPriorityStatus } from './core-instruction.js';
 import {
   BucketRequest,
   StreamingSyncLine,
@@ -28,7 +29,6 @@ import {
   isStreamingSyncCheckpointPartiallyComplete,
   isStreamingSyncData
 } from './streaming-sync-types.js';
-import { EstablishSyncStream, Instruction, SyncPriorityStatus } from './core-instruction.js';
 
 export enum LockType {
   CRUD = 'crud',
@@ -170,7 +170,9 @@ export interface AdditionalConnectionOptions {
 /** @internal */
 export type RequiredAdditionalConnectionOptions = Required<AdditionalConnectionOptions>;
 
-export interface StreamingSyncImplementation extends BaseObserver<StreamingSyncImplementationListener>, Disposable {
+export interface StreamingSyncImplementation
+  extends BaseObserverInterface<StreamingSyncImplementationListener>,
+    Disposable {
   /**
    * Connects to the sync service
    */
@@ -222,6 +224,9 @@ export abstract class AbstractStreamingSyncImplementation
   protected _lastSyncedAt: Date | null;
   protected options: AbstractStreamingSyncImplementationOptions;
   protected abortController: AbortController | null;
+  // In rare cases, mostly for tests, uploads can be triggered without being properly connected.
+  // This allows ensuring that all upload processes can be aborted.
+  protected uploadAbortController: AbortController | null;
   protected crudUpdateListener?: () => void;
   protected streamingSyncPromise?: Promise<void>;
 
@@ -315,8 +320,10 @@ export abstract class AbstractStreamingSyncImplementation
   }
 
   async dispose() {
+    super.dispose();
     this.crudUpdateListener?.();
     this.crudUpdateListener = undefined;
+    this.uploadAbortController?.abort();
   }
 
   abstract obtainLock<T>(lockOptions: LockOptions<T>): Promise<T>;
@@ -341,7 +348,17 @@ export abstract class AbstractStreamingSyncImplementation
          */
         let checkedCrudItem: CrudEntry | undefined;
 
-        while (true) {
+        const controller = new AbortController();
+        this.uploadAbortController = controller;
+        this.abortController?.signal.addEventListener(
+          'abort',
+          () => {
+            controller.abort();
+          },
+          { once: true }
+        );
+
+        while (!controller.signal.aborted) {
           this.updateSyncStatus({
             dataFlow: {
               uploading: true
@@ -381,7 +398,7 @@ The next upload iteration will be delayed.`);
                 uploadError: ex
               }
             });
-            await this.delayRetry();
+            await this.delayRetry(controller.signal);
             if (!this.isConnected) {
               // Exit the upload loop if the sync stream is no longer connected
               break;
@@ -397,6 +414,7 @@ The next upload iteration will be delayed.`);
             });
           }
         }
+        this.uploadAbortController = null;
       }
     });
   }

packages/common/src/utils/BaseObserver.ts

@@ -1,5 +1,5 @@
 export interface Disposable {
-  dispose: () => Promise<void>;
+  dispose: () => Promise<void> | void;
 }
 
 export type BaseListener = Record<string, ((...event: any) => any) | undefined>;
@@ -13,6 +13,10 @@ export class BaseObserver<T extends BaseListener = BaseListener> implements Base
 
   constructor() {}
 
+  dispose(): void {
+    this.listeners.clear();
+  }
+
   /**
    * Register a listener for updates to the PowerSync client.
    */

packages/web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -207,6 +207,8 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
   async dispose(): Promise<void> {
     await this.waitForReady();
 
+    await super.dispose();
+
     await new Promise<void>((resolve) => {
       // Listen for the close acknowledgment from the worker
       this.messagePort.addEventListener('message', (event) => {

packages/web/src/worker/sync/SharedSyncImplementation.ts

@@ -297,7 +297,6 @@ export class SharedSyncImplementation
       });
 
       const shouldReconnect = !!this.connectionManager.syncStreamImplementation && this.ports.length > 0;
-
       return {
         shouldReconnect,
         trackedPort
@@ -473,10 +472,8 @@ export class SharedSyncImplementation
    */
   private async _testUpdateAllStatuses(status: SyncStatusOptions) {
     if (!this.connectionManager.syncStreamImplementation) {
-      // This is just for testing purposes
-      this.connectionManager.syncStreamImplementation = this.generateStreamingImplementation();
+      throw new Error('Cannot update status without a sync stream implementation');
     }
-
     // Only assigning, don't call listeners for this test
     this.connectionManager.syncStreamImplementation!.syncStatus = new SyncStatus(status);
     this.updateAllStatuses(status);

packages/web/tests/multiple_instances.test.ts

@@ -132,11 +132,12 @@ describe('Multiple Instances', { sequential: true }, () => {
         await connector1.uploadData(db);
       },
       identifier,
+      retryDelayMs: 90_000, // Large delay to allow for testing
       db: db.database as WebDBAdapter
     };
 
     const stream1 = new SharedWebStreamingSyncImplementation(syncOptions1);
-
+    await stream1.connect();
     // Generate the second streaming sync implementation
     const connector2 = new TestConnector();
     const syncOptions2: SharedWebStreamingSyncImplementationOptions = {
@@ -188,20 +189,25 @@ describe('Multiple Instances', { sequential: true }, () => {
       triggerUpload1 = resolve;
     });
 
-    // Create the first streaming client
-    const stream1 = new SharedWebStreamingSyncImplementation({
+    const sharedSyncOptions = {
       adapter: new SqliteBucketStorage(db.database, new Mutex()),
       remote: new WebRemote(connector1),
-      uploadCrud: async () => {
-        triggerUpload1();
-        connector1.uploadData(db);
-      },
       db: db.database as WebDBAdapter,
       identifier,
-      retryDelayMs: 100,
+      // The large delay here allows us to test between connection retries
+      retryDelayMs: 90_000,
       flags: {
         broadcastLogs: true
       }
+    };
+
+    // Create the first streaming client
+    const stream1 = new SharedWebStreamingSyncImplementation({
+      ...sharedSyncOptions,
+      uploadCrud: async () => {
+        triggerUpload1();
+        connector1.uploadData(db);
+      }
     });
 
     // Generate the second streaming sync implementation
@@ -216,18 +222,11 @@ describe('Multiple Instances', { sequential: true }, () => {
     });
 
     const stream2 = new SharedWebStreamingSyncImplementation({
-      adapter: new SqliteBucketStorage(db.database, new Mutex()),
-      remote: new WebRemote(connector1),
+      ...sharedSyncOptions,
       uploadCrud: async () => {
         triggerUpload2();
         connector2.uploadData(db);
-      },
-      identifier,
-      retryDelayMs: 100,
-      flags: {
-        broadcastLogs: true
-      },
-      db: db.database as WebDBAdapter
+      }
     });
 
     // Waits for the stream to be marked as connected
@@ -243,7 +242,9 @@ describe('Multiple Instances', { sequential: true }, () => {
     });
 
     // hack to set the status to connected for tests
-    (stream1 as any)['_testUpdateStatus'](new SyncStatus({ connected: true }));
+    await stream1.connect();
+    // Hack, set the status to connected in order to trigger the upload
+    await (stream1 as any)['_testUpdateStatus'](new SyncStatus({ connected: true }));
 
     // The status in the second stream client should be updated
     await stream2UpdatedPromise;
@@ -256,7 +257,6 @@ describe('Multiple Instances', { sequential: true }, () => {
       '[email protected]'
     ]);
 
-    // Manual trigger since tests don't entirely configure watches for ps_crud
     stream1.triggerCrudUpload();
     // The second connector should be called to upload
     await upload2TriggeredPromise;
@@ -265,14 +265,21 @@ describe('Multiple Instances', { sequential: true }, () => {
     expect(spy2).toHaveBeenCalledOnce();
 
     // Close the second client, leaving only the first one
+    /**
+     * This test is a bit hacky. If we dispose the second client, the shared sync worker
+     * will try and reconnect, but we don't actually want it to do that since that connection attempt
+     * will fail and it will report as `connected:false`.
+     * We can hack as disconnected for now.
+     */
     await stream2.dispose();
 
+    // Hack, set the status to connected in order to trigger the upload
+    await (stream1 as any)['_testUpdateStatus'](new SyncStatus({ connected: true }));
     stream1.triggerCrudUpload();
     // It should now upload from the first client
     await upload1TriggeredPromise;
 
     expect(spy1).toHaveBeenCalledOnce();
-
     await stream1.dispose();
   });
 });