Changed binlog backpressure mechanism to be based on processing queue memory usage rather than number of events

Author: Rentacookie

modules/module-mysql/package.json

@@ -33,7 +33,7 @@
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
     "@powersync/service-jsonbig": "workspace:*",
-    "@powersync/mysql-zongji": "0.0.0-dev-20250527085137",
+    "@powersync/mysql-zongji": "0.0.0-dev-20250528105319",
     "async": "^3.2.4",
     "mysql2": "^3.11.0",
     "semver": "^7.5.4",

modules/module-mysql/src/replication/zongji/BinLogListener.ts

@@ -34,6 +34,10 @@ export class BinLogListener {
 
   zongji: ZongJi;
   processingQueue: async.QueueObject<BinLogEvent>;
+  /**
+   *  The combined size in bytes of all the binlog events currently in the processing queue.
+   */
+  queueMemoryUsage: number;
 
   constructor(public options: BinLogListenerOptions) {
     this.connectionManager = options.connectionManager;
@@ -42,9 +46,18 @@ export class BinLogListener {
     this.currentGTID = null;
 
     this.processingQueue = async.queue(this.createQueueWorker(), 1);
+    this.queueMemoryUsage = 0;
     this.zongji = this.createZongjiListener();
   }
 
+  /**
+   *  The queue memory limit in bytes as defined in the connection options.
+   *  @private
+   */
+  private get queueMemoryLimit(): number {
+    return this.connectionManager.options.binlog_queue_memory_limit * 1024 * 1024;
+  }
+
   public async start(): Promise<void> {
     if (this.isStopped) {
       return;
@@ -62,6 +75,12 @@ export class BinLogListener {
     } satisfies StartOptions);
 
     return new Promise<void>((resolve, reject) => {
+      // Handle an edge case where the listener has already been stopped before completing startup
+      if (this.isStopped) {
+        logger.info('BinLog listener was stopped before startup completed.');
+        resolve();
+      }
+
       this.zongji.on('error', (error) => {
         if (!this.isStopped) {
           logger.error('Binlog listener error:', error);
@@ -86,12 +105,6 @@ export class BinLogListener {
         resolve();
         logger.info('BinLog listener stopped. Replication ended.');
       });
-
-      // Handle the edge case where the listener has already been stopped before completing startup
-      if (this.isStopped) {
-        logger.info('BinLog listener was stopped before startup completed.');
-        resolve();
-      }
     });
   }
 
@@ -112,11 +125,12 @@ export class BinLogListener {
     zongji.on('binlog', async (evt) => {
       logger.info(`Received Binlog event:${evt.getEventName()}`);
       this.processingQueue.push(evt);
+      this.queueMemoryUsage += evt.size;
 
       // When the processing queue grows past the threshold, we pause the binlog listener
-      if (this.processingQueue.length() > this.connectionManager.options.max_binlog_queue_size) {
+      if (this.isQueueOverCapacity()) {
         logger.info(
-          `Max Binlog processing queue length [${this.connectionManager.options.max_binlog_queue_size}] reached. Pausing Binlog listener.`
+          `Binlog processing queue has reached its memory limit of [${this.connectionManager.options.binlog_queue_memory_limit}MB]. Pausing Binlog listener.`
         );
         zongji.pause();
         await this.processingQueue.empty();
@@ -201,6 +215,12 @@ export class BinLogListener {
           await this.eventHandler.onCommit(LSN);
           break;
       }
+
+      this.queueMemoryUsage -= evt.size;
     };
   }
+
+  isQueueOverCapacity(): boolean {
+    return this.queueMemoryUsage >= this.queueMemoryLimit;
+  }
 }

modules/module-mysql/src/types/types.ts

@@ -24,7 +24,7 @@ export interface NormalizedMySQLConnectionConfig {
 
   lookup?: LookupFunction;
 
-  max_binlog_queue_size: number;
+  binlog_queue_memory_limit: number;
 }
 
 export const MySQLConnectionConfig = service_types.configFile.DataSourceConfig.and(
@@ -43,8 +43,8 @@ export const MySQLConnectionConfig = service_types.configFile.DataSourceConfig.a
     client_private_key: t.string.optional(),
 
     reject_ip_ranges: t.array(t.string).optional(),
-    // The maximum number of binlog events that can be queued in memory before throttling is applied.
-    max_binlog_queue_size: t.number.optional()
+    // The combined size of binlog events that can be queued in memory before throttling is applied.
+    binlog_queue_memory_limit: t.number.optional()
   })
 );
 
@@ -118,8 +118,8 @@ export function normalizeConnectionConfig(options: MySQLConnectionConfig): Norma
 
     server_id: options.server_id ?? 1,
 
-    // Based on profiling, a queue size of 1000 uses about 50MB of memory.
-    max_binlog_queue_size: options.max_binlog_queue_size ?? 1000,
+    // Binlog processing queue memory limit before throttling is applied.
+    binlog_queue_memory_limit: options.binlog_queue_memory_limit ?? 50,
 
     lookup
   };

modules/module-mysql/test/src/BinLogListener.test.ts

@@ -6,12 +6,13 @@ import { v4 as uuid } from 'uuid';
 import * as common from '@module/common/common-index.js';
 import { createRandomServerId } from '@module/utils/mysql-utils.js';
 import { TableMapEntry } from '@powersync/mysql-zongji';
+import crypto from 'crypto';
 
 describe('BinlogListener tests', () => {
-  const MAX_QUEUE_SIZE = 10;
+  const MAX_QUEUE_CAPACITY_MB = 1;
   const BINLOG_LISTENER_CONNECTION_OPTIONS = {
     ...TEST_CONNECTION_OPTIONS,
-    max_binlog_queue_size: MAX_QUEUE_SIZE
+    binlog_queue_memory_limit: MAX_QUEUE_CAPACITY_MB
   };
 
   let connectionManager: MySQLConnectionManager;
@@ -22,7 +23,7 @@ describe('BinlogListener tests', () => {
     connectionManager = new MySQLConnectionManager(BINLOG_LISTENER_CONNECTION_OPTIONS, {});
     const connection = await connectionManager.getConnection();
     await clearTestDb(connection);
-    await connection.query(`CREATE TABLE test_DATA (id CHAR(36) PRIMARY KEY, description text)`);
+    await connection.query(`CREATE TABLE test_DATA (id CHAR(36) PRIMARY KEY, description MEDIUMTEXT)`);
     connection.release();
     const fromGTID = await getFromGTID(connectionManager);
 
@@ -52,24 +53,30 @@ describe('BinlogListener tests', () => {
     expect(queueStopSpy).toHaveBeenCalled();
   });
 
-  test('Pause Zongji binlog listener when processing queue reaches max size', async () => {
+  test('Pause Zongji binlog listener when processing queue reaches maximum memory size', async () => {
     const pauseSpy = vi.spyOn(binLogListener.zongji, 'pause');
     const resumeSpy = vi.spyOn(binLogListener.zongji, 'resume');
-    const queueSpy = vi.spyOn(binLogListener.processingQueue, 'length');
 
-    const ROW_COUNT = 100;
+    // Pause the event handler to force a backlog on the processing queue
+    eventHandler.pause();
+
+    const ROW_COUNT = 10;
     await insertRows(connectionManager, ROW_COUNT);
 
     const startPromise = binLogListener.start();
 
+    // Wait for listener to pause due to queue reaching capacity
+    await vi.waitFor(() => expect(pauseSpy).toHaveBeenCalled(), { timeout: 5000 });
+
+    expect(binLogListener.isQueueOverCapacity()).toBeTruthy();
+    // Resume event processing
+    eventHandler.unpause!();
+
     await vi.waitFor(() => expect(eventHandler.rowsWritten).equals(ROW_COUNT), { timeout: 5000 });
     binLogListener.stop();
     await expect(startPromise).resolves.toBeUndefined();
-
-    // Count how many times the queue reached the max size. Consequently, we expect the listener to have paused and resumed that many times.
-    const overThresholdCount = queueSpy.mock.results.map((r) => r.value).filter((v) => v === MAX_QUEUE_SIZE).length;
-    expect(pauseSpy).toHaveBeenCalledTimes(overThresholdCount);
-    expect(resumeSpy).toHaveBeenCalledTimes(overThresholdCount);
+    // Confirm resume was called after unpausing
+    expect(resumeSpy).toHaveBeenCalled();
   });
 
   test('Binlog events are correctly forwarded to provided binlog events handler', async () => {
@@ -101,7 +108,9 @@ async function getFromGTID(connectionManager: MySQLConnectionManager) {
 
 async function insertRows(connectionManager: MySQLConnectionManager, count: number) {
   for (let i = 0; i < count; i++) {
-    await connectionManager.query(`INSERT INTO test_DATA(id, description) VALUES('${uuid()}','test${i}')`);
+    await connectionManager.query(
+      `INSERT INTO test_DATA(id, description) VALUES('${uuid()}','test${i} ${crypto.randomBytes(100_000).toString('hex')}')`
+    );
   }
 }
 
@@ -119,7 +128,19 @@ class TestBinLogEventHandler implements BinLogEventHandler {
   rowsDeleted = 0;
   commitCount = 0;
 
+  unpause: ((value: void | PromiseLike<void>) => void) | undefined;
+  private pausedPromise: Promise<void> | undefined;
+
+  pause() {
+    this.pausedPromise = new Promise((resolve) => {
+      this.unpause = resolve;
+    });
+  }
+
   async onWrite(rows: Row[], tableMap: TableMapEntry) {
+    if (this.pausedPromise) {
+      await this.pausedPromise;
+    }
     this.rowsWritten = this.rowsWritten + rows.length;
   }