diff --git a/.changeset/mysql-tcp-keepalive.md b/.changeset/mysql-tcp-keepalive.md new file mode 100644 index 000000000..c59ac56ca --- /dev/null +++ b/.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. diff --git a/modules/module-mysql/package.json b/modules/module-mysql/package.json index ac1905adc..a5acd3908 100644 --- a/modules/module-mysql/package.json +++ b/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", diff --git a/modules/module-mysql/src/replication/BinLogReplicationJob.ts b/modules/module-mysql/src/replication/BinLogReplicationJob.ts index c303c21b0..321a801d1 100644 --- a/modules/module-mysql/src/replication/BinLogReplicationJob.ts +++ b/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() { diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index 5e2396df2..0dbe42abc 100644 --- a/modules/module-mysql/src/replication/BinLogStream.ts +++ b/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; @@ -462,6 +464,7 @@ export class BinLogStream { activeServerUuid: this.activeServerUuid!, eventHandler: binlogEventHandler }); + this.binLogListener = binlogListener; this.abortSignal.addEventListener( 'abort', @@ -697,6 +700,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'); diff --git a/modules/module-mysql/src/replication/MySQLConnectionManager.ts b/modules/module-mysql/src/replication/MySQLConnectionManager.ts index 71806bd1e..c58152ed7 100644 --- a/modules/module-mysql/src/replication/MySQLConnectionManager.ts +++ b/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 { /** * Pool that can create streamable connections @@ -20,6 +31,7 @@ export class MySQLConnectionManager extends BaseObserver; /** @@ -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; /** @@ -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,7 +240,9 @@ export class BinLogListener { private async restartZongji(): Promise { if (this.zongji.stopped) { - this.zongji = this.createZongjiListener(); + const { zongji, controlConnection } = this.createZongjiListener(); + this.zongji = zongji; + this.controlConnection = controlConnection; await this.start(true); } } @@ -222,12 +250,29 @@ export class BinLogListener { private async stopZongji(): Promise { if (!this.zongji.stopped) { this.logger.info('Stopping BinLog Listener...'); - await new Promise((resolve) => { + let stopped = false; + const stopPromise = new Promise((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 { 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 { diff --git a/modules/module-mysql/src/types/vlasky-mysql.d.ts b/modules/module-mysql/src/types/vlasky-mysql.d.ts new file mode 100644 index 000000000..ab3b61173 --- /dev/null +++ b/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): VlaskyConnection; +} diff --git a/modules/module-mysql/src/utils/mysql-utils.ts b/modules/module-mysql/src/utils/mysql-utils.ts index 444c106ba..e658ccbba 100644 --- a/modules/module-mysql/src/utils/mysql-utils.ts +++ b/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 } : {}), diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index c5c60f0e4..7c02d4fe0 100644 --- a/modules/module-mysql/test/src/BinLogListener.test.ts +++ b/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 = { @@ -59,6 +70,104 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { expect(stopSpy).toHaveBeenCalled(); expect(queueStopSpy).toHaveBeenCalled(); + // Zongji does not destroy connections it did not create, so the listener has to. + expect(binLogListener.controlConnection.state).toBe('disconnected'); + }); + + 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 } = binLogListener.zongji as unknown as { connection: ConnectionWithConfig }; + const controlConnection = binLogListener.controlConnection as ConnectionWithConfig; + + for (const conn of [connection, controlConnection]) { + 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. + vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation(() => {}); + + await binLogListener.stop(); + + expect(binLogListener.zongji.stopped).toBeTruthy(); + }); + + test('Probe on a healthy control connection passes', { timeout: 20_000 }, async () => { + await binLogListener.start(); + + // No mocking: the real driver must accept the options form of query and answer the probe. + const controlConnection = binLogListener.controlConnection; + const realQuery = controlConnection.query.bind(controlConnection); + const probeError = new Promise((resolve) => { + vi.spyOn(controlConnection, 'query').mockImplementation(((options: any, callback: any) => { + realQuery(options, (error: any, results: any, fields: any) => { + callback(error, results, fields); + resolve(error); + }); + }) as any); + }); + + binLogListener.probeControlConnection(); + + expect(await probeError).toBeNull(); + expect(binLogListener.zongji.stopped).toBeFalsy(); + + await binLogListener.stop(); + }); + + test('Probe detects an unresponsive control connection', { timeout: 20_000 }, async () => { + binLogListener = await createBinlogListener({ + connectionManager, + sourceTables: [new TablePattern(connectionManager.databaseName, 'test_DATA')], + eventHandler, + ctrlConnectionProbeTimeoutMs: 500 + }); + await binLogListener.start(); + + // The probe query starts executing but never gets a response, like a connection that died + // without either side being notified: the driver's query timeout fires. + vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation(((_options: any, callback: any) => { + const error: any = new Error('Query inactivity timeout'); + error.code = 'PROTOCOL_SEQUENCE_TIMEOUT'; + setTimeout(() => callback(error), 10); + }) as any); + + const replication = binLogListener.replicateUntilStopped(); + binLogListener.probeControlConnection(); + + await expect(replication).rejects.toThrow('control connection is unresponsive'); + expect(binLogListener.zongji.stopped).toBeTruthy(); + }); + + test('Probe queued behind a busy control connection does not report it dead', { timeout: 20_000 }, async () => { + binLogListener = await createBinlogListener({ + connectionManager, + sourceTables: [new TablePattern(connectionManager.databaseName, 'test_DATA')], + eventHandler, + ctrlConnectionProbeTimeoutMs: 500 + }); + await binLogListener.start(); + + // The probe never even starts executing, as if queued behind a long-running metadata query on + // a healthy connection. The probe must not report the connection dead, and further probes must + // not pile up behind the pending one. + const querySpy = vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation((() => {}) as any); + + binLogListener.probeControlConnection(); + binLogListener.probeControlConnection(); + + expect(querySpy).toHaveBeenCalledTimes(1); + expect(binLogListener.zongji.stopped).toBeFalsy(); + + querySpy.mockRestore(); + await binLogListener.stop(); }); test('Zongji listener is stopped when processing queue reaches maximum memory size', async () => { diff --git a/modules/module-mysql/test/src/mysql-utils.test.ts b/modules/module-mysql/test/src/mysql-utils.test.ts index 7b0579083..b52574fd8 100644 --- a/modules/module-mysql/test/src/mysql-utils.test.ts +++ b/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 } }).config; + + expect(connectionConfig.enableKeepAlive).toBe(true); + expect(connectionConfig.keepAliveInitialDelay).toBe(TCP_KEEPALIVE_INITIAL_DELAY); + + await pool.promise().end(); + }); }); diff --git a/modules/module-mysql/test/src/util.ts b/modules/module-mysql/test/src/util.ts index ee580a1d0..7207ee2ca 100644 --- a/modules/module-mysql/test/src/util.ts +++ b/modules/module-mysql/test/src/util.ts @@ -94,6 +94,7 @@ export interface CreateBinlogListenerParams { eventHandler: BinLogEventHandler; sourceTables: TablePattern[]; startGTID?: common.ReplicatedGTID; + ctrlConnectionProbeTimeoutMs?: number; } export async function createBinlogListener(params: CreateBinlogListenerParams): Promise { let { connectionManager, eventHandler, sourceTables, startGTID } = params; @@ -110,7 +111,8 @@ export async function createBinlogListener(params: CreateBinlogListenerParams): startGTID: startGTID!, sourceTables: sourceTables, serverId: createRandomServerId(1), - activeServerUuid: activeServerUuid + activeServerUuid: activeServerUuid, + ctrlConnectionProbeTimeoutMs: params.ctrlConnectionProbeTimeoutMs }); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1a491078e..17f01729a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -427,6 +427,9 @@ importers: '@powersync/service-types': specifier: workspace:* version: link:../../packages/types + '@vlasky/mysql': + specifier: ^2.18.6 + version: 2.18.6 async: specifier: ^3.2.4 version: 3.2.5