diff --git a/.changeset/mysql-gtid-consistency-fixes.md b/.changeset/mysql-gtid-consistency-fixes.md new file mode 100644 index 000000000..8ad33175f --- /dev/null +++ b/.changeset/mysql-gtid-consistency-fixes.md @@ -0,0 +1,5 @@ +--- +'@powersync/service-module-mysql': minor +--- + +Improve MySQL GTID consistency and resume safety. PowerSync now derives replication heads from the active server's executed GTID set, validates stored GTIDs and binlog coordinates before resuming, and re-snapshots after source rewinds or server UUID changes. Replica sources and foreign-origin binlog transactions are rejected until multi-origin transaction ordering is supported, while heartbeat keepalives use the last committed position to prevent idle checkpoint stalls. diff --git a/modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts b/modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts index bcb3514ac..95961b4b8 100644 --- a/modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts +++ b/modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts @@ -260,12 +260,12 @@ export class MySQLRouteAPIAdapter implements api.RouteAPI { const { bucketStorage } = options; const lastCheckpoint = await bucketStorage.getCheckpoint(); - const current = lastCheckpoint.lsn - ? common.ReplicatedGTID.fromSerialized(lastCheckpoint.lsn) - : common.ReplicatedGTID.ZERO; - const connection = await this.pool.getConnection(); const head = await common.readExecutedGtid(connection); + + const current = lastCheckpoint.lsn + ? common.ReplicatedGTID.fromSerialized(lastCheckpoint.lsn) + : common.ReplicatedGTID.ZERO(await common.readServerUuid(connection)); const lag = await current.distanceTo(connection, head); connection.release(); if (lag == null) { diff --git a/modules/module-mysql/src/common/ReplicatedGTID.ts b/modules/module-mysql/src/common/ReplicatedGTID.ts index dc7713e91..433a942d2 100644 --- a/modules/module-mysql/src/common/ReplicatedGTID.ts +++ b/modules/module-mysql/src/common/ReplicatedGTID.ts @@ -1,3 +1,4 @@ +import { ReplicationAssertionError } from '@powersync/lib-services-framework'; import mysql from 'mysql2/promise'; import * as uuid from 'uuid'; import * as mysql_utils from '../utils/mysql-utils.js'; @@ -8,7 +9,11 @@ export type BinLogPosition = { }; export type ReplicatedGTIDSpecification = { - raw_gtid: string; + /** + * The raw Global Transaction ID. This is of the format `server_uuid:transaction_id`. + * Must be a single GTID — not a GTID set (multiple UUIDs) or interval range. + */ + rawGtid: string; /** * The (end) position in a BinLog file where this transaction has been replicated in. */ @@ -16,12 +21,12 @@ export type ReplicatedGTIDSpecification = { }; export type BinLogGTIDFormat = { - server_id: Buffer; - transaction_range: number; + serverUuid: Buffer; + transactionId: number; }; export type BinLogGTIDEvent = { - raw_gtid: BinLogGTIDFormat; + rawGtid: BinLogGTIDFormat; position: BinLogPosition; }; @@ -31,30 +36,43 @@ export type BinLogGTIDEvent = { * and position where this GTID could be located. */ export class ReplicatedGTID { + private options: ReplicatedGTIDSpecification; + + constructor(options: ReplicatedGTIDSpecification) { + const rawGtid = options.rawGtid.trim(); + assertSingleGtid(rawGtid); + this.options = { ...options, rawGtid }; + } + static fromSerialized(comparable: string): ReplicatedGTID { return new ReplicatedGTID(ReplicatedGTID.deserialize(comparable)); } private static deserialize(comparable: string): ReplicatedGTIDSpecification { const components = comparable.split('|'); - if (components.length < 3) { - throw new Error(`Invalid serialized GTID: ${comparable}`); + if (components.length < 4) { + throw new ReplicationAssertionError(`Invalid serialized GTID: ${comparable}`); + } + + const offset = parseInt(components[3], 10); + if (Number.isNaN(offset)) { + throw new ReplicationAssertionError(`Invalid BinLog offset in serialized GTID: ${comparable}`); } return { - raw_gtid: components[1], + rawGtid: components[1], position: { filename: components[2], - offset: parseInt(components[3]) + offset: offset } satisfies BinLogPosition }; } static fromBinLogEvent(event: BinLogGTIDEvent) { - const { raw_gtid, position } = event; - const stringGTID = `${uuid.stringify(raw_gtid.server_id)}:${raw_gtid.transaction_range}`; + const { rawGtid, position } = event; + const stringGTID = `${uuid.stringify(rawGtid.serverUuid)}:${rawGtid.transactionId}`; return new ReplicatedGTID({ - raw_gtid: stringGTID, + rawGtid: stringGTID, position }); } @@ -62,9 +80,12 @@ export class ReplicatedGTID { /** * Special case for the zero GTID which means no transactions have been executed. */ - static ZERO = new ReplicatedGTID({ raw_gtid: '0:0', position: { filename: '', offset: 0 } }); - - constructor(protected options: ReplicatedGTIDSpecification) {} + static ZERO(serverUuid: string): ReplicatedGTID { + return new ReplicatedGTID({ + rawGtid: `${serverUuid}:0`, + position: { filename: '', offset: 0 } + }); + } /** * Get the BinLog position of this replicated GTID event @@ -74,14 +95,17 @@ export class ReplicatedGTID { } /** - * Get the raw Global Transaction ID. This of the format `server_id:transaction_ranges` + * Get the raw Global Transaction ID. This is of the format `server_uuid:transaction_id` */ get raw() { - return this.options.raw_gtid; + return this.options.rawGtid; } - get serverId() { - return this.options.raw_gtid.split(':')[0]; + /** + * The server UUID of the server this transaction originated from + */ + get serverUuid() { + return this.options.rawGtid.split(':')[0]; } /** @@ -94,21 +118,9 @@ export class ReplicatedGTID { */ get comparable(): string { const { raw, position } = this; - const [, transactionRanges] = this.raw.split(':'); + const [, transactionId] = this.raw.split(':'); - // This means no transactions have been executed on the database yet - if (!transactionRanges) { - return ReplicatedGTID.ZERO.comparable; - } - - let maxTransactionId = 0; - - for (const range of transactionRanges.split(',')) { - const [start, end] = range.split('-'); - maxTransactionId = Math.max(maxTransactionId, parseInt(start, 10), parseInt(end || start, 10)); - } - - const paddedTransactionId = maxTransactionId.toString().padStart(16, '0'); + const paddedTransactionId = transactionId.toString().padStart(16, '0'); return [paddedTransactionId, raw, position.filename, position.offset].join('|'); } @@ -161,3 +173,24 @@ export class ReplicatedGTID { ); } } + +/** + * Asserts that the given gtid string is a single GTID of the form `server_uuid:transaction_id`, + * not a GTID set such as `uuid:1-17` or `uuid1:1,uuid2:2`. + */ +function assertSingleGtid(gtid: string): void { + // GTID sets join UUID sets with commas (often with newlines: `,\n`). + if (gtid.includes(',') || gtid.includes('\n')) { + throw new ReplicationAssertionError(`Expected a single GTID (server_uuid:transaction_id), got a GTID set: ${gtid}`); + } + + const parts = gtid.split(':'); + if (parts.length !== 2 || parts[0].length === 0 || parts[1].length === 0) { + throw new ReplicationAssertionError(`Expected a single GTID (server_uuid:transaction_id), got: ${gtid}`); + } + + // Intervals use `n-m`; a single transaction id must be a non-negative integer. + if (!/^\d+$/.test(parts[1])) { + throw new ReplicationAssertionError(`Expected a single transaction id, got: ${gtid}`); + } +} diff --git a/modules/module-mysql/src/common/check-source-configuration.ts b/modules/module-mysql/src/common/check-source-configuration.ts index 8bb0888f3..9cc9937cd 100644 --- a/modules/module-mysql/src/common/check-source-configuration.ts +++ b/modules/module-mysql/src/common/check-source-configuration.ts @@ -2,6 +2,7 @@ import mysqlPromise from 'mysql2/promise'; import * as mysql_utils from '../utils/mysql-utils.js'; const MIN_SUPPORTED_VERSION = '5.7.0'; +const REPLICA_TERMINOLOGY_VERSION = '8.0.22'; export async function checkSourceConfiguration(connection: mysqlPromise.Connection): Promise { const errors: string[] = []; @@ -11,6 +12,20 @@ export async function checkSourceConfiguration(connection: mysqlPromise.Connecti errors.push(`MySQL versions older than ${MIN_SUPPORTED_VERSION} are not supported. Your version is: ${version}.`); } + const replicaStatusQuery = mysql_utils.isVersionAtLeast(version, REPLICA_TERMINOLOGY_VERSION) + ? 'SHOW REPLICA STATUS' + : 'SHOW SLAVE STATUS'; + const [replicaStatuses] = await mysql_utils.retriedQuery({ + connection, + query: replicaStatusQuery + }); + + if (replicaStatuses.length > 0) { + errors.push( + 'Connecting PowerSync to a MySQL replica is not supported. Please connect PowerSync directly to the primary server.' + ); + } + const [[result]] = await mysql_utils.retriedQuery({ connection, query: ` diff --git a/modules/module-mysql/src/common/read-executed-gtid.ts b/modules/module-mysql/src/common/read-executed-gtid.ts index 9f60c3362..1be9a6635 100644 --- a/modules/module-mysql/src/common/read-executed-gtid.ts +++ b/modules/module-mysql/src/common/read-executed-gtid.ts @@ -2,6 +2,17 @@ import mysqlPromise from 'mysql2/promise'; import * as mysql_utils from '../utils/mysql-utils.js'; import { ReplicatedGTID } from './ReplicatedGTID.js'; +/** + * Gets the `@@server_uuid` of the current connected server + */ +export async function readServerUuid(connection: mysqlPromise.Connection): Promise { + const [[result]] = await mysql_utils.retriedQuery({ + connection, + query: `SELECT @@server_uuid AS server_uuid` + }); + return result.server_uuid; +} + /** * Gets the current master HEAD GTID */ @@ -28,21 +39,80 @@ export async function readExecutedGtid(connection: mysqlPromise.Connection): Pro offset: parseInt(binlogStatus.Position) }; + const activeServerUuid = await readServerUuid(connection); + const executedGtidSet = binlogStatus.Executed_Gtid_Set.trim(); + + if (executedGtidSet.length === 0) { + // New server with no transactions executed yet. Keep the current binlog + // coordinate so this synthetic GTID can still be validated after a restart. + return new ReplicatedGTID({ + rawGtid: `${activeServerUuid}:0`, + position + }); + } + + const gtidSets = executedGtidSet.split(','); + const latestActiveGtid = await getLatestActiveGtid(gtidSets, activeServerUuid); return new ReplicatedGTID({ - // The head always points to the next position to start replication from - position, - raw_gtid: binlogStatus.Executed_Gtid_Set + rawGtid: latestActiveGtid, + position }); } -export async function isBinlogStillAvailable( +export async function getLatestActiveGtid(gtidSets: string[], activeServerUuid: string): Promise { + for (const gtidSet of gtidSets) { + const [serverUuid, ...intervals] = gtidSet.trim().split(':'); + if (serverUuid === activeServerUuid) { + let maxTransactionId: number | null = null; + for (const interval of intervals) { + const [start, end] = interval.split('-'); + const startId = parseInt(start, 10); + const endId = end !== undefined ? parseInt(end, 10) : startId; + if (!Number.isNaN(startId)) { + maxTransactionId = Math.max(maxTransactionId ?? 0, startId); + } + if (!Number.isNaN(endId)) { + maxTransactionId = Math.max(maxTransactionId ?? 0, endId); + } + } + return activeServerUuid + ':' + maxTransactionId; + } + } + + return `${activeServerUuid}:0`; +} + +/** + * Checks that a stored resume GTID is still part of the server's executed history and that its + * binlog coordinate is still readable. This detects source rewinds where a restored server keeps + * the same UUID or recreates a binlog with the same filename but a shorter length. + */ +export async function isGtidPositionStillAvailable( connection: mysqlPromise.Connection, - binlogFile: string + gtid: ReplicatedGTID ): Promise { const [logFiles] = await mysql_utils.retriedQuery({ connection, query: `SHOW BINARY LOGS;` }); + const logFile = logFiles.find((file) => file['Log_name'] == gtid.position.filename); + + if (!logFile || Number(logFile['File_size']) < gtid.position.offset) { + return false; + } + + // Transaction zero is PowerSync's synthetic position before the first + // transaction from this server UUID. It is not valid MySQL GTID_SET syntax, + // so its availability is determined by the binlog coordinate above. + if (gtid.raw.split(':')[1] === '0') { + return true; + } + + const [[result]] = await mysql_utils.retriedQuery({ + connection, + query: `SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed`, + params: [gtid.raw] + }); - return logFiles.some((f) => f['Log_name'] == binlogFile); + return result.is_executed === 1; } diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index 3fed923ce..7274a46fc 100644 --- a/modules/module-mysql/src/replication/BinLogStream.ts +++ b/modules/module-mysql/src/replication/BinLogStream.ts @@ -73,6 +73,8 @@ export class BinLogStream { private readonly logger: Logger; + private activeServerUuid: string | null = null; + private tableCache = new Map(); private replicationLag = new ReplicationLagTracker(); @@ -220,16 +222,23 @@ export class BinLogStream { this.logger.info(`Initial replication already done.`); if (lastKnowGTID) { - // Check if the specific binlog file is still available. If it isn't, we need to snapshot again. const connection = await this.connections.getConnection(); try { - const isAvailable = await common.isBinlogStillAvailable(connection, lastKnowGTID.position.filename); + // Check if the active server uuid matches the one in the GTID + if (this.activeServerUuid !== lastKnowGTID.serverUuid) { + this.logger.info( + `The source server uuid has changed. Active server uuid ${this.activeServerUuid} does not match the server uuid from the resume checkpoint: ${lastKnowGTID.serverUuid}, re-snapshotting to ensure consistency.` + ); + return false; + } + + const isAvailable = await common.isGtidPositionStillAvailable(connection, lastKnowGTID); if (!isAvailable) { this.logger.info( - `BinLog file ${lastKnowGTID.position.filename} is no longer available, starting initial replication again.` + `Resume GTID ${lastKnowGTID.raw} at ${lastKnowGTID.position.filename}:${lastKnowGTID.position.offset} is no longer present in the executed GTID history or available BinLogs, re-snapshotting to ensure consistency.` ); + return false; } - return isAvailable; } finally { connection.release(); } @@ -267,7 +276,7 @@ export class BinLogStream { const flushResults = await this.storage.startBatch( { logger: this.logger, - zeroLSN: common.ReplicatedGTID.ZERO.comparable, + zeroLSN: common.ReplicatedGTID.ZERO(this.activeServerUuid!).comparable, defaultSchema: this.defaultSchema, storeCurrentData: false }, @@ -363,7 +372,19 @@ export class BinLogStream { } } + private async ensureActiveServerUuid() { + if (this.activeServerUuid == null) { + const connection = await this.connections.getConnection(); + try { + this.activeServerUuid = await common.readServerUuid(connection); + } finally { + connection.release(); + } + } + } + async initReplication() { + await this.ensureActiveServerUuid(); const connection = await this.connections.getConnection(); const errors = await common.checkSourceConfiguration(connection); connection.release(); @@ -382,7 +403,7 @@ export class BinLogStream { await this.storage.startBatch( { logger: this.logger, - zeroLSN: common.ReplicatedGTID.ZERO.comparable, + zeroLSN: common.ReplicatedGTID.ZERO(this.activeServerUuid!).comparable, defaultSchema: this.defaultSchema, storeCurrentData: false }, @@ -406,21 +427,30 @@ export class BinLogStream { } async streamChanges() { + await this.ensureActiveServerUuid(); const serverId = createRandomServerId(this.storage.replicationStreamId); const connection = await this.connections.getConnection(); - const { resumeLsn: resume_lsn } = await this.storage.getStatus(); - if (resume_lsn) { - this.logger.info(`Existing resume LSN found: ${resume_lsn}`); + let fromGTID: common.ReplicatedGTID; + try { + const { resumeLsn: resume_lsn } = await this.storage.getStatus(); + if (resume_lsn) { + this.logger.info(`Existing resume LSN found: ${resume_lsn}`); + } + fromGTID = resume_lsn + ? common.ReplicatedGTID.fromSerialized(resume_lsn) + : await common.readExecutedGtid(connection); + } finally { + connection.release(); } - const fromGTID = resume_lsn - ? common.ReplicatedGTID.fromSerialized(resume_lsn) - : await common.readExecutedGtid(connection); - connection.release(); if (!this.stopped) { await this.storage.startBatch( - { zeroLSN: common.ReplicatedGTID.ZERO.comparable, defaultSchema: this.defaultSchema, storeCurrentData: false }, + { + zeroLSN: common.ReplicatedGTID.ZERO(this.activeServerUuid!).comparable, + defaultSchema: this.defaultSchema, + storeCurrentData: false + }, async (batch) => { const binlogEventHandler = this.createBinlogEventHandler(batch); const binlogListener = new BinLogListener({ @@ -429,6 +459,7 @@ export class BinLogStream { startGTID: fromGTID, connectionManager: this.connections, serverId: serverId, + activeServerUuid: this.activeServerUuid!, eventHandler: binlogEventHandler }); diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index 8b83ab00b..0edc568d9 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -1,4 +1,4 @@ -import { Logger, logger as defaultLogger } from '@powersync/lib-services-framework'; +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 async from 'async'; @@ -76,7 +76,14 @@ export interface BinLogListenerOptions { connectionManager: MySQLConnectionManager; eventHandler: BinLogEventHandler; sourceTables: TablePattern[]; + /** + * Id that identifies this replication client. + */ serverId: number; + /** + * The server uuid of the source MySQL server that is being replicated. + */ + activeServerUuid: string; startGTID: common.ReplicatedGTID; logger?: Logger; keepAliveInactivitySeconds?: number; @@ -88,8 +95,6 @@ export interface BinLogListenerOptions { */ export class BinLogListener { private sqlParser: ParserType; - private connectionManager: MySQLConnectionManager; - private eventHandler: BinLogEventHandler; private binLogPosition: common.BinLogPosition; private currentGTID: common.ReplicatedGTID; private logger: Logger; @@ -101,6 +106,7 @@ export class BinLogListener { // Flag to indicate if are currently in a transaction that involves multiple row mutation events. private isTransactionOpen = false; + zongji: ZongJi; processingQueue: async.QueueObject; @@ -111,9 +117,8 @@ export class BinLogListener { constructor(public options: BinLogListenerOptions) { this.logger = options.logger ?? defaultLogger; - this.connectionManager = options.connectionManager; - this.eventHandler = options.eventHandler; - this.binLogPosition = options.startGTID.position; + // Copy the position: the listener mutates it as events are processed, and the caller's startGTID must not change + this.binLogPosition = { ...options.startGTID.position }; this.currentGTID = options.startGTID; this.sqlParser = new Parser(); this.processingQueue = this.createProcessingQueue(); @@ -122,6 +127,18 @@ export class BinLogListener { this.databaseFilter = this.createDatabaseFilter(options.sourceTables); } + private get connectionManager(): MySQLConnectionManager { + return this.options.connectionManager; + } + + private get eventHandler(): BinLogEventHandler { + return this.options.eventHandler; + } + + private get activeServerUuid(): string { + return this.options.activeServerUuid; + } + /** * The queue memory limit in bytes as defined in the connection options. * @private @@ -292,29 +309,41 @@ export class BinLogListener { return async (evt: BinLogEvent) => { switch (true) { case zongji_utils.eventIsGTIDLog(evt): - this.currentGTID = common.ReplicatedGTID.fromBinLogEvent({ - raw_gtid: { - server_id: evt.serverId, - transaction_range: evt.transactionRange + const transactionGTID = common.ReplicatedGTID.fromBinLogEvent({ + rawGtid: { + serverUuid: evt.serverId, // The server uuid this transaction originated from + transactionId: evt.transactionRange }, position: { filename: this.binLogPosition.filename, offset: evt.nextPosition } }); + + if (transactionGTID.serverUuid !== this.activeServerUuid) { + throw new ReplicationAssertionError( + `Detected a transaction from a different MySQL server UUID: ${transactionGTID.serverUuid} than the server that is currently being replicated from: ${this.activeServerUuid}. ` + + `A re-snapshot is required to ensure consistency.` + ); + } + + this.currentGTID = transactionGTID; this.binLogPosition.offset = evt.nextPosition; + await this.eventHandler.onTransactionStart({ timestamp: new Date(evt.timestamp) }); this.logger.info(`Processed GTID event: ${this.currentGTID.comparable}`); break; case zongji_utils.eventIsRotation(evt): // The first event when starting replication is a synthetic Rotate event - // It describes the last binlog file and position that the replica client processed + // It describes the the position and file that the replica requested to start from + const isNewFile = this.binLogPosition.filename !== evt.binlogName; + this.binLogPosition.filename = evt.binlogName; - this.binLogPosition.offset = evt.nextPosition !== 0 ? evt.nextPosition : evt.position; + this.binLogPosition.offset = evt.position; + await this.eventHandler.onRotate(); - const newFile = this.binLogPosition.filename !== evt.binlogName; - if (newFile) { + if (isNewFile) { this.logger.info( `Processed Rotate event. New BinLog file is: ${this.binLogPosition.filename}:${this.binLogPosition.offset}` ); @@ -359,11 +388,7 @@ export class BinLogListener { break; case zongji_utils.eventIsXid(evt): this.isTransactionOpen = false; - this.binLogPosition.offset = evt.nextPosition; - const LSN = new common.ReplicatedGTID({ - raw_gtid: this.currentGTID.raw, - position: this.binLogPosition - }).comparable; + const LSN = this.advanceCommitPosition(evt.nextPosition); await this.eventHandler.onCommit(LSN); this.logger.info(`Processed Xid event - transaction complete. LSN: ${LSN}.`); break; @@ -376,6 +401,21 @@ export class BinLogListener { }; } + /** + * Advances the binlog position to the end of a committed transaction and updates the currentGTID to match. + * This ensures subsequent heartbeat keepalives report an LSN that is not behind the last commit LSN, + * which would otherwise block checkpoint creation until the next transaction arrives. + * Returns the commit LSN. + */ + private advanceCommitPosition(nextPosition: number): string { + this.binLogPosition.offset = nextPosition; + this.currentGTID = new common.ReplicatedGTID({ + rawGtid: this.currentGTID.raw, + position: { ...this.binLogPosition } + }); + return this.currentGTID.comparable; + } + private async processQueryEvent(event: BinLogQueryEvent): Promise { const { query, nextPosition } = event; @@ -398,11 +438,7 @@ export class BinLogListener { // DDL queries are auto commited, but do not come with a corresponding Xid event, in those cases we trigger a manual commit if we are not already in a transaction. // Some DDL queries include row events, and in those cases will include a Xid event. if (!this.isTransactionOpen) { - this.binLogPosition.offset = nextPosition; - const LSN = new common.ReplicatedGTID({ - raw_gtid: this.currentGTID.raw, - position: this.binLogPosition - }).comparable; + const LSN = this.advanceCommitPosition(nextPosition); await this.eventHandler.onCommit(LSN); } @@ -419,11 +455,7 @@ export class BinLogListener { await this.restartZongji(); } } else if (!this.isTransactionOpen) { - this.binLogPosition.offset = nextPosition; - const LSN = new common.ReplicatedGTID({ - raw_gtid: this.currentGTID.raw, - position: this.binLogPosition - }).comparable; + const LSN = this.advanceCommitPosition(nextPosition); await this.eventHandler.onCommit(LSN); } } diff --git a/modules/module-mysql/test/src/BinLogListener.test.ts b/modules/module-mysql/test/src/BinLogListener.test.ts index 135125c28..c5c60f0e4 100644 --- a/modules/module-mysql/test/src/BinLogListener.test.ts +++ b/modules/module-mysql/test/src/BinLogListener.test.ts @@ -111,6 +111,24 @@ describe('BinlogListener tests', { timeout: 60_000 }, () => { expect(eventHandler.lastKeepAlive).toEqual(binLogListener.options.startGTID.comparable); }); + test('Keepalive LSN after a commit is not less than the commit LSN', async () => { + binLogListener.options.keepAliveInactivitySeconds = 1; + await binLogListener.start(); + + await insertRows(connectionManager, 1); + await vi.waitFor(() => expect(eventHandler.commitCount).equals(1), { timeout: 5000 }); + const commitLsn = eventHandler.lastCommitLsn!; + + // Wait for a heartbeat keepalive that arrives after the commit. + // A keepalive LSN behind the commit LSN blocks checkpoint creation until the next transaction arrives. + await vi.waitFor(() => expect(eventHandler.lastKeepAlive && eventHandler.lastKeepAlive >= commitLsn).toBeTruthy(), { + timeout: 10_000 + }); + await binLogListener.stop(); + // No binlog rotation happens in this test, so the keepalive LSN should exactly match the commit LSN + expect(eventHandler.lastKeepAlive).toEqual(commitLsn); + }); + test('Schema change event: Rename table', async () => { await binLogListener.start(); await connectionManager.query(`ALTER TABLE test_DATA RENAME test_DATA_new`); diff --git a/modules/module-mysql/test/src/ReplicatedGTID.test.ts b/modules/module-mysql/test/src/ReplicatedGTID.test.ts new file mode 100644 index 000000000..f182b3229 --- /dev/null +++ b/modules/module-mysql/test/src/ReplicatedGTID.test.ts @@ -0,0 +1,138 @@ +import { ReplicatedGTID } from '@module/common/ReplicatedGTID.js'; +import * as uuid from 'uuid'; +import { describe, expect, test } from 'vitest'; + +describe('ReplicatedGTID', () => { + const SERVER_UUID = 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004'; + const POSITION = { filename: 'binlog.000042', offset: 1234 }; + + describe('single GTID', () => { + test('exposes its raw value, server UUID, and binlog position', () => { + const gtid = new ReplicatedGTID({ + rawGtid: `${SERVER_UUID}:5`, + position: POSITION + }); + + expect(gtid.raw).toEqual(`${SERVER_UUID}:5`); + expect(gtid.serverUuid).toEqual(SERVER_UUID); + expect(gtid.position).toEqual(POSITION); + }); + + test('formats a comparable LSN using the transaction id', () => { + const gtid = new ReplicatedGTID({ + rawGtid: `${SERVER_UUID}:17`, + position: POSITION + }); + + expect(gtid.comparable).toEqual(`0000000000000017|${SERVER_UUID}:17|binlog.000042|1234`); + expect(gtid.toString()).toEqual(gtid.comparable); + }); + + test('normalizes surrounding whitespace', () => { + const gtid = new ReplicatedGTID({ + rawGtid: ` \n\t${SERVER_UUID}:17 \r\n`, + position: POSITION + }); + + expect(gtid.raw).toEqual(`${SERVER_UUID}:17`); + expect(gtid.serverUuid).toEqual(SERVER_UUID); + expect(gtid.comparable).toEqual(`0000000000000017|${SERVER_UUID}:17|binlog.000042|1234`); + }); + + test('keeps the ZERO GTID format stable', () => { + expect(ReplicatedGTID.ZERO(SERVER_UUID).raw).toEqual(`${SERVER_UUID}:0`); + expect(ReplicatedGTID.ZERO(SERVER_UUID).comparable).toEqual(`0000000000000000|${SERVER_UUID}:0||0`); + }); + }); + + describe('validation', () => { + test.each([ + ['', 'missing server UUID and transaction id'], + [SERVER_UUID, 'missing transaction id'], + [`${SERVER_UUID}:`, 'empty transaction id'], + [`:${17}`, 'empty server UUID'], + [`${SERVER_UUID}:1-17`, 'transaction interval'], + [`${SERVER_UUID}:1:17`, 'multiple transaction components'], + [`${SERVER_UUID}:abc`, 'non-numeric transaction id'], + [`${SERVER_UUID}:-1`, 'negative transaction id'], + [`${SERVER_UUID}:17,another-server:9`, 'comma-separated GTID set'], + [`${SERVER_UUID}:17,\nanother-server:9`, 'newline-separated GTID set'] + ])('rejects %s (%s)', (rawGtid) => { + expect(() => new ReplicatedGTID({ rawGtid, position: POSITION })).toThrow(); + }); + }); + + describe('serialization', () => { + test('round-trips a single GTID', () => { + const gtid = new ReplicatedGTID({ + rawGtid: `${SERVER_UUID}:17`, + position: POSITION + }); + + const deserialized = ReplicatedGTID.fromSerialized(gtid.comparable); + + expect(deserialized.raw).toEqual(gtid.raw); + expect(deserialized.serverUuid).toEqual(SERVER_UUID); + expect(deserialized.position).toEqual(POSITION); + expect(deserialized.comparable).toEqual(gtid.comparable); + }); + + test('throws on malformed serialized GTIDs', () => { + expect(() => ReplicatedGTID.fromSerialized('abc')).toThrow('Invalid serialized GTID'); + expect(() => ReplicatedGTID.fromSerialized(`0000000000000001|${SERVER_UUID}:1|binlog.000001`)).toThrow( + 'Invalid serialized GTID' + ); + expect(() => ReplicatedGTID.fromSerialized(`0000000000000001|${SERVER_UUID}:1|binlog.000001|notanumber`)).toThrow( + 'Invalid BinLog offset' + ); + }); + + test('rejects a serialized GTID set', () => { + const serialized = `0000000000000017|${SERVER_UUID}:1-17|binlog.000042|1234`; + + expect(() => ReplicatedGTID.fromSerialized(serialized)).toThrow('Expected a single transaction id'); + }); + }); + + describe('binlog events', () => { + test('creates a single GTID from a binlog event', () => { + const gtid = ReplicatedGTID.fromBinLogEvent({ + rawGtid: { + serverUuid: Buffer.from(uuid.parse(SERVER_UUID)), + transactionId: 17 + }, + position: POSITION + }); + + expect(gtid.raw).toEqual(`${SERVER_UUID}:17`); + expect(gtid.position).toEqual(POSITION); + expect(gtid.comparable).toEqual(`0000000000000017|${SERVER_UUID}:17|binlog.000042|1234`); + }); + }); + + describe('LSN ordering', () => { + test('orders GTIDs from the same server by transaction id', () => { + const earlier = new ReplicatedGTID({ rawGtid: `${SERVER_UUID}:9`, position: POSITION }); + const later = new ReplicatedGTID({ rawGtid: `${SERVER_UUID}:18`, position: POSITION }); + + expect(earlier.comparable < later.comparable).toBeTruthy(); + }); + + test('orders LSNs for the same transaction by binlog offset', () => { + // The binlog offset is not zero-padded, so lexicographic ordering only holds for + // offsets with the same number of digits. This format cannot change while existing + // LSNs remain persisted in bucket storage. + const rawGtid = `${SERVER_UUID}:18`; + const transactionStart = new ReplicatedGTID({ + rawGtid, + position: { filename: 'binlog.000042', offset: 157 } + }); + const transactionEnd = new ReplicatedGTID({ + rawGtid, + position: { filename: 'binlog.000042', offset: 300 } + }); + + expect(transactionStart.comparable < transactionEnd.comparable).toBeTruthy(); + }); + }); +}); diff --git a/modules/module-mysql/test/src/check-source-configuration.test.ts b/modules/module-mysql/test/src/check-source-configuration.test.ts new file mode 100644 index 000000000..86c5c8769 --- /dev/null +++ b/modules/module-mysql/test/src/check-source-configuration.test.ts @@ -0,0 +1,70 @@ +import { checkSourceConfiguration } from '@module/common/check-source-configuration.js'; +import { describe, expect, test } from 'vitest'; +import { createMockMySQLConnection } from './util.js'; + +describe('checkSourceConfiguration', () => { + test('accepts a primary MySQL server', async () => { + const { connection, query } = createConnection({ version: '8.4.0', replicaStatuses: [] }); + + await expect(checkSourceConfiguration(connection)).resolves.toEqual([]); + expect(query).toHaveBeenCalledWith('SHOW REPLICA STATUS', []); + }); + + test('rejects a replica on MySQL 8.0.22 and later', async () => { + const { connection, query } = createConnection({ + version: '8.0.22', + replicaStatuses: [{ Channel_Name: '' }] + }); + + await expect(checkSourceConfiguration(connection)).resolves.toContain( + 'Connecting PowerSync to a MySQL replica is not supported. Please connect PowerSync directly to the primary server.' + ); + expect(query).toHaveBeenCalledWith('SHOW REPLICA STATUS', []); + expect(query).not.toHaveBeenCalledWith('SHOW SLAVE STATUS', []); + }); + + test('uses legacy replica-status syntax before MySQL 8.0.22', async () => { + const { connection, query } = createConnection({ + version: '5.7.44', + replicaStatuses: [{ Channel_Name: '' }] + }); + + await expect(checkSourceConfiguration(connection)).resolves.toContain( + 'Connecting PowerSync to a MySQL replica is not supported. Please connect PowerSync directly to the primary server.' + ); + expect(query).toHaveBeenCalledWith('SHOW SLAVE STATUS', []); + expect(query).not.toHaveBeenCalledWith('SHOW REPLICA STATUS', []); + }); + + function createConnection(options: { version: string; replicaStatuses: Record[] }) { + return createMockMySQLConnection(async (sql) => { + switch (sql.trim()) { + case 'SELECT VERSION() as version': + return [[{ version: options.version }], []]; + case 'SHOW REPLICA STATUS': + case 'SHOW SLAVE STATUS': + return [options.replicaStatuses, []]; + case "SHOW VARIABLES LIKE 'binlog_format';": + return [[{ Value: 'ROW' }], []]; + case "SHOW GLOBAL VARIABLES LIKE 'binlog_row_image';": + return [[{ Value: 'FULL' }], []]; + default: + if (sql.includes('@@GLOBAL.gtid_mode AS gtid_mode')) { + return [ + [ + { + gtid_mode: 'ON', + log_bin: 1, + server_id: 1, + binlog_file: '/var/lib/mysql/binlog', + binlog_index_file: '/var/lib/mysql/binlog.index' + } + ], + [] + ]; + } + throw new Error(`Unexpected query: ${sql}`); + } + }); + } +}); diff --git a/modules/module-mysql/test/src/read-executed-gtid.test.ts b/modules/module-mysql/test/src/read-executed-gtid.test.ts new file mode 100644 index 000000000..6a09410be --- /dev/null +++ b/modules/module-mysql/test/src/read-executed-gtid.test.ts @@ -0,0 +1,188 @@ +import { ReplicatedGTID } from '@module/common/ReplicatedGTID.js'; +import { + getLatestActiveGtid, + isGtidPositionStillAvailable, + readExecutedGtid +} from '@module/common/read-executed-gtid.js'; +import { describe, expect, test } from 'vitest'; +import { createMockMySQLConnection } from './util.js'; + +describe('read-executed-gtid', () => { + const ACTIVE_SERVER_UUID = 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004'; + const STALE_SERVER_UUID = '314306f3-ff7b-11ef-a0e0-566fbaa00002'; + + describe('getLatestActiveGtid', () => { + test('returns the highest transaction id for the active server', async () => { + const gtid = await getLatestActiveGtid( + [`${STALE_SERVER_UUID}:1-1000`, `\n${ACTIVE_SERVER_UUID}:1-5:9:12-17`], + ACTIVE_SERVER_UUID + ); + + expect(gtid).toEqual(`${ACTIVE_SERVER_UUID}:17`); + }); + + test('supports a bare transaction id', async () => { + const gtid = await getLatestActiveGtid([`${ACTIVE_SERVER_UUID}:42`], ACTIVE_SERVER_UUID); + + expect(gtid).toEqual(`${ACTIVE_SERVER_UUID}:42`); + }); + + test('returns the active server ZERO GTID when it is absent from the GTID sets', async () => { + await expect(getLatestActiveGtid([`${STALE_SERVER_UUID}:1-1000`], ACTIVE_SERVER_UUID)).resolves.toEqual( + `${ACTIVE_SERVER_UUID}:0` + ); + }); + }); + + describe('readExecutedGtid', () => { + test('reads binary log status on MySQL 8.4 and selects the active server GTID', async () => { + const { connection, query } = createConnection({ + version: '8.4.0', + executedGtidSet: `${STALE_SERVER_UUID}:1-1000,\n${ACTIVE_SERVER_UUID}:1-5:11-18` + }); + + const gtid = await readExecutedGtid(connection); + + expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:18`); + expect(gtid.position).toEqual({ filename: 'binlog.000042', offset: 1234 }); + expect(query).toHaveBeenCalledWith('SHOW BINARY LOG STATUS', []); + expect(query).not.toHaveBeenCalledWith('SHOW MASTER STATUS', []); + }); + + test('reads master status on MySQL versions before 8.4', async () => { + const { connection, query } = createConnection({ + version: '8.0.40', + executedGtidSet: `${ACTIVE_SERVER_UUID}:1-17` + }); + + const gtid = await readExecutedGtid(connection); + + expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:17`); + expect(query).toHaveBeenCalledWith('SHOW MASTER STATUS', []); + expect(query).not.toHaveBeenCalledWith('SHOW BINARY LOG STATUS', []); + }); + + test('returns the active server ZERO GTID when no transactions have executed', async () => { + const { connection } = createConnection({ + version: '8.4.0', + executedGtidSet: ' \n\t ' + }); + + const gtid = await readExecutedGtid(connection); + + expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:0`); + expect(gtid.position).toEqual({ filename: 'binlog.000042', offset: 1234 }); + expect(gtid.comparable).toEqual(`0000000000000000|${ACTIVE_SERVER_UUID}:0|binlog.000042|1234`); + }); + + test('uses the active server ZERO GTID at the current position when only historical UUIDs exist', async () => { + const { connection } = createConnection({ + version: '8.4.0', + executedGtidSet: `${STALE_SERVER_UUID}:1-1000` + }); + + const gtid = await readExecutedGtid(connection); + + expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:0`); + expect(gtid.position).toEqual({ filename: 'binlog.000042', offset: 1234 }); + expect(gtid.comparable).toEqual(`0000000000000000|${ACTIVE_SERVER_UUID}:0|binlog.000042|1234`); + }); + }); + + describe('isGtidPositionStillAvailable', () => { + const RESUME_GTID = new ReplicatedGTID({ + rawGtid: `${ACTIVE_SERVER_UUID}:17`, + position: { filename: 'binlog.000042', offset: 1234 } + }); + + test('returns true when the GTID is executed and its binlog coordinate is available', async () => { + const { connection, query } = createResumeCheckConnection({ + isExecuted: 1, + logFiles: [{ Log_name: 'binlog.000042', File_size: 2000 }] + }); + + await expect(isGtidPositionStillAvailable(connection, RESUME_GTID)).resolves.toBe(true); + expect(query).toHaveBeenCalledWith('SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed', [ + RESUME_GTID.raw + ]); + }); + + test('returns false when the GTID is absent after a source rewind', async () => { + const { connection } = createResumeCheckConnection({ + isExecuted: 0, + logFiles: [{ Log_name: 'binlog.000042', File_size: 2000 }] + }); + + await expect(isGtidPositionStillAvailable(connection, RESUME_GTID)).resolves.toBe(false); + }); + + test('validates the synthetic ZERO GTID using only its binlog coordinate', async () => { + const zeroGtid = new ReplicatedGTID({ + rawGtid: `${ACTIVE_SERVER_UUID}:0`, + position: { filename: 'binlog.000042', offset: 1234 } + }); + const { connection, query } = createResumeCheckConnection({ + isExecuted: 0, + logFiles: [{ Log_name: 'binlog.000042', File_size: 2000 }] + }); + + await expect(isGtidPositionStillAvailable(connection, zeroGtid)).resolves.toBe(true); + expect(query).not.toHaveBeenCalledWith('SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed', [ + zeroGtid.raw + ]); + }); + + test.each([ + [[], 'the binlog file is absent'], + [[{ Log_name: 'binlog.000042', File_size: 1000 }], 'the stored offset is past the end of the binlog'] + ])('returns false when %s (%s)', async (logFiles) => { + const { connection, query } = createResumeCheckConnection({ isExecuted: 1, logFiles }); + + await expect(isGtidPositionStillAvailable(connection, RESUME_GTID)).resolves.toBe(false); + expect(query).not.toHaveBeenCalledWith('SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed', [ + RESUME_GTID.raw + ]); + }); + }); + + function createConnection(options: { version: string; executedGtidSet: string }) { + return createMockMySQLConnection(async (sql) => { + switch (sql) { + case 'SELECT VERSION() as version': + return [[{ version: options.version }], []]; + case 'SHOW BINARY LOG STATUS': + case 'SHOW MASTER STATUS': + return [ + [ + { + File: 'binlog.000042', + Position: '1234', + Executed_Gtid_Set: options.executedGtidSet + } + ], + [] + ]; + case 'SELECT @@server_uuid AS server_uuid': + return [[{ server_uuid: ACTIVE_SERVER_UUID }], []]; + default: + throw new Error(`Unexpected query: ${sql}`); + } + }); + } + + function createResumeCheckConnection(options: { + isExecuted: number; + logFiles: { Log_name: string; File_size: number }[]; + }) { + return createMockMySQLConnection(async (sql) => { + switch (sql) { + case 'SHOW BINARY LOGS;': + return [options.logFiles, []]; + case 'SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed': + return [[{ is_executed: options.isExecuted }], []]; + default: + throw new Error(`Unexpected query: ${sql}`); + } + }); + } +}); diff --git a/modules/module-mysql/test/src/util.ts b/modules/module-mysql/test/src/util.ts index 23eb076bc..ee580a1d0 100644 --- a/modules/module-mysql/test/src/util.ts +++ b/modules/module-mysql/test/src/util.ts @@ -9,7 +9,7 @@ import * as mongo_storage from '@powersync/service-module-mongodb-storage'; import * as postgres_storage from '@powersync/service-module-postgres-storage'; import { TablePattern } from '@powersync/service-sync-rules'; import mysqlPromise from 'mysql2/promise'; -import { describe, TestOptions } from 'vitest'; +import { describe, TestOptions, vi } from 'vitest'; import { env } from './env.js'; export const TEST_URI = env.MYSQL_TEST_URI; @@ -28,6 +28,17 @@ export const INITIALIZED_POSTGRES_STORAGE_FACTORY = postgres_storage.test_utils. url: env.PG_STORAGE_TEST_URL }); +export function createMockMySQLConnection(queryHandler: (sql: string, params?: unknown[]) => Promise): { + connection: mysqlPromise.Connection; + query: ReturnType; +} { + const query = vi.fn(queryHandler); + return { + connection: { query } as unknown as mysqlPromise.Connection, + query + }; +} + export function describeWithStorage(options: TestOptions, fn: (factory: TestStorageConfig) => void) { describe.skipIf(!env.TEST_MONGO_STORAGE)(`mongodb storage`, options, function () { fn(INITIALIZED_MONGO_STORAGE_FACTORY); @@ -71,6 +82,13 @@ export async function getFromGTID(connectionManager: MySQLConnectionManager) { return fromGTID; } +export async function getActiveServerUuid(connectionManager: MySQLConnectionManager) { + const connection = await connectionManager.getConnection(); + const activeServerUuid = await common.readServerUuid(connection); + connection.release(); + return activeServerUuid; +} + export interface CreateBinlogListenerParams { connectionManager: MySQLConnectionManager; eventHandler: BinLogEventHandler; @@ -84,12 +102,15 @@ export async function createBinlogListener(params: CreateBinlogListenerParams): startGTID = await getFromGTID(connectionManager); } + const activeServerUuid = await getActiveServerUuid(connectionManager); + return new BinLogListener({ connectionManager: connectionManager, eventHandler: eventHandler, startGTID: startGTID!, sourceTables: sourceTables, - serverId: createRandomServerId(1) + serverId: createRandomServerId(1), + activeServerUuid: activeServerUuid }); } @@ -100,6 +121,7 @@ export class TestBinLogEventHandler implements BinLogEventHandler { commitCount = 0; schemaChanges: SchemaChange[] = []; lastKeepAlive: string | undefined; + lastCommitLsn: string | undefined; unpause: ((value: void | PromiseLike) => void) | undefined; private pausedPromise: Promise | undefined; @@ -127,6 +149,7 @@ export class TestBinLogEventHandler implements BinLogEventHandler { async onCommit(lsn: string) { this.commitCount++; + this.lastCommitLsn = lsn; } async onSchemaChange(change: SchemaChange) {