[Fix] concurrent connections issue (#101)

Author: stevensJourney

.changeset/lemon-zoos-sort.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-web': patch
+---
+
+Added some serialization checks for broadcasted logs from shared web worker. Unserializable items will return a warning.

.changeset/lucky-planes-prove.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-common': patch
+---
+
+Fixed issue where sync stream exceptions would not close previous streaming connections.

.changeset/modern-terms-admire.md

@@ -0,0 +1,5 @@
+---
+'@journeyapps/powersync-sdk-web': patch
+---
+
+Fixed issue where SyncBucketStorage logs would not be broadcasted from the shared sync worker to individual tabs.

packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -220,7 +220,10 @@ export abstract class AbstractStreamingSyncImplementation
     }
   }
 
-  connect() {
+  async connect() {
+    if (this.abortController) {
+      await this.disconnect();
+    }
     this.abortController = new AbortController();
     this.streamingSync(this.abortController.signal);
     return this.waitForStatus({ connected: true });
@@ -231,6 +234,8 @@ export abstract class AbstractStreamingSyncImplementation
       throw new Error('Disconnect not possible');
     }
     this.abortController.abort('Disconnected');
+    this.abortController = null;
+    this.updateSyncStatus({ connected: false });
   }
 
   /**
@@ -249,7 +254,14 @@ export abstract class AbstractStreamingSyncImplementation
       crudUpdate: () => this.triggerCrudUpload()
     });
 
+    /**
+     * Create a new abort controller which aborts items downstream.
+     * This is needed to close any previous connections on exception.
+     */
+    let nestedAbortController = new AbortController();
+
     signal.addEventListener('abort', () => {
+      nestedAbortController.abort();
       this.crudUpdateListener?.();
       this.crudUpdateListener = undefined;
       this.updateSyncStatus({
@@ -265,7 +277,10 @@ export abstract class AbstractStreamingSyncImplementation
         if (signal?.aborted) {
           break;
         }
-        await this.streamingSyncIteration(signal);
+        const { retry } = await this.streamingSyncIteration(nestedAbortController.signal);
+        if (!retry) {
+          break;
+        }
         // Continue immediately
       } catch (ex) {
         this.logger.error(ex);
@@ -274,8 +289,17 @@ export abstract class AbstractStreamingSyncImplementation
         });
         // On error, wait a little before retrying
         await this.delayRetry();
+      } finally {
+        // Abort any open network requests. Create a new nested controller for retry.
+        nestedAbortController.abort();
+        nestedAbortController = new AbortController();
       }
     }
+
+    // Mark as disconnected if here
+    if (this.abortController) {
+      await this.disconnect();
+    }
   }
 
   protected async streamingSyncIteration(signal: AbortSignal, progress?: () => void): Promise<{ retry?: boolean }> {

packages/powersync-sdk-web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -165,8 +165,6 @@ export class SharedWebStreamingSyncImplementation extends WebStreamingSyncImplem
       data: {}
     };
 
-    (localStorage as any).setItem('posting close' + Math.random(), `$}`);
-
     this.messagePort.postMessage(closeMessagePayload);
 
     // Release the proxy

packages/powersync-sdk-web/src/worker/sync/BroadcastLogger.ts

@@ -40,17 +40,17 @@ export class BroadcastLogger implements ILogger {
 
   log(...x: any[]): void {
     console.log(...x);
-    this.clients.forEach((p) => p.clientProvider.log(...x));
+    this.sanitizeArgs(x, (params) => this.clients.forEach((p) => p.clientProvider.log(...params)));
   }
 
   warn(...x: any[]): void {
     console.warn(...x);
-    this.clients.forEach((p) => p.clientProvider.warn(...x));
+    this.sanitizeArgs(x, (params) => this.clients.forEach((p) => p.clientProvider.warn(...params)));
   }
 
   error(...x: any[]): void {
     console.error(...x);
-    this.clients.forEach((p) => p.clientProvider.error(...x));
+    this.sanitizeArgs(x, (params) => this.clients.forEach((p) => p.clientProvider.error(...params)));
   }
 
   time(label: string): void {
@@ -76,4 +76,22 @@ export class BroadcastLogger implements ILogger {
     // Levels are not adjustable on this level.
     return true;
   }
+
+  /**
+   * Guards against any logging errors.
+   * We don't want a logging exception to cause further issues upstream
+   */
+  private sanitizeArgs(x: any[], handler: (...params: any[]) => void) {
+    const sanitizedParams = x.map((param) => {
+      try {
+        // Try and clone here first. If it fails it won't be passable over a MessagePort
+        return structuredClone(x);
+      } catch (ex) {
+        console.error(ex);
+        return 'Could not serialize log params. Check shared worker logs for more details.';
+      }
+    });
+
+    return handler(...sanitizedParams);
+  }
 }

packages/powersync-sdk-web/src/worker/sync/SharedSyncImplementation.ts

@@ -125,7 +125,8 @@ export class SharedSyncImplementation
           flags: { enableMultiTabs: true },
           logger: this.broadCastLogger
         }),
-        new Mutex()
+        new Mutex(),
+        this.broadCastLogger
       ),
       remote: new WebRemote({
         fetchCredentials: async () => {