[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/BinLogReplicationJob.ts

@@ -22,7 +22,9 @@ export class BinLogReplicationJob extends replication.AbstractReplicationJob {
   }
 
   async keepAlive() {
-    // Keepalives are handled by the binlog heartbeat mechanism
+    // The binlog connection is kept alive by the MySQL server heartbeat mechanism. The control
+    // connection carries no traffic between metadata queries, so probe its liveness here.
+    this.lastStream?.probeControlConnection();
   }
 
   async replicate() {

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

@@ -79,6 +79,8 @@ export class BinLogStream {
 
   private replicationLag = new ReplicationLagTracker();
 
+  private binLogListener: BinLogListener | null = null;
+
   constructor(private options: BinLogStreamOptions) {
     this.logger = options.logger ?? defaultLogger;
     this.storage = options.storage;
@@ -465,6 +467,7 @@ export class BinLogStream {
             activeServerUuid: this.activeServerUuid!,
             eventHandler: binlogEventHandler
           });
+          this.binLogListener = binlogListener;
 
           this.abortSignal.addEventListener(
             'abort',
@@ -700,6 +703,14 @@ export class BinLogStream {
     return this.replicationLag.getLagMillis();
   }
 
+  /**
+   * Probe the liveness of the BinLog Listener's control connection. Called from the replication
+   * job's keepAlive. Does nothing before streaming starts (during the initial snapshot).
+   */
+  probeControlConnection(): void {
+    this.binLogListener?.probeControlConnection();
+  }
+
   async tryRollback(promiseConnection: mysqlPromise.Connection) {
     try {
       await promiseConnection.query('ROLLBACK');

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,19 @@ 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;
+
+/**
+ *  Maximum time in milliseconds a control connection liveness probe may execute before the
+ *  connection is considered dead. Time the probe spends queued behind other control queries
+ *  does not count towards this.
+ */
+export const CTRL_CONNECTION_PROBE_TIMEOUT = 5_000;
+
 export type Row = Record<string, any>;
 
 /**
@@ -87,6 +101,7 @@ export interface BinLogListenerOptions {
   startGTID: common.ReplicatedGTID;
   logger?: Logger;
   keepAliveInactivitySeconds?: number;
+  ctrlConnectionProbeTimeoutMs?: number;
 }
 
 /**
@@ -104,10 +119,19 @@ export class BinLogListener {
   private isStopped: boolean = false;
   private isStopping: boolean = false;
 
+  // Set while a control connection probe is awaiting a response, so repeated probes do not pile up behind it.
+  private probePending: boolean = false;
+
   // Flag to indicate if are currently in a transaction that involves multiple row mutation events.
   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 +146,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 +240,39 @@ 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();
+      // destroy() drops pending query callbacks, so a probe waiting on this connection would
+      // otherwise stay pending forever and disable probing after a restart.
+      this.probePending = false;
       this.logger.info('BinLog Listener stopped.');
     }
   }
@@ -253,6 +298,41 @@ export class BinLogListener {
     }
   }
 
+  /**
+   *  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 the replication job's keepAlive additionally
+   *  probes it with a lightweight query and stops the listener (restarting replication) if the probe
+   *  does not respond in time.
+   *
+   *  The driver starts the query timeout when the query begins executing, not when it is queued, so a
+   *  probe waiting behind a legitimately slow metadata query does not produce a false failure. A probe
+   *  queued on a dead socket is failed together with the queued query by TCP keepalive on the socket.
+   */
+  public probeControlConnection(): void {
+    if (this.probePending || this.zongji.stopped || this.isStopped || this.isStopping) {
+      return;
+    }
+    this.probePending = true;
+    const controlConnection = this.controlConnection;
+    const timeout = this.options.ctrlConnectionProbeTimeoutMs ?? CTRL_CONNECTION_PROBE_TIMEOUT;
+    controlConnection.query({ sql: 'SELECT 1', timeout }, (error) => {
+      this.probePending = false;
+      // Only act if the probe failed on a connection that is still supposed to be alive.
+      if (
+        error != null &&
+        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();
+        this.stop();
+      }
+    });
+  }
+
   private createProcessingQueue(): async.QueueObject<BinLogEvent> {
     const queue = async.queue(this.createQueueWorker(), 1);
 
@@ -268,8 +348,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 +379,7 @@ export class BinLogListener {
       }
     });
 
-    return zongji;
+    return connections;
   }
 
   isQueueOverCapacity(): boolean {

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

@@ -0,0 +1,17 @@
+// 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;
+    /**
+     * Options form of query. The driver starts `timeout` when the query begins executing, not when
+     * it is queued behind other queries on the connection.
+     */
+    query(options: { sql: string; timeout?: number }, callback: (error: any, results: any, fields: any) => void): void;
+  }
+
+  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 } : {}),