[MySQL] Fix replication stalls caused by silently dropped idle connections

.changeset/mysql-tcp-keepalive.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-module-mysql': patch
+---
+
+Enable TCP keepalive on the Zongji binlog and control connections and on the connection pool, so that idle connections are no longer silently dropped by stateful firewalls, which froze replication for ~950 seconds per occurrence with no error or health signal. Also bound the BinLog listener stop sequence, so that a dead control connection cannot stall shutdown for that same window.

modules/module-mysql/src/replication/MySQLConnectionManager.ts

@@ -1,5 +1,5 @@
 import { BaseObserver, logger } from '@powersync/lib-services-framework';
-import { ZongJi } from '@powersync/mysql-zongji';
+import { ZongJi, ZongjiOptions } from '@powersync/mysql-zongji';
 import mysql, { FieldPacket, RowDataPacket } from 'mysql2';
 import mysqlPromise from 'mysql2/promise';
 import { NormalizedMySQLConnectionConfig } from '../types/types.js';
@@ -49,15 +49,23 @@ export class MySQLConnectionManager extends BaseObserver<MySQLConnectionManagerL
    * Create a new replication listener
    */
   createBinlogListener(): ZongJi {
-    const listener = new ZongJi({
+    // The keepalive options are not part of the published ZongjiOptions type yet, but are passed
+    // through to @vlasky/mysql for both the binlog and the control connection.
+    const options: ZongjiOptions & { enableKeepAlive: boolean; keepAliveInitialDelay: number } = {
       host: this.options.hostname,
       port: this.options.port,
       user: this.options.username,
       password: this.options.password,
+      // TCP keepalive is disabled by default in @vlasky/mysql. Without it, the idle control
+      // connection can be silently dropped by stateful firewalls, freezing replication on the
+      // next table metadata query until the TCP retransmission timeout (~950s).
+      enableKeepAlive: true,
+      keepAliveInitialDelay: mysql_utils.TCP_KEEPALIVE_INITIAL_DELAY,
       // We want to avoid parsing date/time values to Date, because that drops sub-millisecond precision.
       dateStrings: true,
       timeZone: 'Z'
-    });
+    };
+    const listener = new ZongJi(options);
 
     this.binlogListeners.push(listener);
 

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

@@ -1,5 +1,12 @@
 import { Logger, ReplicationAssertionError, logger as defaultLogger } from '@powersync/lib-services-framework';
-import { BinLogEvent, BinLogQueryEvent, StartOptions, TableMapEntry, ZongJi } from '@powersync/mysql-zongji';
+import {
+  BinLogEvent,
+  BinLogQueryEvent,
+  MySQLConnection,
+  StartOptions,
+  TableMapEntry,
+  ZongJi
+} from '@powersync/mysql-zongji';
 import { TablePattern } from '@powersync/service-sync-rules';
 import async from 'async';
 import pkg, {
@@ -33,6 +40,16 @@ const { Parser } = pkg;
  *  Seconds of inactivity after which a keepalive event is sent by the MySQL server.
  */
 export const KEEPALIVE_INACTIVITY_THRESHOLD = 30;
+
+/**
+ *  Maximum time in milliseconds to wait for Zongji to stop before force-closing its control connection.
+ */
+export const ZONGJI_STOP_TIMEOUT = 5_000;
+
+// The Zongji type definitions do not expose the control connection it uses for table metadata
+// queries and the KILL query issued during stop.
+type ZongJiWithControlConnection = ZongJi & { ctrlConnection: MySQLConnection };
+
 export type Row = Record<string, any>;
 
 /**
@@ -222,12 +239,24 @@ export class BinLogListener {
   private async stopZongji(): Promise<void> {
     if (!this.zongji.stopped) {
       this.logger.info('Stopping BinLog Listener...');
-      await new Promise<void>((resolve) => {
+      let stopped = false;
+      const stopPromise = new Promise<void>((resolve) => {
         this.zongji.once('stopped', () => {
+          stopped = true;
           resolve();
         });
         this.zongji.stop();
       });
+      // Zongji only emits 'stopped' once the KILL query on its control connection has completed.
+      // If that connection has been dead for a while, the query can block on TCP retransmissions
+      // for many minutes, so we destroy the socket after a timeout to unblock the stop.
+      const timeout = timers.setTimeout(ZONGJI_STOP_TIMEOUT, undefined, { ref: false }).then(() => {
+        if (!stopped) {
+          this.logger.warn('Timed out waiting for the BinLog Listener to stop. Closing the control connection.');
+          (this.zongji as ZongJiWithControlConnection).ctrlConnection._socket?.destroy();
+        }
+      });
+      await Promise.race([stopPromise, timeout]);
       this.logger.info('BinLog Listener stopped.');
     }
   }

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

@@ -12,6 +12,13 @@ export type RetriedQueryOptions = {
   retries?: number;
 };
 
+/**
+ *  TCP keepalive initial delay in milliseconds for connections to the MySQL server.
+ *  Keepalive prevents long-lived idle connections from being silently dropped by stateful
+ *  firewalls, which commonly time out idle flows after an hour.
+ */
+export const TCP_KEEPALIVE_INITIAL_DELAY = 40_000;
+
 /**
  * Retry a simple query - up to 2 attempts total.
  */
@@ -54,6 +61,10 @@ export function createPool(config: types.NormalizedMySQLConnectionConfig, option
     timezone: 'Z', // Ensure no auto timezone manipulation of the dates occur
     jsonStrings: true, // Return JSON columns as strings
     dateStrings: true, // We parse and format them ourselves
+    // mysql2 enables TCP keepalive by default, but without an initial delay the OS default of
+    // 7200 seconds applies, which is too late for common 3600 second firewall idle timeouts.
+    enableKeepAlive: true,
+    keepAliveInitialDelay: TCP_KEEPALIVE_INITIAL_DELAY,
     // Apply URL connection parameters (explicit options override these via spread below)
     ...(params.connectTimeout != null ? { connectTimeout: params.connectTimeout } : {}),
     ...(params.connectionLimit != null ? { connectionLimit: params.connectionLimit } : {}),

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

@@ -1,6 +1,12 @@
 import { MySQLConnectionManager } from '@module/replication/MySQLConnectionManager.js';
 import { BinLogListener, SchemaChange, SchemaChangeType } from '@module/replication/zongji/BinLogListener.js';
-import { getMySQLVersion, qualifiedMySQLTable, satisfiesVersion } from '@module/utils/mysql-utils.js';
+import {
+  getMySQLVersion,
+  qualifiedMySQLTable,
+  satisfiesVersion,
+  TCP_KEEPALIVE_INITIAL_DELAY
+} from '@module/utils/mysql-utils.js';
+import { MySQLConnection } from '@powersync/mysql-zongji';
 import { TablePattern } from '@powersync/service-sync-rules';
 import crypto from 'crypto';
 import { v4 as uuid } from 'uuid';
@@ -13,6 +19,11 @@ import {
   TestBinLogEventHandler
 } from './util.js';
 
+// The zongji type definitions do not expose the connection config or the control connection.
+type ConnectionWithConfig = MySQLConnection & {
+  config: { enableKeepAlive?: boolean; keepAliveInitialDelay?: number };
+};
+
 describe('BinlogListener tests', { timeout: 60_000 }, () => {
   const MAX_QUEUE_CAPACITY_MB = 1;
   const BINLOG_LISTENER_CONNECTION_OPTIONS = {
@@ -61,6 +72,34 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => {
     expect(queueStopSpy).toHaveBeenCalled();
   });
 
+  test('TCP keepalive is enabled on the binlog and control connections', async () => {
+    // Without keepalive, the control connection can idle for hours and be silently dropped by
+    // stateful firewalls. The next metadata query then blocks until the kernel gives up on TCP
+    // retransmissions, freezing the whole binlog pipeline for ~15 minutes.
+    const { connection, ctrlConnection } = binLogListener.zongji as unknown as {
+      connection: ConnectionWithConfig;
+      ctrlConnection: ConnectionWithConfig;
+    };
+
+    for (const conn of [connection, ctrlConnection]) {
+      expect(conn.config.enableKeepAlive).toBe(true);
+      expect(conn.config.keepAliveInitialDelay).toBe(TCP_KEEPALIVE_INITIAL_DELAY);
+    }
+  });
+
+  test('Stop completes when the control connection is unresponsive', { timeout: 20_000 }, async () => {
+    await binLogListener.start();
+
+    // Simulate a control connection that was silently dropped by the network: the KILL query
+    // issued by zongji.stop() never gets a response.
+    const { ctrlConnection } = binLogListener.zongji as unknown as { ctrlConnection: MySQLConnection };
+    vi.spyOn(ctrlConnection, 'query').mockImplementation(() => {});
+
+    await binLogListener.stop();
+
+    expect(binLogListener.zongji.stopped).toBeTruthy();
+  });
+
   test('Zongji listener is stopped when processing queue reaches maximum memory size', async () => {
     const stopSpy = vi.spyOn(binLogListener.zongji, 'stop');
 

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

@@ -1,4 +1,5 @@
-import { isVersionAtLeast } from '@module/utils/mysql-utils.js';
+import * as types from '@module/types/types.js';
+import { createPool, isVersionAtLeast, TCP_KEEPALIVE_INITIAL_DELAY } from '@module/utils/mysql-utils.js';
 import { describe, expect, test } from 'vitest';
 
 describe('MySQL Utility Tests', () => {
@@ -14,4 +15,21 @@ describe('MySQL Utility Tests', () => {
     expect(isVersionAtLeast(olderVersion, '8.0')).toBeFalsy();
     expect(isVersionAtLeast(improperSemver, '5.7')).toBeTruthy();
   });
+
+  test('Pool connections are configured with a TCP keepalive initial delay', async () => {
+    // mysql2 enables keepalive by default, but without an initial delay the OS default of
+    // 7200 seconds applies, which is too late for common 3600 second firewall idle timeouts.
+    const config = types.normalizeConnectionConfig({
+      type: 'mysql',
+      uri: 'mysql://root:password@localhost:3306/mydatabase'
+    });
+    // The pool is lazy, so no connection is made here.
+    const pool = createPool(config);
+    const { connectionConfig } = (pool as unknown as { config: { connectionConfig: Record<string, unknown> } }).config;
+
+    expect(connectionConfig.enableKeepAlive).toBe(true);
+    expect(connectionConfig.keepAliveInitialDelay).toBe(TCP_KEEPALIVE_INITIAL_DELAY);
+
+    await pool.promise().end();
+  });
 });