Align crud uploads with Dart implementation

Author: simolus3

.changeset/warm-dots-sort.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Attempt a CRUD upload everytime `connect()` is called, even if we're unable to connect.

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

@@ -3,7 +3,6 @@ import Logger, { ILogger } from 'js-logger';
 import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
 import { AbortOperation } from '../../../utils/AbortOperation.js';
 import { BaseListener, BaseObserver, BaseObserverInterface, Disposable } from '../../../utils/BaseObserver.js';
-import { throttleLeadingTrailing } from '../../../utils/async.js';
 import { BucketStorageAdapter, PowerSyncControlCommand } from '../bucket/BucketStorageAdapter.js';
 import { CrudEntry } from '../bucket/CrudEntry.js';
 import { AbstractRemote, FetchStrategy, SyncStreamOptions } from './AbstractRemote.js';
@@ -17,7 +16,7 @@ import {
   doneResult,
   injectable,
   InjectableIterator,
-  map,
+  notifyIterator,
   SimpleAsyncIterator,
   valueResult
 } from '../../../utils/stream_transform.js';
@@ -221,16 +220,13 @@ export abstract class AbstractStreamingSyncImplementation
 {
   protected options: AbstractStreamingSyncImplementationOptions;
   protected abortController: AbortController | null;
-  // In rare cases, mostly for tests, uploads can be triggered without being properly connected.
-  // This allows ensuring that all upload processes can be aborted.
-  protected uploadAbortController: AbortController | undefined;
   protected crudUpdateListener?: () => void;
   protected streamingSyncPromise?: Promise<void>;
   protected logger: ILogger;
   private activeStreams: SubscribedStream[];
   private connectionMayHaveChanged = false;
+  private crudUploadNotifier = notifyIterator();
 
-  private isUploadingCrud: boolean = false;
   private notifyCompletedUploads?: () => void;
   private handleActiveStreamsChange?: () => void;
 
@@ -254,17 +250,7 @@ export abstract class AbstractStreamingSyncImplementation
     });
     this.abortController = null;
 
-    this.triggerCrudUpload = throttleLeadingTrailing(() => {
-      if (!this.syncStatus.connected || this.isUploadingCrud) {
-        return;
-      }
-
-      this.isUploadingCrud = true;
-      this._uploadAllCrud().finally(() => {
-        this.notifyCompletedUploads?.();
-        this.isUploadingCrud = false;
-      });
-    }, this.options.crudUploadThrottleMs!);
+    this.triggerCrudUpload = () => this.crudUploadNotifier.notify();
   }
 
   async waitForReady() {}
@@ -320,7 +306,6 @@ export abstract class AbstractStreamingSyncImplementation
     super.dispose();
     this.crudUpdateListener?.();
     this.crudUpdateListener = undefined;
-    this.uploadAbortController?.abort();
   }
 
   abstract obtainLock<T>(lockOptions: LockOptions<T>): Promise<T>;
@@ -334,7 +319,19 @@ export abstract class AbstractStreamingSyncImplementation
     return checkpoint;
   }
 
-  protected async _uploadAllCrud(): Promise<void> {
+  private async crudUploadLoop(signal: AbortSignal): Promise<void> {
+    while (!signal.aborted) {
+      await Promise.all([
+        // Start the initial CRUD upload on connect. Then, keep polling until we're done.
+        this._uploadAllCrud(signal),
+        this.delayRetry(signal, this.options.crudUploadThrottleMs!)
+      ]);
+
+      await this.crudUploadNotifier.next();
+    }
+  }
+
+  private async _uploadAllCrud(signal: AbortSignal): Promise<void> {
     return this.obtainLock({
       type: LockType.CRUD,
       callback: async () => {
@@ -343,17 +340,7 @@ export abstract class AbstractStreamingSyncImplementation
          */
         let checkedCrudItem: CrudEntry | undefined;
 
-        const controller = new AbortController();
-        this.uploadAbortController = controller;
-        this.abortController?.signal.addEventListener(
-          'abort',
-          () => {
-            controller.abort();
-          },
-          { once: true }
-        );
-
-        while (!controller.signal.aborted) {
+        while (!signal.aborted) {
           try {
             /**
              * This is the first item in the FIFO CRUD queue.
@@ -398,7 +385,7 @@ The next upload iteration will be delayed.`);
                 uploadError: ex as Error
               }
             });
-            await this.delayRetry(controller.signal);
+            await this.delayRetry(signal);
             if (!this.isConnected) {
               // Exit the upload loop if the sync stream is no longer connected
               break;
@@ -407,14 +394,15 @@ The next upload iteration will be delayed.`);
               `Caught exception when uploading. Upload will retry after a delay. Exception: ${(ex as Error).message}`
             );
           } finally {
+            this.notifyCompletedUploads?.();
+
             this.updateSyncStatus({
               dataFlow: {
                 uploading: false
               }
             });
           }
         }
-        this.uploadAbortController = undefined;
       }
     });
   }
@@ -469,15 +457,7 @@ The next upload iteration will be delayed.`);
     this.updateSyncStatus({ connected: false, connecting: false });
   }
 
-  /**
-   * @deprecated use [connect instead]
-   */
-  async streamingSync(signal?: AbortSignal, options?: PowerSyncConnectionOptions): Promise<void> {
-    if (!signal) {
-      this.abortController = new AbortController();
-      signal = this.abortController.signal;
-    }
-
+  private async streamingSync(signal: AbortSignal, options?: PowerSyncConnectionOptions): Promise<void> {
     /**
      * Listen for CRUD updates and trigger upstream uploads
      */
@@ -509,6 +489,8 @@ The next upload iteration will be delayed.`);
       });
     });
 
+    this.crudUploadLoop(signal);
+
     /**
      * This loops runs until [retry] is false or the abort signal is set to aborted.
      * Aborting the nestedAbortController will:
@@ -902,14 +884,13 @@ 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, delay = this.options.retryDelayMs): 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;
 
@@ -923,7 +904,7 @@ The next upload iteration will be delayed.`);
       };
 
       signal?.addEventListener('abort', endDelay, { once: true });
-      timeoutId = setTimeout(endDelay, retryDelayMs);
+      timeoutId = setTimeout(endDelay, delay);
     });
   }
 

packages/common/src/utils/async.ts

@@ -18,33 +18,3 @@ export function throttleTrailing(func: () => void, wait: number) {
     }
   };
 }
-
-/**
- * Throttle a function to be called at most once every "wait" milliseconds,
- * on the leading and trailing edge.
- *
- * Roughly equivalent to lodash/throttle with {leading: true, trailing: true}
- */
-export function throttleLeadingTrailing(func: () => void, wait: number) {
-  let timeoutId: ReturnType<typeof setTimeout> | null = null;
-  let lastCallTime: number = 0;
-
-  const invokeFunction = () => {
-    func();
-    lastCallTime = Date.now();
-    timeoutId = null;
-  };
-
-  return function () {
-    const now = Date.now();
-    const timeToWait = wait - (now - lastCallTime);
-
-    if (timeToWait <= 0) {
-      // Leading edge: Call the function immediately if enough time has passed
-      invokeFunction();
-    } else if (!timeoutId) {
-      // Set a timeout for the trailing edge if not already set
-      timeoutId = setTimeout(invokeFunction, timeToWait);
-    }
-  };
-}

packages/common/src/utils/stream_transform.ts

@@ -121,6 +121,50 @@ export function injectable<T>(source: SimpleAsyncIterator<T>): InjectableIterato
   };
 }
 
+export interface NotifyIterator extends SimpleAsyncIterator<undefined> {
+  notify(): void;
+}
+
+/**
+ * An iterator dispatching notifications without associated data.
+ */
+export function notifyIterator(): NotifyIterator {
+  // Possible states:
+  //   - Idle (null), no outstanding listener and no outstanding notification.
+  //   - Waiting (function), has an outstanding listener and no notification.
+  //   - Pending (true), no outstanding listener but an undelivered notification.
+  let state: null | ((complete: IteratorResult<undefined, any>) => void) | true = null;
+  const item = valueResult(undefined);
+
+  return {
+    notify() {
+      if (typeof state == 'function') {
+        // waiting -> idle
+        state(item);
+        state = null;
+      } else {
+        state = true; // idle -> pending or pending -> pending.
+      }
+    },
+    next() {
+      return new Promise((resolve) => {
+        if (typeof state == 'function') {
+          throw new Error('concurrent call to next');
+        }
+
+        if (state != null) {
+          // pending -> idle
+          resolve(item);
+          state = null;
+        } else {
+          // idle -> waiting
+          state = resolve;
+        }
+      });
+    }
+  };
+}
+
 /**
  * Splits a byte stream at line endings, emitting each line as a string.
  */

packages/common/tests/utils/stream_transform.test.ts

@@ -4,6 +4,7 @@ import {
   extractBsonObjects,
   extractJsonLines,
   injectable,
+  notifyIterator,
   SimpleAsyncIterator,
   valueResult
 } from '../../src/utils/stream_transform';
@@ -120,6 +121,48 @@ describe('injectable', () => {
   });
 });
 
+describe('notifyIterator', () => {
+  test('waits for event', async () => {
+    const iterator = notifyIterator();
+    let didReceiveEvent = false;
+    iterator.next().then((e) => {
+      expect(e.done).toStrictEqual(false);
+      didReceiveEvent = true;
+    });
+
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeFalsy();
+    iterator.notify();
+
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeTruthy();
+  });
+
+  test('merges events', async () => {
+    const iterator = notifyIterator();
+    for (let i = 0; i < 1000; i++) {
+      iterator.notify();
+    }
+
+    let didReceiveEvent = false;
+    iterator.next().then((e) => {
+      expect(e.done).toStrictEqual(false);
+      didReceiveEvent = true;
+    });
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeTruthy();
+
+    // The first iterator.next() should have consumed everything.
+    didReceiveEvent = false;
+    iterator.next().then((e) => {
+      expect(e.done).toStrictEqual(false);
+      didReceiveEvent = true;
+    });
+    await Promise.resolve();
+    expect(didReceiveEvent).toBeFalsy();
+  });
+});
+
 describe('bson objects', () => {
   test('empty stream', async () => {
     async function* source() {}

packages/node/tests/sync.test.ts

@@ -544,6 +544,28 @@ function defineSyncTests(bson: boolean) {
     expect(rows).toStrictEqual([{ name: 'from server' }]);
   });
 
+  mockSyncServiceTest('should upload on start of iteration', async ({ syncService }) => {
+    let database = await syncService.createDatabase();
+    await database.execute('INSERT INTO lists (id, name) values (uuid(), ?)', ['local write']);
+
+    syncService.installRequestInterceptor(async (request) => {
+      if (request.url.indexOf('/sync/stream') != -1) {
+        throw new Error('Pretend that the service is unavailable');
+      }
+    });
+
+    const connector = new TestConnector();
+    database.connect(connector, { ...options, retryDelayMs: 10_000, crudUploadThrottleMs: 100 });
+    await database.waitForStatus((s) => s.dataFlowStatus.downloadError != null);
+
+    // We'll never connect due to the error, but we should still try to upload once.
+    expect(connector.uploadDataInvocations).toStrictEqual(1);
+
+    // And even though we're still not connected, we should attempt uploads on crud changes.
+    await database.execute('INSERT INTO lists (id, name) values (uuid(), ?)', ['second local write']);
+    await vi.waitFor(() => expect(connector.uploadDataInvocations).toStrictEqual(2));
+  });
+
   mockSyncServiceTest('handles uploads across checkpoints', async ({ syncService }) => {
     const logger = createLogger('test', { logLevel: Logger.TRACE });
     const logMessages: string[] = [];