[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 add a periodic liveness probe on the control connection that restarts replication within about a minute when the connection stops responding, and bound the BinLog listener stop sequence so that a dead control connection cannot stall shutdown.

modules/module-mysql/package.json

@@ -19,11 +19,12 @@
   },
   "dependencies": {
     "@powersync/lib-services-framework": "workspace:*",
+    "@powersync/mysql-zongji": "^0.6.0",
     "@powersync/service-core": "workspace:*",
+    "@powersync/service-jsonbig": "workspace:*",
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
-    "@powersync/service-jsonbig": "workspace:*",
-    "@powersync/mysql-zongji": "^0.6.0",
+    "@vlasky/mysql": "^2.18.6",
     "async": "^3.2.4",
     "mysql2": "^3.11.0",
     "node-sql-parser": "^5.3.9",

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

@@ -1,5 +1,6 @@
 import { BaseObserver, logger } from '@powersync/lib-services-framework';
-import { ZongJi } from '@powersync/mysql-zongji';
+import { ZongJi, ZongjiOptions } from '@powersync/mysql-zongji';
+import { createConnection, VlaskyConnection } from '@vlasky/mysql';
 import mysql, { FieldPacket, RowDataPacket } from 'mysql2';
 import mysqlPromise from 'mysql2/promise';
 import { NormalizedMySQLConnectionConfig } from '../types/types.js';
@@ -9,6 +10,16 @@ export interface MySQLConnectionManagerListener {
   onEnded(): void;
 }
 
+export interface BinlogListenerConnections {
+  zongji: ZongJi;
+  /**
+   *  The connection Zongji uses for table metadata queries and the KILL query issued during stop.
+   *  Created by us so that we keep a handle on it: Zongji does not destroy connections it did not
+   *  create, so the owner of the BinLogListener is responsible for destroying it.
+   */
+  controlConnection: VlaskyConnection;
+}
+
 export class MySQLConnectionManager extends BaseObserver<MySQLConnectionManagerListener> {
   /**
    *  Pool that can create streamable connections
@@ -20,6 +31,7 @@ export class MySQLConnectionManager extends BaseObserver<MySQLConnectionManagerL
   private readonly promisePool: mysqlPromise.Pool;
 
   private binlogListeners: ZongJi[] = [];
+  private controlConnections: VlaskyConnection[] = [];
 
   private isClosed = false;
 
@@ -46,22 +58,33 @@ export class MySQLConnectionManager extends BaseObserver<MySQLConnectionManagerL
   }
 
   /**
-   * Create a new replication listener
+   * Create a new replication listener, along with the control connection it uses.
    */
-  createBinlogListener(): ZongJi {
-    const listener = new ZongJi({
+  createBinlogListener(): BinlogListenerConnections {
+    // We create the control connection ourselves and pass it to Zongji, so that we keep a handle
+    // on it for liveness probes and cleanup. Zongji creates its binlog connection from a copy of
+    // this connection's config, so the options here apply to both connections.
+    const controlConnection = createConnection({
       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'
     });
+    // The published ZongjiOptions type does not cover passing in an existing connection yet.
+    const listener = new ZongJi(controlConnection as unknown as ZongjiOptions);
 
     this.binlogListeners.push(listener);
+    this.controlConnections.push(controlConnection);
 
-    return listener;
+    return { zongji: listener, controlConnection };
   }
 
   /**
@@ -114,6 +137,11 @@ export class MySQLConnectionManager extends BaseObserver<MySQLConnectionManagerL
       listener.stop();
     }
 
+    // Zongji does not destroy connections it did not create.
+    for (const connection of this.controlConnections) {
+      connection.destroy();
+    }
+
     try {
       await this.promisePool.end();
     } catch (error) {

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

@@ -1,6 +1,7 @@
 import { Logger, ReplicationAssertionError, logger as defaultLogger } from '@powersync/lib-services-framework';
 import { BinLogEvent, BinLogQueryEvent, StartOptions, TableMapEntry, ZongJi } from '@powersync/mysql-zongji';
 import { TablePattern } from '@powersync/service-sync-rules';
+import { VlaskyConnection } from '@vlasky/mysql';
 import async from 'async';
 import pkg, {
   AST,
@@ -24,7 +25,7 @@ import {
   isTruncate,
   matchedSchemaChangeQuery
 } from '../../utils/parser-utils.js';
-import { MySQLConnectionManager } from '../MySQLConnectionManager.js';
+import { BinlogListenerConnections, MySQLConnectionManager } from '../MySQLConnectionManager.js';
 import * as zongji_utils from './zongji-utils.js';
 
 const { Parser } = pkg;
@@ -33,6 +34,22 @@ 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;
+
+/**
+ *  Interval in milliseconds between liveness probes on the Zongji control connection.
+ */
+export const CTRL_CONNECTION_KEEPALIVE_INTERVAL = 60_000;
+
+/**
+ *  Maximum time in milliseconds to wait for a control connection liveness probe to respond.
+ */
+export const CTRL_CONNECTION_KEEPALIVE_TIMEOUT = 5_000;
+
 export type Row = Record<string, any>;
 
 /**
@@ -87,6 +104,8 @@ export interface BinLogListenerOptions {
   startGTID: common.ReplicatedGTID;
   logger?: Logger;
   keepAliveInactivitySeconds?: number;
+  ctrlConnectionKeepAliveIntervalMs?: number;
+  ctrlConnectionKeepAliveTimeoutMs?: number;
 }
 
 /**
@@ -108,6 +127,12 @@ export class BinLogListener {
   private isTransactionOpen = false;
 
   zongji: ZongJi;
+  /**
+   *  The connection Zongji uses for table metadata queries and its shutdown KILL query. We create
+   *  it ourselves so that we keep a handle on it for liveness probes and cleanup: Zongji does not
+   *  destroy connections it did not create.
+   */
+  controlConnection: VlaskyConnection;
   processingQueue: async.QueueObject<BinLogEvent>;
 
   /**
@@ -122,7 +147,9 @@ export class BinLogListener {
     this.currentGTID = options.startGTID;
     this.sqlParser = new Parser();
     this.processingQueue = this.createProcessingQueue();
-    this.zongji = this.createZongjiListener();
+    const { zongji, controlConnection } = this.createZongjiListener();
+    this.zongji = zongji;
+    this.controlConnection = controlConnection;
     this.listenerError = null;
     this.databaseFilter = this.createDatabaseFilter(options.sourceTables);
   }
@@ -214,20 +241,36 @@ export class BinLogListener {
 
   private async restartZongji(): Promise<void> {
     if (this.zongji.stopped) {
-      this.zongji = this.createZongjiListener();
+      const { zongji, controlConnection } = this.createZongjiListener();
+      this.zongji = zongji;
+      this.controlConnection = controlConnection;
       await this.start(true);
     }
   }
 
   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.controlConnection._socket?.destroy();
+        }
+      });
+      await Promise.race([stopPromise, timeout]);
+      // Zongji does not destroy connections it did not create.
+      this.controlConnection.destroy();
       this.logger.info('BinLog Listener stopped.');
     }
   }
@@ -243,16 +286,61 @@ export class BinLogListener {
   }
 
   public async replicateUntilStopped(): Promise<void> {
+    const keepAlive = this.keepAliveControlConnectionUntilStopped();
     while (!this.isStopped) {
       await timers.setTimeout(1_000);
     }
+    await keepAlive;
 
     if (this.listenerError) {
       this.logger.error('BinLog Listener stopped due to an error:', this.listenerError);
       throw this.listenerError;
     }
   }
 
+  /**
+   *  The binlog connection is kept alive by the MySQL server heartbeat, but the control connection
+   *  carries no traffic between metadata queries. TCP keepalive stops it from being dropped when idle,
+   *  but detects an already dead connection slowly, so we additionally probe it with a lightweight
+   *  query and restart replication if it does not respond in time.
+   */
+  private async keepAliveControlConnectionUntilStopped(): Promise<void> {
+    const interval = this.options.ctrlConnectionKeepAliveIntervalMs ?? CTRL_CONNECTION_KEEPALIVE_INTERVAL;
+    let idleTime = 0;
+    while (!this.isStopped) {
+      await timers.setTimeout(1_000);
+      idleTime += 1_000;
+      if (idleTime >= interval && !(this.isStopped || this.isStopping)) {
+        idleTime = 0;
+        await this.probeControlConnection();
+      }
+    }
+  }
+
+  private async probeControlConnection(): Promise<void> {
+    const controlConnection = this.controlConnection;
+    if (this.zongji.stopped) {
+      return;
+    }
+    const timeout = this.options.ctrlConnectionKeepAliveTimeoutMs ?? CTRL_CONNECTION_KEEPALIVE_TIMEOUT;
+    const responded = await new Promise<boolean>((resolve) => {
+      controlConnection.query('SELECT 1', (error) => resolve(!error));
+      timers.setTimeout(timeout, undefined, { ref: false }).then(() => resolve(false));
+    });
+    // Only act if the probe failed on a connection that is still supposed to be alive.
+    if (
+      !responded &&
+      this.controlConnection === controlConnection &&
+      !this.zongji.stopped &&
+      !(this.isStopped || this.isStopping)
+    ) {
+      this.logger.warn('MySQL control connection is unresponsive. Stopping the BinLog Listener...');
+      this.listenerError = new Error('MySQL control connection is unresponsive.');
+      controlConnection._socket?.destroy();
+      await this.stop();
+    }
+  }
+
   private createProcessingQueue(): async.QueueObject<BinLogEvent> {
     const queue = async.queue(this.createQueueWorker(), 1);
 
@@ -268,8 +356,9 @@ export class BinLogListener {
     return queue;
   }
 
-  private createZongjiListener(): ZongJi {
-    const zongji = this.connectionManager.createBinlogListener();
+  private createZongjiListener(): BinlogListenerConnections {
+    const connections = this.connectionManager.createBinlogListener();
+    const { zongji } = connections;
 
     zongji.on('binlog', async (evt) => {
       this.logger.debug(`Received BinLog event:${evt.getEventName()}`);
@@ -298,7 +387,7 @@ export class BinLogListener {
       }
     });
 
-    return zongji;
+    return connections;
   }
 
   isQueueOverCapacity(): boolean {

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

@@ -0,0 +1,12 @@
+// Minimal type declarations for @vlasky/mysql, which ships without any.
+// Only the surface used by this module is declared.
+declare module '@vlasky/mysql' {
+  import { MySQLConnection } from '@powersync/mysql-zongji';
+
+  export interface VlaskyConnection extends MySQLConnection {
+    destroy(): void;
+    state: string;
+  }
+
+  export function createConnection(options: Record<string, unknown>): VlaskyConnection;
+}

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 } : {}),