Rust: Trigger CRUD upload on connect

Author: simolus3

.changeset/light-clocks-hang.md

@@ -0,0 +1,5 @@
+---
+'@powersync/common': patch
+---
+
+Rust client: Properly upload CRUD entries made while offline.

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

@@ -840,6 +840,7 @@ The next upload iteration will be delayed.`);
     const adapter = this.options.adapter;
     const remote = this.options.remote;
     let receivingLines: Promise<void> | null = null;
+    let hadSyncLine = false;
 
     const abortController = new AbortController();
     signal.addEventListener('abort', () => abortController.abort());
@@ -884,6 +885,11 @@ The next upload iteration will be delayed.`);
             return;
           }
 
+          if (!hadSyncLine) {
+            syncImplementation.triggerCrudUpload();
+            hadSyncLine = true;
+          }
+
           await control(line.command, line.payload);
         }
       } finally {

packages/common/src/utils/DataStream.ts

@@ -39,6 +39,7 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
   protected isClosed: boolean;
 
   protected processingPromise: Promise<void> | null;
+  protected notifyDataAdded: ((_: null) => void) | null;
 
   protected logger: ILogger;
 
@@ -90,6 +91,7 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
     }
 
     this.dataQueue.push(data);
+    this.notifyDataAdded?.(null);
 
     this.processQueue();
   }
@@ -151,7 +153,11 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
       await this.iterateAsyncErrored(async (l) => l.highWater?.());
     }
 
-    return (this.processingPromise = this._processQueue());
+    const promise = (this.processingPromise = this._processQueue());
+    promise.finally(() => {
+      return (this.processingPromise = null);
+    });
+    return promise;
   }
 
   /**
@@ -178,7 +184,7 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
 
   protected async _processQueue() {
     if (this.isClosed || !this.hasDataReader()) {
-      Promise.resolve().then(() => (this.processingPromise = null));
+      await Promise.resolve();
       return;
     }
 
@@ -188,10 +194,13 @@ export class DataStream<Data extends any = any> extends BaseObserver<DataStreamL
     }
 
     if (this.dataQueue.length <= this.lowWatermark) {
-      await this.iterateAsyncErrored(async (l) => l.lowWater?.());
-    }
+      const dataAdded = new Promise((resolve) => {
+        this.notifyDataAdded = resolve;
+      });
 
-    this.processingPromise = null;
+      await Promise.race([this.iterateAsyncErrored(async (l) => l.lowWater?.()), dataAdded]);
+      this.notifyDataAdded = null;
+    }
 
     if (this.dataQueue.length) {
       // Next tick

packages/node/tests/sync.test.ts

@@ -424,7 +424,7 @@ function defineSyncTests(impl: SyncClientImplementation) {
 
     mockSyncServiceTest('interrupt and defrag', async ({ syncService }) => {
       let database = await syncService.createDatabase();
-      database.connect(new TestConnector(), { connectionMethod: SyncStreamConnectionMethod.HTTP });
+      database.connect(new TestConnector(), options);
       await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
 
       syncService.pushLine({
@@ -442,7 +442,7 @@ function defineSyncTests(impl: SyncClientImplementation) {
       await database.close();
       await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(0));
       database = await syncService.createDatabase();
-      database.connect(new TestConnector(), { connectionMethod: SyncStreamConnectionMethod.HTTP });
+      database.connect(new TestConnector(), options);
       await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
 
       // A sync rule deploy could reset buckets, making the new bucket smaller than the existing one.
@@ -459,6 +459,39 @@ function defineSyncTests(impl: SyncClientImplementation) {
       await waitForSyncStatus(database, (s) => s.downloadProgress == null);
     });
   });
+
+  mockSyncServiceTest('should upload after connecting', async ({ syncService }) => {
+    let database = await syncService.createDatabase();
+
+    database.execute('INSERT INTO lists (id, name) values (uuid(), ?)', ['local write']);
+    const query = database.watchWithAsyncGenerator('SELECT name FROM lists')[Symbol.asyncIterator]();
+    let rows = (await query.next()).value.rows._array;
+    expect(rows).toStrictEqual([{ name: 'local write' }]);
+
+    database.connect(new TestConnector(), options);
+    await vi.waitFor(() => expect(syncService.connectedListeners).toHaveLength(1));
+
+    syncService.pushLine({ checkpoint: { last_op_id: '1', write_checkpoint: '1', buckets: [bucket('a', 1)] } });
+    syncService.pushLine({
+      data: {
+        bucket: 'a',
+        data: [
+          {
+            checksum: 0,
+            op_id: '1',
+            op: 'PUT',
+            object_id: '1',
+            object_type: 'lists',
+            data: '{"name": "from server"}'
+          }
+        ]
+      }
+    });
+    syncService.pushLine({ checkpoint_complete: { last_op_id: '1' } });
+
+    rows = (await query.next()).value.rows._array;
+    expect(rows).toStrictEqual([{ name: 'from server' }]);
+  });
 }
 
 function bucket(name: string, count: number, options: { priority: number } = { priority: 3 }): BucketChecksum {

packages/node/tests/utils.ts

@@ -116,6 +116,13 @@ export const mockSyncServiceTest = tempDirectoryTest.extend<{
         });
 
         return new Response(syncLines.pipeThrough(asLines) as any, { status: 200 });
+      } else if (request.url.indexOf('/write-checkpoint2.json') != -1) {
+        return new Response(
+          JSON.stringify({
+            data: { write_checkpoint: '1' }
+          }),
+          { status: 200 }
+        );
       } else {
         return new Response('Not found', { status: 404 });
       }