From 2f56362d1e452c66686c3204de5fcf59fba8e98e Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 20 Aug 2026 12:41:54 +0200 Subject: [PATCH 1/9] [MySQL] Enable TCP keepalive on replication and pool connections --- .../src/replication/MySQLConnectionManager.ts | 14 ++++++++++--- modules/module-mysql/src/utils/mysql-utils.ts | 11 ++++++++++ .../module-mysql/test/src/mysql-utils.test.ts | 20 ++++++++++++++++++- 3 files changed, 41 insertions(+), 4 deletions(-) diff --git a/modules/module-mysql/src/replication/MySQLConnectionManager.ts b/modules/module-mysql/src/replication/MySQLConnectionManager.ts index 71806bd1e..d905fcbf2 100644 --- a/modules/module-mysql/src/replication/MySQLConnectionManager.ts +++ b/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 { @@ -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(); + }); }); From eb855dfbfeb59687d018082ff0bec82dbaf8dff4 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 20 Aug 2026 12:42:24 +0200 Subject: [PATCH 2/9] [MySQL] Add timeout when stopping the BinLog listener --- .../src/replication/zongji/BinLogListener.ts | 33 +++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index 0edc568d9..b26965b0f 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/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; /** @@ -222,12 +239,24 @@ 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.zongji as ZongJiWithControlConnection).ctrlConnection._socket?.destroy(); + } + }); + await Promise.race([stopPromise, timeout]); this.logger.info('BinLog Listener stopped.'); } } From 77c8dd53b40e6f0da5ac80c09ecd174404ac7821 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 20 Aug 2026 12:42:49 +0200 Subject: [PATCH 3/9] [MySQL] Add keepalive and stop timeout tests and changeset --- .changeset/mysql-tcp-keepalive.md | 5 +++ .../test/src/BinLogListener.test.ts | 41 ++++++++++++++++++- 2 files changed, 45 insertions(+), 1 deletion(-) create mode 100644 .changeset/mysql-tcp-keepalive.md diff --git a/.changeset/mysql-tcp-keepalive.md b/.changeset/mysql-tcp-keepalive.md new file mode 100644 index 000000000..1754d2ed0 --- /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 bound the BinLog listener stop sequence, so that a dead control connection cannot stall shutdown for that same window. diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index c5c60f0e4..db6e5c697 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 = { @@ -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'); From 78dde33166e5535ab3aa5260113d8e823a0c8b19 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 20 Aug 2026 13:47:19 +0200 Subject: [PATCH 4/9] [MySQL] Add liveness probe on the control connection --- .changeset/mysql-tcp-keepalive.md | 2 +- .../src/replication/zongji/BinLogListener.ts | 52 +++++++++++++++++++ .../test/src/BinLogListener.test.ts | 19 +++++++ modules/module-mysql/test/src/util.ts | 6 ++- 4 files changed, 77 insertions(+), 2 deletions(-) diff --git a/.changeset/mysql-tcp-keepalive.md b/.changeset/mysql-tcp-keepalive.md index 1754d2ed0..c59ac56ca 100644 --- a/.changeset/mysql-tcp-keepalive.md +++ b/.changeset/mysql-tcp-keepalive.md @@ -2,4 +2,4 @@ '@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. +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/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index b26965b0f..fba5b89ba 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -46,6 +46,16 @@ export const KEEPALIVE_INACTIVITY_THRESHOLD = 30; */ 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; + // 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 }; @@ -104,6 +114,8 @@ export interface BinLogListenerOptions { startGTID: common.ReplicatedGTID; logger?: Logger; keepAliveInactivitySeconds?: number; + ctrlConnectionKeepAliveIntervalMs?: number; + ctrlConnectionKeepAliveTimeoutMs?: number; } /** @@ -272,9 +284,11 @@ export class BinLogListener { } public async replicateUntilStopped(): Promise { + 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); @@ -282,6 +296,44 @@ 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 we additionally probe it with a lightweight + * query and restart replication if it does not respond in time. + */ + private async keepAliveControlConnectionUntilStopped(): Promise { + 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 { + const zongji = this.zongji as ZongJiWithControlConnection; + if (zongji.stopped) { + return; + } + const timeout = this.options.ctrlConnectionKeepAliveTimeoutMs ?? CTRL_CONNECTION_KEEPALIVE_TIMEOUT; + const responded = await new Promise((resolve) => { + zongji.ctrlConnection.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.zongji === zongji && !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.'); + zongji.ctrlConnection._socket?.destroy(); + await this.stop(); + } + } + private createProcessingQueue(): async.QueueObject { const queue = async.queue(this.createQueueWorker(), 1); diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index db6e5c697..67b6b0b53 100644 --- a/modules/module-mysql/test/src/BinLogListener.test.ts +++ b/modules/module-mysql/test/src/BinLogListener.test.ts @@ -100,6 +100,25 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { expect(binLogListener.zongji.stopped).toBeTruthy(); }); + test('Keepalive probe detects an unresponsive control connection', { timeout: 20_000 }, async () => { + binLogListener = await createBinlogListener({ + connectionManager, + sourceTables: [new TablePattern(connectionManager.databaseName, 'test_DATA')], + eventHandler, + ctrlConnectionKeepAliveIntervalMs: 1_000, + ctrlConnectionKeepAliveTimeoutMs: 500 + }); + await binLogListener.start(); + + // Queries to the control connection never get a response, like a connection that died + // without either side being notified. + const { ctrlConnection } = binLogListener.zongji as unknown as { ctrlConnection: MySQLConnection }; + vi.spyOn(ctrlConnection, 'query').mockImplementation(() => {}); + + await expect(binLogListener.replicateUntilStopped()).rejects.toThrow('control connection is unresponsive'); + 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'); diff --git a/modules/module-mysql/test/src/util.ts b/modules/module-mysql/test/src/util.ts index ee580a1d0..0552cb296 100644 --- a/modules/module-mysql/test/src/util.ts +++ b/modules/module-mysql/test/src/util.ts @@ -94,6 +94,8 @@ export interface CreateBinlogListenerParams { eventHandler: BinLogEventHandler; sourceTables: TablePattern[]; startGTID?: common.ReplicatedGTID; + ctrlConnectionKeepAliveIntervalMs?: number; + ctrlConnectionKeepAliveTimeoutMs?: number; } export async function createBinlogListener(params: CreateBinlogListenerParams): Promise { let { connectionManager, eventHandler, sourceTables, startGTID } = params; @@ -110,7 +112,9 @@ export async function createBinlogListener(params: CreateBinlogListenerParams): startGTID: startGTID!, sourceTables: sourceTables, serverId: createRandomServerId(1), - activeServerUuid: activeServerUuid + activeServerUuid: activeServerUuid, + ctrlConnectionKeepAliveIntervalMs: params.ctrlConnectionKeepAliveIntervalMs, + ctrlConnectionKeepAliveTimeoutMs: params.ctrlConnectionKeepAliveTimeoutMs }); } From debe229ce0a0fb88a111285a5d674690a0ce8abc Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 20 Aug 2026 14:25:21 +0200 Subject: [PATCH 5/9] [MySQL] Probe the control connection and own its lifecycle --- modules/module-mysql/package.json | 5 +- .../src/replication/MySQLConnectionManager.ts | 36 +++++++++--- .../src/replication/zongji/BinLogListener.ts | 56 +++++++++++-------- .../module-mysql/src/types/vlasky-mysql.d.ts | 12 ++++ .../test/src/BinLogListener.test.ts | 16 +++--- pnpm-lock.yaml | 3 + 6 files changed, 85 insertions(+), 43 deletions(-) create mode 100644 modules/module-mysql/src/types/vlasky-mysql.d.ts 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/MySQLConnectionManager.ts b/modules/module-mysql/src/replication/MySQLConnectionManager.ts index d905fcbf2..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, 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; /** @@ -137,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; /** @@ -151,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); } @@ -243,7 +241,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); } } @@ -265,10 +265,12 @@ export class BinLogListener { 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(); + 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.'); } } @@ -316,20 +318,25 @@ export class BinLogListener { } private async probeControlConnection(): Promise { - const zongji = this.zongji as ZongJiWithControlConnection; - if (zongji.stopped) { + const controlConnection = this.controlConnection; + if (this.zongji.stopped) { return; } const timeout = this.options.ctrlConnectionKeepAliveTimeoutMs ?? CTRL_CONNECTION_KEEPALIVE_TIMEOUT; const responded = await new Promise((resolve) => { - zongji.ctrlConnection.query('SELECT 1', (error) => resolve(!error)); + 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.zongji === zongji && !zongji.stopped && !(this.isStopped || this.isStopping)) { + 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.'); - zongji.ctrlConnection._socket?.destroy(); + controlConnection._socket?.destroy(); await this.stop(); } } @@ -349,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()}`); @@ -379,7 +387,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..afda5624b --- /dev/null +++ b/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): VlaskyConnection; +} diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index 67b6b0b53..e30b0faee 100644 --- a/modules/module-mysql/test/src/BinLogListener.test.ts +++ b/modules/module-mysql/test/src/BinLogListener.test.ts @@ -70,18 +70,18 @@ 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, ctrlConnection } = binLogListener.zongji as unknown as { - connection: ConnectionWithConfig; - ctrlConnection: ConnectionWithConfig; - }; + const { connection } = binLogListener.zongji as unknown as { connection: ConnectionWithConfig }; + const controlConnection = binLogListener.controlConnection as ConnectionWithConfig; - for (const conn of [connection, ctrlConnection]) { + for (const conn of [connection, controlConnection]) { expect(conn.config.enableKeepAlive).toBe(true); expect(conn.config.keepAliveInitialDelay).toBe(TCP_KEEPALIVE_INITIAL_DELAY); } @@ -92,8 +92,7 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { // 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(() => {}); + vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation(() => {}); await binLogListener.stop(); @@ -112,8 +111,7 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { // Queries to the control connection never get a response, like a connection that died // without either side being notified. - const { ctrlConnection } = binLogListener.zongji as unknown as { ctrlConnection: MySQLConnection }; - vi.spyOn(ctrlConnection, 'query').mockImplementation(() => {}); + vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation(() => {}); await expect(binLogListener.replicateUntilStopped()).rejects.toThrow('control connection is unresponsive'); expect(binLogListener.zongji.stopped).toBeTruthy(); 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 From 78f21c95e97a7378ac369cee2b1230f57cb5921e Mon Sep 17 00:00:00 2001 From: bean1352 Date: Mon, 24 Aug 2026 13:27:30 +0200 Subject: [PATCH 6/9] [MySQL] Use the driver query timeout for the control connection probe --- .../src/replication/zongji/BinLogListener.ts | 79 ++++++++----------- .../module-mysql/src/types/vlasky-mysql.d.ts | 5 ++ 2 files changed, 39 insertions(+), 45 deletions(-) diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index 5b57b3812..6c8f9f6ea 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -41,14 +41,11 @@ export const KEEPALIVE_INACTIVITY_THRESHOLD = 30; export const ZONGJI_STOP_TIMEOUT = 5_000; /** - * Interval in milliseconds between liveness probes on the Zongji control connection. + * 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_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 const CTRL_CONNECTION_PROBE_TIMEOUT = 5_000; export type Row = Record; @@ -104,8 +101,7 @@ export interface BinLogListenerOptions { startGTID: common.ReplicatedGTID; logger?: Logger; keepAliveInactivitySeconds?: number; - ctrlConnectionKeepAliveIntervalMs?: number; - ctrlConnectionKeepAliveTimeoutMs?: number; + ctrlConnectionProbeTimeoutMs?: number; } /** @@ -123,6 +119,9 @@ 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; @@ -286,11 +285,9 @@ export class BinLogListener { } public async replicateUntilStopped(): Promise { - 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); @@ -301,44 +298,36 @@ 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 we additionally probe it with a lightweight - * query and restart replication if it does not respond in time. + * 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. */ - private async keepAliveControlConnectionUntilStopped(): Promise { - 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 { - const controlConnection = this.controlConnection; - if (this.zongji.stopped) { + public probeControlConnection(): void { + if (this.probePending || this.zongji.stopped || this.isStopped || this.isStopping) { return; } - const timeout = this.options.ctrlConnectionKeepAliveTimeoutMs ?? CTRL_CONNECTION_KEEPALIVE_TIMEOUT; - const responded = await new Promise((resolve) => { - controlConnection.query('SELECT 1', (error) => resolve(!error)); - timers.setTimeout(timeout, undefined, { ref: false }).then(() => resolve(false)); + 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(); + } }); - // 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 { diff --git a/modules/module-mysql/src/types/vlasky-mysql.d.ts b/modules/module-mysql/src/types/vlasky-mysql.d.ts index afda5624b..ab3b61173 100644 --- a/modules/module-mysql/src/types/vlasky-mysql.d.ts +++ b/modules/module-mysql/src/types/vlasky-mysql.d.ts @@ -6,6 +6,11 @@ declare module '@vlasky/mysql' { 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; From 16a91f912f4e875b3865876e15a57f28d35d2c53 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Mon, 24 Aug 2026 13:28:12 +0200 Subject: [PATCH 7/9] [MySQL] Trigger the control connection probe from the replication job keepAlive --- .../src/replication/BinLogReplicationJob.ts | 4 +++- modules/module-mysql/src/replication/BinLogStream.ts | 11 +++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) 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'); From 2f6db3b21d271817f5f299b101d9c3dbad84f54c Mon Sep 17 00:00:00 2001 From: bean1352 Date: Mon, 24 Aug 2026 13:28:43 +0200 Subject: [PATCH 8/9] [MySQL] Update control connection probe tests --- .../test/src/BinLogListener.test.ts | 67 +++++++++++++++++-- modules/module-mysql/test/src/util.ts | 6 +- 2 files changed, 62 insertions(+), 11 deletions(-) diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index e30b0faee..7c02d4fe0 100644 --- a/modules/module-mysql/test/src/BinLogListener.test.ts +++ b/modules/module-mysql/test/src/BinLogListener.test.ts @@ -99,24 +99,77 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { expect(binLogListener.zongji.stopped).toBeTruthy(); }); - test('Keepalive probe detects an unresponsive control connection', { timeout: 20_000 }, async () => { + 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, - ctrlConnectionKeepAliveIntervalMs: 1_000, - ctrlConnectionKeepAliveTimeoutMs: 500 + ctrlConnectionProbeTimeoutMs: 500 }); await binLogListener.start(); - // Queries to the control connection never get a response, like a connection that died - // without either side being notified. - vi.spyOn(binLogListener.controlConnection, 'query').mockImplementation(() => {}); + // 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(binLogListener.replicateUntilStopped()).rejects.toThrow('control connection is unresponsive'); + 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 () => { const stopSpy = vi.spyOn(binLogListener.zongji, 'stop'); diff --git a/modules/module-mysql/test/src/util.ts b/modules/module-mysql/test/src/util.ts index 0552cb296..7207ee2ca 100644 --- a/modules/module-mysql/test/src/util.ts +++ b/modules/module-mysql/test/src/util.ts @@ -94,8 +94,7 @@ export interface CreateBinlogListenerParams { eventHandler: BinLogEventHandler; sourceTables: TablePattern[]; startGTID?: common.ReplicatedGTID; - ctrlConnectionKeepAliveIntervalMs?: number; - ctrlConnectionKeepAliveTimeoutMs?: number; + ctrlConnectionProbeTimeoutMs?: number; } export async function createBinlogListener(params: CreateBinlogListenerParams): Promise { let { connectionManager, eventHandler, sourceTables, startGTID } = params; @@ -113,8 +112,7 @@ export async function createBinlogListener(params: CreateBinlogListenerParams): sourceTables: sourceTables, serverId: createRandomServerId(1), activeServerUuid: activeServerUuid, - ctrlConnectionKeepAliveIntervalMs: params.ctrlConnectionKeepAliveIntervalMs, - ctrlConnectionKeepAliveTimeoutMs: params.ctrlConnectionKeepAliveTimeoutMs + ctrlConnectionProbeTimeoutMs: params.ctrlConnectionProbeTimeoutMs }); } From ddf1e3dd0cebbd31c503f9f8a5c3799b625bb6c6 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Mon, 24 Aug 2026 13:32:32 +0200 Subject: [PATCH 9/9] [MySQL] Reset pending probe flag when the control connection is destroyed --- modules/module-mysql/src/replication/zongji/BinLogListener.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index 6c8f9f6ea..eaf218903 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -270,6 +270,9 @@ export class BinLogListener { 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.'); } }