improve retry delay logic. Update diagnostics app error detection.

Author: stevensJourney

.changeset/fuzzy-beers-occur.md

@@ -0,0 +1,5 @@
+---
+'@powersync/diagnostics-app': patch
+---
+
+Fix bug where clicking signOut would not disconnect from the PowerSync service. Updated implementation to fetch sync errors from the SyncStatus.

packages/common/src/client/ConnectionManager.ts

@@ -122,7 +122,9 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
       if (this.pendingConnectionOptions) {
         // Pending options have been placed while connecting.
         // Need to reconnect.
-        this.connectingPromise = this.connectInternal().finally(checkConnection);
+        this.connectingPromise = this.connectInternal()
+          .catch(() => {})
+          .finally(checkConnection);
         return this.connectingPromise;
       } else {
         // Clear the connecting promise, done.
@@ -131,7 +133,9 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
       }
     };
 
-    this.connectingPromise ??= this.connectInternal().finally(checkConnection);
+    this.connectingPromise ??= this.connectInternal()
+      .catch(() => {})
+      .finally(checkConnection);
     return this.connectingPromise;
   }
 

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

@@ -442,6 +442,7 @@ The next upload iteration will be delayed.`);
      */
     while (true) {
       this.updateSyncStatus({ connecting: true });
+      let shouldDelayRetry = true;
       try {
         if (signal?.aborted) {
           break;
@@ -458,10 +459,10 @@ The next upload iteration will be delayed.`);
          * The nested abort controller will cleanup any open network requests and streams.
          * The WebRemote should only abort pending fetch requests or close active Readable streams.
          */
-        let delay = true;
+
         if (ex instanceof AbortOperation) {
           this.logger.warn(ex);
-          delay = false;
+          shouldDelayRetry = false;
           // A disconnect was requested, we should not delay since there is no explicit retry
         } else {
           this.logger.error(ex);
@@ -472,11 +473,6 @@ The next upload iteration will be delayed.`);
             downloadError: ex
           }
         });
-
-        // On error, wait a little before retrying
-        if (delay) {
-          await this.delayRetry();
-        }
       } finally {
         if (!signal.aborted) {
           nestedAbortController.abort(new AbortOperation('Closing sync stream network requests before retry.'));
@@ -487,6 +483,11 @@ The next upload iteration will be delayed.`);
           connected: false,
           connecting: true // May be unnecessary
         });
+
+        // On error, wait a little before retrying
+        if (shouldDelayRetry) {
+          await this.delayRetry(nestedAbortController.signal);
+        }
       }
     }
 
@@ -854,7 +855,29 @@ The next upload iteration will be delayed.`);
     this.iterateListeners((cb) => cb.statusUpdated?.(options));
   }
 
-  private async delayRetry() {
-    return new Promise((resolve) => setTimeout(resolve, this.options.retryDelayMs));
+  private async delayRetry(signal?: AbortSignal): Promise<void> {
+    return new Promise((resolve) => {
+      if (signal?.aborted) {
+        // If the signal is already aborted, resolve immediately
+        resolve();
+        return;
+      }
+
+      const { retryDelayMs } = this.options;
+
+      let timeoutId: ReturnType<typeof setTimeout> | undefined;
+
+      const endDelay = () => {
+        resolve();
+        if (timeoutId) {
+          clearTimeout(timeoutId);
+          timeoutId = undefined;
+        }
+        signal?.removeEventListener('abort', endDelay);
+      };
+
+      signal?.addEventListener('abort', endDelay, { once: true });
+      timeoutId = setTimeout(endDelay, retryDelayMs);
+    });
   }
 }

tools/diagnostics-app/src/app/views/layout.tsx

@@ -6,8 +6,8 @@ import SouthIcon from '@mui/icons-material/South';
 import StorageIcon from '@mui/icons-material/Storage';
 import TableChartIcon from '@mui/icons-material/TableChart';
 import TerminalIcon from '@mui/icons-material/Terminal';
-import WifiIcon from '@mui/icons-material/Wifi';
 import UserIcon from '@mui/icons-material/VerifiedUser';
+import WifiIcon from '@mui/icons-material/Wifi';
 
 import {
   AppBar,
@@ -34,7 +34,7 @@ import {
   SYNC_DIAGNOSTICS_ROUTE
 } from '@/app/router';
 import { useNavigationPanel } from '@/components/navigation/NavigationPanelContext';
-import { signOut, sync, syncErrorTracker } from '@/library/powersync/ConnectionManager';
+import { signOut, sync } from '@/library/powersync/ConnectionManager';
 import { usePowerSync } from '@powersync/react';
 import { useNavigate } from 'react-router-dom';
 
@@ -103,19 +103,12 @@ export default function ViewsLayout({ children }: { children: React.ReactNode })
     const l = sync.registerListener({
       statusChanged: (status) => {
         setSyncStatus(status);
+        setSyncError(status.dataFlowStatus.downloadError ?? null);
       }
     });
     return () => l();
   }, []);
 
-  React.useEffect(() => {
-    const l = syncErrorTracker.registerListener({
-      lastErrorUpdated(error) {
-        setSyncError(error);
-      }
-    });
-    return () => l();
-  }, []);
   const drawerWidth = 320;
 
   const drawer = (
@@ -220,4 +213,4 @@ namespace S {
     object-fit: contain;
     padding: 20px;
   `;
-}
+}

tools/diagnostics-app/src/app/views/sync-diagnostics.tsx

@@ -1,5 +1,5 @@
 import { NavigationPage } from '@/components/navigation/NavigationPage';
-import { clearData, db, sync, syncErrorTracker } from '@/library/powersync/ConnectionManager';
+import { clearData, db, sync } from '@/library/powersync/ConnectionManager';
 import {
   Box,
   Button,
@@ -73,7 +73,6 @@ FROM local_bucket_data local`;
 export default function SyncDiagnosticsPage() {
   const [bucketRows, setBucketRows] = React.useState<null | any[]>(null);
   const [tableRows, setTableRows] = React.useState<null | any[]>(null);
-  const [syncError, setSyncError] = React.useState<Error | null>(syncErrorTracker.lastSyncError);
   const [lastSyncedAt, setlastSyncedAt] = React.useState<Date | null>(null);
 
   const bucketRowsLoading = bucketRows == null;
@@ -125,15 +124,6 @@ export default function SyncDiagnosticsPage() {
     };
   }, []);
 
-  React.useEffect(() => {
-    const l = syncErrorTracker.registerListener({
-      lastErrorUpdated(error) {
-        setSyncError(error);
-      }
-    });
-    return () => l();
-  }, []);
-
   const columns: GridColDef[] = [
     { field: 'name', headerName: 'Name', flex: 2 },
     { field: 'tables', headerName: 'Table(s)', flex: 1, type: 'text' },

tools/diagnostics-app/src/library/powersync/ConnectionManager.ts

@@ -1,20 +1,19 @@
 import {
   BaseListener,
-  BaseObserver,
+  createBaseLogger,
+  LogLevel,
   PowerSyncDatabase,
-  WebRemote,
-  WebStreamingSyncImplementation,
-  WebStreamingSyncImplementationOptions,
-  WASQLiteOpenFactory,
   TemporaryStorageOption,
+  WASQLiteOpenFactory,
   WASQLiteVFS,
-  createBaseLogger,
-  LogLevel
+  WebRemote,
+  WebStreamingSyncImplementation,
+  WebStreamingSyncImplementationOptions
 } from '@powersync/web';
+import { safeParse } from '../safeParse/safeParse';
 import { DynamicSchemaManager } from './DynamicSchemaManager';
 import { RecordingStorageAdapter } from './RecordingStorageAdapter';
 import { TokenConnector } from './TokenConnector';
-import { safeParse } from '../safeParse/safeParse';
 
 const baseLogger = createBaseLogger();
 baseLogger.useDefaults();
@@ -62,31 +61,6 @@ export interface SyncErrorListener extends BaseListener {
   lastErrorUpdated?: ((error: Error) => void) | undefined;
 }
 
-class SyncErrorTracker extends BaseObserver<SyncErrorListener> {
-  public lastSyncError: Error | null = null;
-
-  constructor() {
-    super();
-    // Big hack: Use the logger to get access to connection errors
-    const defaultHandler = baseLogger.createDefaultHandler();
-    baseLogger.setHandler((messages, context) => {
-      defaultHandler(messages, context);
-      if (context.name == 'PowerSyncStream' && context.level.name == 'ERROR') {
-        if (messages[0] instanceof Error) {
-          this.lastSyncError = messages[0];
-        } else {
-          this.lastSyncError = new Error('' + messages[0]);
-        }
-        this.iterateListeners((listener) => {
-          listener.lastErrorUpdated?.(this.lastSyncError!);
-        });
-      }
-    });
-  }
-}
-
-export const syncErrorTracker = new SyncErrorTracker();
-
 if (connector.hasCredentials()) {
   connect();
 }
@@ -95,11 +69,10 @@ export async function connect() {
   const params = getParams();
   await sync.connect({ params });
   if (!sync.syncStatus.connected) {
+    const error = sync.syncStatus.dataFlowStatus.downloadError ?? new Error('Failed to connect');
     // Disconnect but don't wait for it
     sync.disconnect();
-    throw syncErrorTracker.lastSyncError ?? new Error('Failed to connect');
-  } else {
-    syncErrorTracker.lastSyncError = null;
+    throw error;
   }
 }
 
@@ -120,6 +93,7 @@ export async function disconnect() {
 
 export async function signOut() {
   connector.clearCredentials();
+  await disconnect();
   await db.disconnectAndClear();
   await schemaManager.clear();
 }