Merge pull request #14 from powersync-ja/fix-snapshot-performance

Fix snapshot performance issues

Author: rkistner

.changeset/tender-emus-beat.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-core': patch
+---
+
+Fix performance issues and improve logging for initial snapshot replication.

.changeset/tender-emus-drum.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-jpgwire': patch
+---
+
+Fix performance issue when reading a lot of data from a socket.

packages/jpgwire/src/pgwire_node.js

@@ -86,6 +86,7 @@ class SocketAdapter {
     this._readPauseAsync = (resolve) => (this._readResume = resolve);
     this._writePauseAsync = (resolve) => (this._writeResume = resolve);
     this._error = null;
+    /** @type {net.Socket} */
     this._socket = socket;
     this._socket.on('readable', (_) => this._readResume());
     this._socket.on('end', (_) => this._readResume());
@@ -126,16 +127,17 @@ class SocketAdapter {
     for (;;) {
       if (this._error) throw this._error; // TODO callstack
       if (this._socket.readableEnded) return null;
-      buf = this._socket.read();
+      // POWERSYNC FIX: Read only as much data as available, instead of reading everything and
+      // unshifting back onto the socket
+      const toRead = Math.min(out.length, this._socket.readableLength);
+      buf = this._socket.read(toRead);
+
       if (buf?.length) break;
       if (!buf) await new Promise(this._readPauseAsync);
     }
+
     if (buf.length > out.length) {
-      out.set(buf.subarray(0, out.length));
-      this._socket.unshift(buf.subarray(out.length));
-      // POWERSYNC: Add metrics
-      recordBytesRead(out.length);
-      return out.length;
+      throw new Error('Read more data than expected');
     }
     out.set(buf);
     // POWERSYNC: Add metrics

packages/service-core/src/replication/WalStream.ts

@@ -274,7 +274,7 @@ WHERE  oid = $1::regclass`,
       params: [{ value: table.qualifiedName, type: 'varchar' }]
     });
     const row = results.rows[0];
-    if (row?.[0] ?? -1n == -1n) {
+    if ((row?.[0] ?? -1n) == -1n) {
       return '?';
     } else {
       return `~${row[0]}`;
@@ -374,8 +374,12 @@ WHERE  oid = $1::regclass`,
     micro.logger.info(`${this.slot_name} Replicating ${table.qualifiedName}`);
     const estimatedCount = await this.estimatedCount(db, table);
     let at = 0;
+    let lastLogIndex = 0;
     const cursor = await db.stream({ statement: `SELECT * FROM ${table.escapedIdentifier}` });
     let columns: { i: number; name: string }[] = [];
+    // pgwire streams rows in chunks.
+    // These chunks can be quite small (as little as 16KB), so we don't flush chunks automatically.
+
     for await (let chunk of cursor) {
       if (chunk.tag == 'RowDescription') {
         let i = 0;
@@ -392,22 +396,21 @@ WHERE  oid = $1::regclass`,
         }
         return q;
       });
-      if (at % 5000 == 0 && rows.length > 0) {
+      if (rows.length > 0 && at - lastLogIndex >= 5000) {
         micro.logger.info(`${this.slot_name} Replicating ${table.qualifiedName} ${at}/${estimatedCount}`);
+        lastLogIndex = at;
       }
       if (this.abort_signal.aborted) {
         throw new Error(`Aborted initial replication of ${this.slot_name}`);
       }
 
       for (let record of WalStream.getQueryData(rows)) {
+        // This auto-flushes when the batch reaches its size limit
         await batch.save({ tag: 'insert', sourceTable: table, before: undefined, after: record });
       }
       at += rows.length;
       Metrics.getInstance().rows_replicated_total.add(rows.length);
 
-      // pgwire streaming uses reasonable chunk sizes, so we flush at the end
-      // of each chunk.
-      await batch.flush();
       await touch();
     }
 

packages/service-core/src/storage/mongo/OperationBatch.ts

@@ -12,7 +12,7 @@ const MAX_BATCH_COUNT = 2000;
 /**
  * Maximum size of operations in the batch (estimated).
  */
-const MAX_RECORD_BATCH_SIZE = 14_000_000;
+const MAX_RECORD_BATCH_SIZE = 5_000_000;
 
 /**
  * Maximum size of size of current_data documents we lookup at a time.