Catch closed errors for hold requests. Use crud throttle time for crud retries.

Author: stevensJourney

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

@@ -429,7 +429,7 @@ The next upload iteration will be delayed.`);
                 uploadError: ex
               }
             });
-            await this.delayRetry(controller.signal);
+            await this.delayRetry(controller.signal, this.options.crudUploadThrottleMs);
             if (!this.isConnected) {
               // Exit the upload loop if the sync stream is no longer connected
               break;
@@ -1216,15 +1216,14 @@ The next upload iteration will be delayed.`);
     this.iterateListeners((cb) => cb.statusUpdated?.(options));
   }
 
-  private async delayRetry(signal?: AbortSignal): Promise<void> {
+  private async delayRetry(signal?: AbortSignal, delayMs?: number): Promise<void> {
     return new Promise((resolve) => {
       if (signal?.aborted) {
         // If the signal is already aborted, resolve immediately
         resolve();
         return;
       }
-
-      const { retryDelayMs } = this.options;
+      const delay = delayMs ?? this.options.retryDelayMs;
 
       let timeoutId: ReturnType<typeof setTimeout> | undefined;
 
@@ -1238,7 +1237,7 @@ The next upload iteration will be delayed.`);
       };
 
       signal?.addEventListener('abort', endDelay, { once: true });
-      timeoutId = setTimeout(endDelay, retryDelayMs);
+      timeoutId = setTimeout(endDelay, delay);
     });
   }
 

packages/web/src/db/adapters/LockedAsyncDatabaseAdapter.ts

@@ -251,12 +251,16 @@ export class LockedAsyncDatabaseAdapter
         if (timeoutId) {
           clearTimeout(timeoutId);
         }
-        const holdId = this.requiresHolds ? await this.baseDB.markHold() : null;
+        let holdId: string | null = null;
         try {
           if (this.requiresReOpen) {
+            this.logger.debug('Re-opening database');
             await this.openInternalDB();
+            this.logger.debug('Database re-opened');
             this.requiresReOpen = false;
           }
+
+          holdId = this.requiresHolds ? await this.baseDB.markHold() : null;
           return await callback();
         } catch (ex) {
           if (ex instanceof WorkerConnectionClosedError) {

packages/web/tests/multiple_instances.test.ts

@@ -284,11 +284,12 @@ describe('Multiple Instances', { sequential: true }, () => {
     await stream2.dispose();
   });
 
-  it('should trigger uploads from last connected clients', async () => {
+  it('should trigger uploads from last connected clients', { timeout: Infinity }, async () => {
     // Generate the first streaming sync implementation
     const connector1 = new TestConnector();
     const spy1 = vi.spyOn(connector1, 'uploadData');
 
+    await new Promise((resolve) => setTimeout(resolve, 5000));
     const db = openDatabase();
     await db.init();
     // They need to use the same identifier to use the same shared worker.
@@ -380,8 +381,6 @@ describe('Multiple Instances', { sequential: true }, () => {
     // Close the second client, leaving only the first one
     await stream2.dispose();
 
-    // The dispose above will disconnect, but we need to wait for the sync stream to be created before we can update the status
-    await vi.waitFor(() => expect(stream1.syncStatus.connecting).true);
     // Hack, set the status to connected in order to trigger the upload
     await (stream1 as any)['_testUpdateStatus'](new SyncStatus({ connected: true }));
     stream1.triggerCrudUpload();