Merge remote-tracking branch 'origin/main' into watches

Author: stevensJourney

.changeset/dull-mugs-carry.md

@@ -0,0 +1,10 @@
+---
+'@powersync/react-native': patch
+'@powersync/diagnostics-app': patch
+'@powersync/common': patch
+'@powersync/node': patch
+'@powersync/op-sqlite': patch
+'@powersync/web': patch
+---
+
+Improve websocket keepalive logic to reduce keepalive errors.

.changeset/lazy-onions-dance.md

@@ -0,0 +1,5 @@
+---
+'@powersync/node': minor
+---
+
+Upgrade undici and use the default undici errors for WebSockets.

.changeset/light-clocks-hang.md

@@ -0,0 +1,8 @@
+---
+'@powersync/common': patch
+'@powersync/node': patch
+'@powersync/web': patch
+'@powersync/react-native': patch
+---
+
+Rust client: Properly upload CRUD entries made while offline.

demos/example-node/package.json

@@ -12,7 +12,7 @@
   "dependencies": {
     "@powersync/node": "workspace:*",
     "dotenv": "^16.4.7",
-    "undici": "^7.10.0"
+    "undici": "^7.11.0"
   },
   "devDependencies": {
     "ts-node": "^10.9.2",

packages/common/src/client/ConnectionManager.ts

@@ -194,7 +194,6 @@ export class ConnectionManager extends BaseObserver<ConnectionManagerListener> {
 
     this.logger.debug('Attempting to connect to PowerSync instance');
     await this.syncStreamImplementation?.connect(appliedOptions!);
-    this.syncStreamImplementation?.triggerCrudUpload();
   }
 
   /**

packages/common/src/client/sync/bucket/BucketStorageAdapter.ts

@@ -106,5 +106,5 @@ export interface BucketStorageAdapter extends BaseObserverInterface<BucketStorag
   /**
    * Invokes the `powersync_control` function for the sync client.
    */
-  control(op: PowerSyncControlCommand, payload: string | ArrayBuffer | null): Promise<string>;
+  control(op: PowerSyncControlCommand, payload: string | Uint8Array | null): Promise<string>;
 }

packages/common/src/client/sync/bucket/SqliteBucketStorage.ts

@@ -364,7 +364,7 @@ export class SqliteBucketStorage extends BaseObserver<BucketStorageListener> imp
     // No-op for now
   }
 
-  async control(op: PowerSyncControlCommand, payload: string | ArrayBuffer | null): Promise<string> {
+  async control(op: PowerSyncControlCommand, payload: string | Uint8Array | ArrayBuffer | null): Promise<string> {
     return await this.writeTransaction(async (tx) => {
       const [[raw]] = await tx.executeRaw('SELECT powersync_control(?, ?)', [op, payload]);
       return raw;

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

@@ -24,8 +24,14 @@ const SYNC_QUEUE_REQUEST_LOW_WATER = 5;
 
 // Keep alive message is sent every period
 const KEEP_ALIVE_MS = 20_000;
-// The ACK must be received in this period
-const KEEP_ALIVE_LIFETIME_MS = 30_000;
+
+// One message of any type must be received in this period.
+const SOCKET_TIMEOUT_MS = 30_000;
+
+// One keepalive message must be received in this period.
+// If there is a backlog of messages (for example on slow connections), keepalive messages could be delayed
+// significantly. Therefore this is longer than the socket timeout.
+const KEEP_ALIVE_LIFETIME_MS = 90_000;
 
 export const DEFAULT_REMOTE_LOGGER = Logger.get('PowerSyncRemote');
 
@@ -267,7 +273,7 @@ export abstract class AbstractRemote {
    */
   async socketStream(options: SocketSyncStreamOptions): Promise<DataStream<StreamingSyncLine>> {
     const bson = await this.getBSON();
-    return await this.socketStreamRaw(options, (data) => bson.deserialize(data), bson);
+    return await this.socketStreamRaw(options, (data) => bson.deserialize(data) as StreamingSyncLine, bson);
   }
 
   /**
@@ -279,9 +285,9 @@ export abstract class AbstractRemote {
    */
   async socketStreamRaw<T>(
     options: SocketSyncStreamOptions,
-    map: (buffer: Buffer) => T,
+    map: (buffer: Uint8Array) => T,
     bson?: typeof BSON
-  ): Promise<DataStream> {
+  ): Promise<DataStream<T>> {
     const { path, fetchStrategy = FetchStrategy.Buffered } = options;
     const mimeType = bson == null ? 'application/json' : 'application/bson';
 
@@ -304,12 +310,26 @@ export abstract class AbstractRemote {
     // automatically as a header.
     const userAgent = this.getUserAgent();
 
+    let keepAliveTimeout: any;
+    const resetTimeout = () => {
+      clearTimeout(keepAliveTimeout);
+      keepAliveTimeout = setTimeout(() => {
+        this.logger.error(`No data received on WebSocket in ${SOCKET_TIMEOUT_MS}ms, closing connection.`);
+        stream.close();
+      }, SOCKET_TIMEOUT_MS);
+    };
+    resetTimeout();
+
     const url = this.options.socketUrlTransformer(request.url);
     const connector = new RSocketConnector({
       transport: new WebsocketClientTransport({
         url,
         wsCreator: (url) => {
-          return this.createSocket(url);
+          const socket = this.createSocket(url);
+          socket.addEventListener('message', (event) => {
+            resetTimeout();
+          });
+          return socket;
         }
       }),
       setup: {
@@ -332,18 +352,23 @@ export abstract class AbstractRemote {
       rsocket = await connector.connect();
     } catch (ex) {
       this.logger.error(`Failed to connect WebSocket`, ex);
+      clearTimeout(keepAliveTimeout);
       throw ex;
     }
 
-    const stream = new DataStream({
+    resetTimeout();
+
+    const stream = new DataStream<T, Uint8Array>({
       logger: this.logger,
       pressure: {
         lowWaterMark: SYNC_QUEUE_REQUEST_LOW_WATER
-      }
+      },
+      mapLine: map
     });
 
     let socketIsClosed = false;
     const closeSocket = () => {
+      clearTimeout(keepAliveTimeout);
       if (socketIsClosed) {
         return;
       }
@@ -411,7 +436,7 @@ export abstract class AbstractRemote {
               return;
             }
 
-            stream.enqueueData(map(data));
+            stream.enqueueData(data);
           },
           onComplete: () => {
             stream.close();
@@ -537,8 +562,9 @@ export abstract class AbstractRemote {
     const decoder = new TextDecoder();
     let buffer = '';
 
-    const stream = new DataStream<T>({
-      logger: this.logger
+    const stream = new DataStream<T, string>({
+      logger: this.logger,
+      mapLine: mapLine
     });
 
     const l = stream.registerListener({
@@ -550,7 +576,7 @@ export abstract class AbstractRemote {
             if (done) {
               const remaining = buffer.trim();
               if (remaining.length != 0) {
-                stream.enqueueData(mapLine(remaining));
+                stream.enqueueData(remaining);
               }
 
               stream.close();
@@ -565,7 +591,7 @@ export abstract class AbstractRemote {
             for (var i = 0; i < lines.length - 1; i++) {
               var l = lines[i].trim();
               if (l.length > 0) {
-                stream.enqueueData(mapLine(l));
+                stream.enqueueData(l);
                 didCompleteLine = true;
               }
             }

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

@@ -858,17 +858,15 @@ 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());
 
     // Pending sync lines received from the service, as well as local events that trigger a powersync_control
     // invocation (local events include refreshed tokens and completed uploads).
     // This is a single data stream so that we can handle all control calls from a single place.
-    let controlInvocations: DataStream<{
-      command: PowerSyncControlCommand;
-      payload?: ArrayBuffer | string;
-    }> | null = null;
+    let controlInvocations: DataStream<EnqueuedCommand, Uint8Array | EnqueuedCommand> | null = null;
 
     async function connect(instr: EstablishSyncStream) {
       const syncOptions: SyncStreamOptions = {
@@ -878,20 +876,34 @@ The next upload iteration will be delayed.`);
       };
 
       if (resolvedOptions.connectionMethod == SyncStreamConnectionMethod.HTTP) {
-        controlInvocations = await remote.postStreamRaw(syncOptions, (line) => ({
-          command: PowerSyncControlCommand.PROCESS_TEXT_LINE,
-          payload: line
-        }));
+        controlInvocations = await remote.postStreamRaw(syncOptions, (line: string | EnqueuedCommand) => {
+          if (typeof line == 'string') {
+            return {
+              command: PowerSyncControlCommand.PROCESS_TEXT_LINE,
+              payload: line
+            };
+          } else {
+            // Directly enqueued by us
+            return line;
+          }
+        });
       } else {
         controlInvocations = await remote.socketStreamRaw(
           {
             ...syncOptions,
             fetchStrategy: resolvedOptions.fetchStrategy
           },
-          (buffer) => ({
-            command: PowerSyncControlCommand.PROCESS_BSON_LINE,
-            payload: buffer
-          })
+          (payload: Uint8Array | EnqueuedCommand) => {
+            if (payload instanceof Uint8Array) {
+              return {
+                command: PowerSyncControlCommand.PROCESS_BSON_LINE,
+                payload: payload
+              };
+            } else {
+              // Directly enqueued by us
+              return payload;
+            }
+          }
         );
       }
 
@@ -903,6 +915,11 @@ The next upload iteration will be delayed.`);
           }
 
           await control(line.command, line.payload);
+
+          if (!hadSyncLine) {
+            syncImplementation.triggerCrudUpload();
+            hadSyncLine = true;
+          }
         }
       } finally {
         const activeInstructions = controlInvocations;
@@ -918,7 +935,7 @@ The next upload iteration will be delayed.`);
       await control(PowerSyncControlCommand.STOP);
     }
 
-    async function control(op: PowerSyncControlCommand, payload?: ArrayBuffer | string) {
+    async function control(op: PowerSyncControlCommand, payload?: Uint8Array | string) {
       const rawResponse = await adapter.control(op, payload ?? null);
       await handleInstructions(JSON.parse(rawResponse));
     }
@@ -1157,3 +1174,8 @@ The next upload iteration will be delayed.`);
     });
   }
 }
+
+interface EnqueuedCommand {
+  command: PowerSyncControlCommand;
+  payload?: Uint8Array | string;
+}