Shared worker: Immediate reconnect on connection interrupt

Author: simolus3

.changeset/purple-comics-explain.md

@@ -0,0 +1,6 @@
+---
+'@powersync/common': patch
+'@powersync/web': patch
+---
+
+Using the Rust sync client on the web, immediately reconnect when the underlying SQLite connection is changed.

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

@@ -28,14 +28,7 @@ import {
   isStreamingSyncCheckpointPartiallyComplete,
   isStreamingSyncData
 } from './streaming-sync-types.js';
-import {
-  extractBsonObjects,
-  extractJsonLines,
-  injectable,
-  InjectableIterator,
-  map,
-  SimpleAsyncIterator
-} from '../../../utils/stream_transform.js';
+import { injectable, InjectableIterator, map, SimpleAsyncIterator } from '../../../utils/stream_transform.js';
 import type { BSON } from 'bson';
 
 export enum LockType {
@@ -216,6 +209,7 @@ export interface StreamingSyncImplementation
   waitForStatus(status: SyncStatusOptions): Promise<void>;
   waitUntilStatusMatches(predicate: (status: SyncStatus) => boolean): Promise<void>;
   updateSubscriptions(subscriptions: SubscribedStream[]): void;
+  markConnectionMayHaveChanged(): void;
 }
 
 export const DEFAULT_CRUD_UPLOAD_THROTTLE_MS = 1000;
@@ -263,6 +257,7 @@ export abstract class AbstractStreamingSyncImplementation
   protected streamingSyncPromise?: Promise<void>;
   protected logger: ILogger;
   private activeStreams: SubscribedStream[];
+  private connectionMayHaveChanged = false;
 
   private isUploadingCrud: boolean = false;
   private notifyCompletedUploads?: () => void;
@@ -579,6 +574,10 @@ The next upload iteration will be delayed.`);
           this.logger.warn(ex);
           shouldDelayRetry = false;
           // A disconnect was requested, we should not delay since there is no explicit retry
+        } else if (this.connectionMayHaveChanged && (ex as Error).message?.indexOf('No iteration is active') >= 0) {
+          this.connectionMayHaveChanged = false;
+          this.logger.info('Sync error after changed connection, retrying immediately');
+          shouldDelayRetry = false;
         } else {
           this.logger.error(ex);
         }
@@ -614,6 +613,17 @@ The next upload iteration will be delayed.`);
     this.updateSyncStatus({ connected: false, connecting: false });
   }
 
+  markConnectionMayHaveChanged() {
+    // By setting this field, we'll immediately retry if the next sync connection causes an error triggered by us not
+    // having an active sync iteration on the connection in use.
+    this.connectionMayHaveChanged = true;
+
+    // This triggers a `powersync_control` invocation if a sync iteration is currently active. This is a cheap call to
+    // make when no subscriptions have actually changed, we're mainly interested in this immediately throwing if no
+    // iteration is active. That allows us to reconnect ASAP, instead of having to wait for the next sync line.
+    this.handleActiveStreamsChange?.();
+  }
+
   private async collectLocalBucketState(): Promise<[BucketRequest[], Map<string, BucketDescription | null>]> {
     const bucketEntries = await this.options.adapter.getBucketStates();
     const req: BucketRequest[] = bucketEntries.map((entry) => ({
@@ -1041,6 +1051,10 @@ The next upload iteration will be delayed.`);
         rawResponse
       );
 
+      if (op != PowerSyncControlCommand.STOP) {
+        // Evidently we have a working connection here, otherwise powersync_control would have failed.
+        syncImplementation.connectionMayHaveChanged = false;
+      }
       await handleInstructions(JSON.parse(rawResponse));
     }
 

packages/node/tests/sync.test.ts

@@ -117,6 +117,24 @@ describe('Sync', () => {
       expect.objectContaining({ key: 'lists/3/subkey_3' })
     ]);
   });
+
+  mockSyncServiceTest('reconnects immediately after changed connection', async ({ syncService }) => {
+    let database = await syncService.createDatabase();
+    database.connect(new TestConnector(), {
+      clientImplementation: SyncClientImplementation.RUST,
+      connectionMethod: SyncStreamConnectionMethod.HTTP,
+      // This large retry delay is to provoke test timeouts if the don't immediately reconnect.
+      retryDelayMs: 60_000
+    });
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+
+    // Replicate what we'd see on the web when switching connections in the shared sync worker: The sync client would
+    // suddenly see a database without an active sync iteration.
+    await database.execute('SELECT powersync_control(?, null)', ['stop']);
+    database.syncStreamImplementation!.markConnectionMayHaveChanged();
+    await database.waitForStatus((s) => !s.connected);
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+  });
 });
 
 function defineSyncTests(impl: SyncClientImplementation, bson: boolean) {

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

@@ -85,4 +85,9 @@ export class SSRStreamingSyncImplementation extends BaseObserver implements Stre
    * No-op in SSR mode.
    */
   updateSubscriptions(): void {}
+
+  /**
+   * No-op in SSR mode.
+   */
+  markConnectionMayHaveChanged(): void {}
 }

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

@@ -618,14 +618,10 @@ export class SharedSyncImplementation extends BaseObserver<SharedSyncImplementat
           const impl = sharedSync.connectionManager.syncStreamImplementation! as WebStreamingSyncImplementation;
           impl?.triggerCrudUpload();
 
-          /**
-           * FIXME or IMPROVE ME
-           * The Rust client implementation stores sync state on the connection level.
-           * Reopening the database causes a state machine error which should cause the
-           * StreamingSyncImplementation to reconnect. It would be nicer if we could trigger
-           * this reconnect earlier.
-           * This reconnect is not required for IndexedDB.
-           */
+          // The Rust client implementation stores sync state on the connection level. Reopening the database causes a
+          // disruption of the connection state and forces us to reconnect. We want to do that as soon as possible to
+          // minimize downtime.
+          impl?.markConnectionMayHaveChanged();
         }
       }