feat: Exposing upload and download errors in SyncStatus (#550)

Author: Chriztiaan

.changeset/unlucky-flies-clap.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': minor
+---
+
+Added `downloadError` and `uploadError` members to `SyncDataFlowStatus` of `SyncStatus`.

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

@@ -290,6 +290,11 @@ The next upload iteration will be delayed.`);
 
               checkedCrudItem = nextCrudItem;
               await this.options.uploadCrud();
+              this.updateSyncStatus({
+                dataFlow: {
+                  uploadError: undefined
+                }
+              });
             } else {
               // Uploading is completed
               await this.options.adapter.updateLocalTarget(() => this.getWriteCheckpoint());
@@ -299,7 +304,8 @@ The next upload iteration will be delayed.`);
             checkedCrudItem = undefined;
             this.updateSyncStatus({
               dataFlow: {
-                uploading: false
+                uploading: false,
+                uploadError: ex
               }
             });
             await this.delayRetry();
@@ -453,6 +459,12 @@ The next upload iteration will be delayed.`);
           this.logger.error(ex);
         }
 
+        this.updateSyncStatus({
+          dataFlow: {
+            downloadError: ex
+          }
+        });
+
         // On error, wait a little before retrying
         await this.delayRetry();
       } finally {
@@ -588,7 +600,8 @@ The next upload iteration will be delayed.`);
                 connected: true,
                 lastSyncedAt: new Date(),
                 dataFlow: {
-                  downloading: false
+                  downloading: false,
+                  downloadError: undefined
                 }
               });
             }
@@ -688,7 +701,10 @@ The next upload iteration will be delayed.`);
               this.updateSyncStatus({
                 connected: true,
                 lastSyncedAt: new Date(),
-                priorityStatusEntries: []
+                priorityStatusEntries: [],
+                dataFlow: {
+                  downloadError: undefined
+                }
               });
             } else if (validatedCheckpoint === targetCheckpoint) {
               const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint!);
@@ -707,7 +723,8 @@ The next upload iteration will be delayed.`);
                   lastSyncedAt: new Date(),
                   priorityStatusEntries: [],
                   dataFlow: {
-                    downloading: false
+                    downloading: false,
+                    downloadError: undefined
                   }
                 });
               }

packages/common/src/db/crud/SyncStatus.ts

@@ -1,6 +1,17 @@
 export type SyncDataFlowStatus = Partial<{
   downloading: boolean;
   uploading: boolean;
+  /**
+   * Error during downloading (including connecting).
+   *
+   * Cleared on the next successful data download.
+   */
+  downloadError?: Error;
+  /**
+   * Error during uploading.
+   * Cleared on the next successful upload.
+   */
+  uploadError?: Error;
 }>;
 
 export interface SyncPriorityStatus {
@@ -112,7 +123,7 @@ export class SyncStatus {
 
   getMessage() {
     const dataFlow = this.dataFlowStatus;
-    return `SyncStatus<connected: ${this.connected} connecting: ${this.connecting} lastSyncedAt: ${this.lastSyncedAt} hasSynced: ${this.hasSynced}. Downloading: ${dataFlow.downloading}. Uploading: ${dataFlow.uploading}`;
+    return `SyncStatus<connected: ${this.connected} connecting: ${this.connecting} lastSyncedAt: ${this.lastSyncedAt} hasSynced: ${this.hasSynced}. Downloading: ${dataFlow.downloading}. Uploading: ${dataFlow.uploading}. UploadError: ${dataFlow.uploadError}, DownloadError?: ${dataFlow.downloadError}>`;
   }
 
   toJSON(): SyncStatusOptions {

packages/web/tests/sync_status.test.ts

@@ -0,0 +1,99 @@
+import { describe, expect, it } from 'vitest';
+import { ConnectedDatabaseUtils, generateConnectedDatabase } from './utils/generateConnectedDatabase';
+
+const UPLOAD_TIMEOUT_MS = 3000;
+
+describe(
+  'Sync Status when streaming',
+  { sequential: true },
+  describeSyncStatusStreamingTests(() =>
+    generateConnectedDatabase({
+      powerSyncOptions: {
+        flags: {
+          useWebWorker: false,
+          enableMultiTabs: false
+        }
+      }
+    })
+  )
+);
+
+function describeSyncStatusStreamingTests(createConnectedDatabase: () => Promise<ConnectedDatabaseUtils>) {
+  return () => {
+    it('Should have downloadError on stream failure', async () => {
+      const { powersync, waitForStream, remote, connector } = await createConnectedDatabase();
+      remote.errorOnStreamStart = true;
+
+      // Making sure the field change takes effect
+      const newStream = waitForStream();
+      remote.streamController?.close();
+      await newStream;
+
+      let resolveDownloadError: () => void;
+      const downloadErrorPromise = new Promise<void>((resolve) => {
+        resolveDownloadError = resolve;
+      });
+      let receivedUploadError = false;
+
+      powersync.registerListener({
+        statusChanged: (status) => {
+          if (status.dataFlowStatus.downloadError) {
+            resolveDownloadError();
+            receivedUploadError = true;
+          }
+        }
+      });
+
+      // Download error should be specified
+      await downloadErrorPromise;
+    });
+
+    it('Should have uploadError on failed uploads', async () => {
+      const { powersync, uploadSpy } = await createConnectedDatabase();
+      expect(powersync.connected).toBe(true);
+
+      let uploadCounter = 0;
+      // This test will throw an exception a few times before successfully uploading
+      const throwCounter = 2;
+      uploadSpy.mockImplementation(async (db) => {
+        if (uploadCounter++ < throwCounter) {
+          throw new Error(`Force upload error`);
+        }
+        // Now actually do the upload
+        const tx = await db.getNextCrudTransaction();
+        await tx?.complete();
+      });
+
+      let resolveUploadError: () => void;
+      const uploadErrorPromise = new Promise<void>((resolve) => {
+        resolveUploadError = resolve;
+      });
+      let receivedUploadError = false;
+
+      let resolveClearedUploadError: () => void;
+      const clearedUploadErrorPromise = new Promise<void>((resolve) => {
+        resolveClearedUploadError = resolve;
+      });
+
+      powersync.registerListener({
+        statusChanged: (status) => {
+          if (status.dataFlowStatus.uploadError) {
+            resolveUploadError();
+            receivedUploadError = true;
+          } else if (receivedUploadError) {
+            resolveClearedUploadError();
+          }
+        }
+      });
+
+      // do something which should trigger an upload
+      await powersync.execute('INSERT INTO users (id, name) VALUES (uuid(), ?)', ['name']);
+
+      // Upload error should be specified
+      await uploadErrorPromise;
+
+      // Upload error should be cleared after successful upload
+      await clearedUploadErrorPromise;
+    });
+  };
+}

packages/web/tests/utils/MockStreamOpenFactory.ts

@@ -34,7 +34,7 @@ export class TestConnector implements PowerSyncBackendConnector {
 
 export class MockRemote extends AbstractRemote {
   streamController: ReadableStreamDefaultController<StreamingSyncLine> | null;
-
+  errorOnStreamStart = false;
   constructor(
     connector: RemoteConnector,
     protected onStreamRequested: () => void
@@ -61,6 +61,7 @@ export class MockRemote extends AbstractRemote {
     }
     throw new Error('Not implemented');
   }
+
   async postStreaming(
     path: string,
     data: any,
@@ -71,6 +72,10 @@ export class MockRemote extends AbstractRemote {
       start: (controller) => {
         this.streamController = controller;
         this.onStreamRequested();
+        if (this.errorOnStreamStart) {
+          controller.error(new Error('Mock error on stream start'));
+        }
+
         signal?.addEventListener('abort', () => {
           try {
             controller.close();