From bb68304e8e7f797ad977100166ce680c7f0f7e87 Mon Sep 17 00:00:00 2001 From: michaelbarnes Date: Wed, 1 Jul 2026 17:21:55 -0600 Subject: [PATCH 01/11] [MySQL] Fix checkpoints stalling on idle servers Heartbeat keepalives re-sent the LSN from the start of the last transaction, while checkpoints store the LSN from the end of the same transaction. On an idle server this blocked checkpoint creation ("Waiting before creating checkpoint" every ~30s) until the next transaction arrived. All commit paths (Xid, DDL auto-commit, non-transactional query) now advance the current GTID position to the commit position, so keepalive LSNs are never behind the last checkpoint LSN. The listener also no longer mutates the caller's startGTID position object. Co-Authored-By: Claude Fable 5 --- .changeset/mysql-idle-keepalive-lsn.md | 5 +++ .../src/replication/zongji/BinLogListener.ts | 37 +++++++++++-------- .../test/src/BinLogListener.test.ts | 18 +++++++++ modules/module-mysql/test/src/util.ts | 2 + 4 files changed, 46 insertions(+), 16 deletions(-) create mode 100644 .changeset/mysql-idle-keepalive-lsn.md diff --git a/.changeset/mysql-idle-keepalive-lsn.md b/.changeset/mysql-idle-keepalive-lsn.md new file mode 100644 index 000000000..48172cfe2 --- /dev/null +++ b/.changeset/mysql-idle-keepalive-lsn.md @@ -0,0 +1,5 @@ +--- +'@powersync/service-module-mysql': patch +--- + +Fix checkpoints stalling on idle MySQL servers. Heartbeat keepalives now report the LSN of the last committed transaction instead of the transaction start position, which previously blocked checkpoint creation ("Waiting before creating checkpoint" logged every ~30s) until the next transaction arrived. diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index 8b83ab00b..a23c528f1 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -113,7 +113,8 @@ export class BinLogListener { 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(); @@ -359,11 +360,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 +373,22 @@ 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({ + raw_gtid: this.currentGTID.raw, + // Copy the position: this.binLogPosition is mutated by subsequent events + position: { ...this.binLogPosition } + }); + return this.currentGTID.comparable; + } + private async processQueryEvent(event: BinLogQueryEvent): Promise { const { query, nextPosition } = event; @@ -398,11 +411,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 +428,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/util.ts b/modules/module-mysql/test/src/util.ts index 23eb076bc..77fa33348 100644 --- a/modules/module-mysql/test/src/util.ts +++ b/modules/module-mysql/test/src/util.ts @@ -100,6 +100,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 +128,7 @@ export class TestBinLogEventHandler implements BinLogEventHandler { async onCommit(lsn: string) { this.commitCount++; + this.lastCommitLsn = lsn; } async onSchemaChange(change: SchemaChange) { From 8e9420d195d2331d91c31ab16cb7c35fd40d38bb Mon Sep 17 00:00:00 2001 From: michaelbarnes Date: Wed, 1 Jul 2026 17:21:55 -0600 Subject: [PATCH 02/11] [MySQL] Fix GTID parsing for multi-server-UUID GTID sets ReplicatedGTID.comparable assumed a single server UUID in the raw GTID. A gtid_executed containing multiple server UUIDs (e.g. after a failover or restore) was mis-parsed into a NaN transaction id, producing LSNs like "0000000000000NaN|...". On servers with a low transaction count this permanently blocked checkpoint creation, and the corrupted LSN could not be recovered by a service restart. The comparable LSN now parses full GTID sets (multiple UUIDs joined with ",\n", multiple intervals per UUID) and uses the maximum transaction id across the set. Unparseable segments are skipped instead of poisoning the result. deserialize now validates the binlog offset instead of silently producing NaN. Co-Authored-By: Claude Fable 5 --- .changeset/mysql-gtid-set-parsing.md | 5 + .../module-mysql/src/common/ReplicatedGTID.ts | 52 +++++-- .../test/src/ReplicatedGTID.test.ts | 142 ++++++++++++++++++ 3 files changed, 186 insertions(+), 13 deletions(-) create mode 100644 .changeset/mysql-gtid-set-parsing.md create mode 100644 modules/module-mysql/test/src/ReplicatedGTID.test.ts diff --git a/.changeset/mysql-gtid-set-parsing.md b/.changeset/mysql-gtid-set-parsing.md new file mode 100644 index 000000000..3c15b104a --- /dev/null +++ b/.changeset/mysql-gtid-set-parsing.md @@ -0,0 +1,5 @@ +--- +'@powersync/service-module-mysql': patch +--- + +Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. diff --git a/modules/module-mysql/src/common/ReplicatedGTID.ts b/modules/module-mysql/src/common/ReplicatedGTID.ts index dc7713e91..82d2c7d34 100644 --- a/modules/module-mysql/src/common/ReplicatedGTID.ts +++ b/modules/module-mysql/src/common/ReplicatedGTID.ts @@ -37,15 +37,20 @@ export class ReplicatedGTID { private static deserialize(comparable: string): ReplicatedGTIDSpecification { const components = comparable.split('|'); - if (components.length < 3) { + if (components.length < 4) { throw new Error(`Invalid serialized GTID: ${comparable}`); } + const offset = parseInt(components[3], 10); + if (Number.isNaN(offset)) { + throw new Error(`Invalid BinLog offset in serialized GTID: ${comparable}`); + } + return { raw_gtid: components[1], position: { filename: components[2], - offset: parseInt(components[3]) + offset: offset } satisfies BinLogPosition }; } @@ -86,26 +91,47 @@ export class ReplicatedGTID { /** * Transforms a GTID into a comparable string format, ensuring lexicographical - * order aligns with the GTID's relative age. This assumes that all GTIDs - * have the same server ID. + * order aligns with the GTID's relative age. + * + * The raw GTID can be a full GTID set consisting of multiple comma-separated + * (optionally whitespace/newline padded) UUID sets, each of the form + * `server_uuid:interval[:interval...]` where an interval is `n` or `n-m`. + * The maximum transaction id across all UUID sets is used for ordering. + * + * Note: this assumes the currently writing server has the highest transaction + * counter in the set. If a stale server UUID in the set has a higher counter + * than the active server (e.g. after a restore to a new server), checkpoints + * can be delayed until the active server's counter catches up. * * @returns A comparable string in the format * `padded_end_transaction|raw_gtid|binlog_filename|binlog_position` */ get comparable(): string { const { raw, position } = this; - const [, transactionRanges] = this.raw.split(':'); - - // This means no transactions have been executed on the database yet - if (!transactionRanges) { - return ReplicatedGTID.ZERO.comparable; - } let maxTransactionId = 0; + let hasTransactions = false; + + for (const uuidSet of raw.split(',')) { + const [, ...intervals] = uuidSet.trim().split(':'); + 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)) { + hasTransactions = true; + maxTransactionId = Math.max(maxTransactionId, startId); + } + if (!Number.isNaN(endId)) { + hasTransactions = true; + maxTransactionId = Math.max(maxTransactionId, endId); + } + } + } - for (const range of transactionRanges.split(',')) { - const [start, end] = range.split('-'); - maxTransactionId = Math.max(maxTransactionId, parseInt(start, 10), parseInt(end || start, 10)); + // This means no transactions have been executed on the database yet + if (!hasTransactions) { + return ReplicatedGTID.ZERO.comparable; } const paddedTransactionId = maxTransactionId.toString().padStart(16, '0'); 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..add446747 --- /dev/null +++ b/modules/module-mysql/test/src/ReplicatedGTID.test.ts @@ -0,0 +1,142 @@ +import { ReplicatedGTID } from '@module/common/ReplicatedGTID.js'; +import { describe, expect, test } from 'vitest'; + +describe('ReplicatedGTID', () => { + const POSITION = { filename: 'binlog.000042', offset: 1234 }; + + describe('comparable', () => { + test('single UUID with a single range', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17', + position: POSITION + }); + expect(gtid.comparable).toEqual('0000000000000017|a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17|binlog.000042|1234'); + }); + + test('single UUID with a bare transaction id', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:5', + position: POSITION + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000005'); + }); + + test('single UUID with multiple intervals uses the maximum transaction id', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-5:11-18', + position: POSITION + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000018'); + }); + + test('multiple server UUIDs joined with a newline (gtid_executed format)', () => { + // SHOW MASTER STATUS returns multi-UUID GTID sets joined with ',\n' + const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; + const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); + expect(gtid.comparable).not.toContain('NaN'); + expect(gtid.comparable.split('|')[0]).toEqual('0000000002734181'); + }); + + test('multiple server UUIDs where the first UUID holds the maximum', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-100,b7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-3', + position: POSITION + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000100'); + }); + + test('multiple server UUIDs with multi-interval members', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-5:20-30,\nb7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-8', + position: POSITION + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000030'); + }); + + test('ZERO GTID format is stable', () => { + expect(ReplicatedGTID.ZERO.comparable).toEqual('0000000000000000|0:0||0'); + }); + + test('empty GTID set falls back to the ZERO GTID', () => { + const empty = new ReplicatedGTID({ raw_gtid: '', position: POSITION }); + expect(empty.comparable).toEqual(ReplicatedGTID.ZERO.comparable); + + const noRanges = new ReplicatedGTID({ raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004', position: POSITION }); + expect(noRanges.comparable).toEqual(ReplicatedGTID.ZERO.comparable); + }); + + test('unparseable segments are skipped and never produce NaN', () => { + const trailingGarbage = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17,\ngarbage-no-colon', + position: POSITION + }); + expect(trailingGarbage.comparable).not.toContain('NaN'); + expect(trailingGarbage.comparable.split('|')[0]).toEqual('0000000000000017'); + + const garbageInterval = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:abc-def:1-9', + position: POSITION + }); + expect(garbageInterval.comparable).not.toContain('NaN'); + expect(garbageInterval.comparable.split('|')[0]).toEqual('0000000000000009'); + }); + }); + + describe('serialization', () => { + test('round-trips a multi-UUID GTID set', () => { + const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; + const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); + + const deserialized = ReplicatedGTID.fromSerialized(gtid.comparable); + expect(deserialized.raw).toEqual(raw); + expect(deserialized.position).toEqual(POSITION); + expect(deserialized.comparable).toEqual(gtid.comparable); + }); + + test('throws on malformed serialized GTIDs', () => { + expect(() => ReplicatedGTID.fromSerialized('abc')).toThrow(); + // Missing binlog offset + expect(() => ReplicatedGTID.fromSerialized('0000000000000001|uuid:1|binlog.000001')).toThrow(); + expect(() => ReplicatedGTID.fromSerialized('0000000000000001|uuid:1|binlog.000001|notanumber')).toThrow(); + }); + }); + + describe('LSN ordering', () => { + test('LSNs for the same transaction order by binlog offset', () => { + // Note: the binlog offset is not zero-padded, so lexicographic ordering only holds for + // offsets with the same number of digits. This is sufficient for the checkpoint gate since + // heartbeat keepalive LSNs are byte-identical to the last commit LSN, but is documented + // here as a known limitation of the format (which cannot change for compatibility with + // LSNs already persisted in bucket storage). + const raw = 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:18'; + const transactionStart = new ReplicatedGTID({ + raw_gtid: raw, + position: { filename: 'binlog.000042', offset: 157 } + }); + const transactionEnd = new ReplicatedGTID({ + raw_gtid: raw, + position: { filename: 'binlog.000042', offset: 300 } + }); + expect(transactionStart.comparable < transactionEnd.comparable).toBeTruthy(); + }); + + test('correct LSNs order against legacy NaN-corrupted LSNs as documented', () => { + // LSNs produced by the previous multi-UUID parsing bug contain a literal 'NaN' padded transaction id. + const legacyPoisoned = '0000000000000NaN|2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17|binlog.000042|1234'; + + // Instances with transaction ids >= 1000 sort above the corrupted LSN and self-heal + const highTransaction = new ReplicatedGTID({ + raw_gtid: '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-39489900', + position: POSITION + }); + expect(highTransaction.comparable > legacyPoisoned).toBeTruthy(); + + // Instances with transaction ids < 1000 still sort below it ('N' > any digit) and require a resync + const lowTransaction = new ReplicatedGTID({ + raw_gtid: '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-999', + position: POSITION + }); + expect(lowTransaction.comparable < legacyPoisoned).toBeTruthy(); + }); + }); +}); From ab7af187641736d2a1ddcead45991c5f882f1ee2 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 6 Aug 2026 09:56:07 +0200 Subject: [PATCH 03/11] Warn when multiple server UUIDs are detected in the GTID set or on the binlog --- .changeset/mysql-gtid-set-parsing.md | 2 +- .../module-mysql/src/common/ReplicatedGTID.ts | 20 +++++++++++++++ .../src/replication/BinLogStream.ts | 9 +++++++ .../src/replication/zongji/BinLogListener.ts | 21 ++++++++++++++++ .../test/src/ReplicatedGTID.test.ts | 25 +++++++++++++++++++ 5 files changed, 76 insertions(+), 1 deletion(-) diff --git a/.changeset/mysql-gtid-set-parsing.md b/.changeset/mysql-gtid-set-parsing.md index 3c15b104a..2a01c5b74 100644 --- a/.changeset/mysql-gtid-set-parsing.md +++ b/.changeset/mysql-gtid-set-parsing.md @@ -2,4 +2,4 @@ '@powersync/service-module-mysql': patch --- -Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. +Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. A warning is now logged when multiple server UUIDs are detected in the executed GTID set or on the binlog, since GTID-based LSN ordering across server UUIDs is not reliable. diff --git a/modules/module-mysql/src/common/ReplicatedGTID.ts b/modules/module-mysql/src/common/ReplicatedGTID.ts index 82d2c7d34..119d5008a 100644 --- a/modules/module-mysql/src/common/ReplicatedGTID.ts +++ b/modules/module-mysql/src/common/ReplicatedGTID.ts @@ -85,10 +85,30 @@ export class ReplicatedGTID { return this.options.raw_gtid; } + /** + * The server UUID of a single-GTID value. For a raw value holding a full GTID set this is only the first + * UUID; use {@link serverIds} for all of them. + */ get serverId() { return this.options.raw_gtid.split(':')[0]; } + /** + * All distinct server UUIDs in the raw GTID set. More than one means the set contains transactions from + * multiple servers, which GTID-based LSN ordering does not reliably support. + */ + get serverIds(): string[] { + const ids = new Set(); + for (const uuidSet of this.options.raw_gtid.split(',')) { + const trimmed = uuidSet.trim(); + const separator = trimmed.indexOf(':'); + if (separator > 0) { + ids.add(trimmed.slice(0, separator)); + } + } + return [...ids]; + } + /** * Transforms a GTID into a comparable string format, ensuring lexicographical * order aligns with the GTID's relative age. diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index 3fed923ce..c5a29c68c 100644 --- a/modules/module-mysql/src/replication/BinLogStream.ts +++ b/modules/module-mysql/src/replication/BinLogStream.ts @@ -418,6 +418,15 @@ export class BinLogStream { : await common.readExecutedGtid(connection); connection.release(); + const gtidServerIds = fromGTID.serverIds; + if (gtidServerIds.length > 1) { + this.logger.warn( + `The executed GTID set contains multiple server UUIDs: ${gtidServerIds.join(', ')}. ` + + `GTID-based LSN ordering is only reliable for transactions from a single server. ` + + `Checkpoints may stall or be delayed if the active server is not the one with the highest transaction count.` + ); + } + if (!this.stopped) { await this.storage.startBatch( { zeroLSN: common.ReplicatedGTID.ZERO.comparable, defaultSchema: this.defaultSchema, storeCurrentData: false }, diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index a23c528f1..cd9543be3 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -101,6 +101,8 @@ export class BinLogListener { // Flag to indicate if are currently in a transaction that involves multiple row mutation events. private isTransactionOpen = false; + // Server UUIDs seen in GTID events, used to warn when the binlog carries transactions from multiple servers. + private seenGtidServerIds = new Set(); zongji: ZongJi; processingQueue: async.QueueObject; @@ -304,6 +306,7 @@ export class BinLogListener { } }); this.binLogPosition.offset = evt.nextPosition; + this.trackGtidServerId(this.currentGTID.serverId); await this.eventHandler.onTransactionStart({ timestamp: new Date(evt.timestamp) }); this.logger.info(`Processed GTID event: ${this.currentGTID.comparable}`); break; @@ -389,6 +392,24 @@ export class BinLogListener { return this.currentGTID.comparable; } + /** + * Warns when transactions from more than one server UUID appear on the binlog. GTID-based LSN ordering is + * only reliable for a single server UUID, so mixed transactions can stall checkpoints or apply them out of + * order. Warns once per newly seen UUID rather than per transaction. + */ + private trackGtidServerId(serverId: string): void { + if (this.seenGtidServerIds.has(serverId)) { + return; + } + this.seenGtidServerIds.add(serverId); + if (this.seenGtidServerIds.size > 1) { + this.logger.warn( + `Transactions from multiple MySQL server UUIDs detected on the binlog: ${[...this.seenGtidServerIds].join(', ')}. ` + + `GTID-based LSN ordering across different server UUIDs is not supported, and checkpoints may stall or be inconsistent.` + ); + } + } + private async processQueryEvent(event: BinLogQueryEvent): Promise { const { query, nextPosition } = event; diff --git a/modules/module-mysql/test/src/ReplicatedGTID.test.ts b/modules/module-mysql/test/src/ReplicatedGTID.test.ts index add446747..af2f8631d 100644 --- a/modules/module-mysql/test/src/ReplicatedGTID.test.ts +++ b/modules/module-mysql/test/src/ReplicatedGTID.test.ts @@ -139,4 +139,29 @@ describe('ReplicatedGTID', () => { expect(lowTransaction.comparable < legacyPoisoned).toBeTruthy(); }); }); + + describe('serverIds', () => { + test('single UUID', () => { + const gtid = new ReplicatedGTID({ raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17', position: POSITION }); + expect(gtid.serverIds).toEqual(['a7d0ff7b-0c0e-11f0-8b38-566fbaa00004']); + }); + + test('multiple UUIDs in a newline-joined executed set', () => { + const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; + const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); + expect(gtid.serverIds).toEqual(['2e35321d-0c0e-11f0-8b38-566fbaa00004', '314306f3-ff7b-11ef-a0e0-566fbaa00002']); + }); + + test('ignores segments without intervals', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17,\ngarbage-no-colon', + position: POSITION + }); + expect(gtid.serverIds).toEqual(['a7d0ff7b-0c0e-11f0-8b38-566fbaa00004']); + }); + + test('empty set has no server ids', () => { + expect(new ReplicatedGTID({ raw_gtid: '', position: POSITION }).serverIds).toEqual([]); + }); + }); }); From e773be77db99e07768d3ef20b54023e353d9c902 Mon Sep 17 00:00:00 2001 From: bean1352 Date: Thu, 6 Aug 2026 11:40:16 +0200 Subject: [PATCH 04/11] Order LSNs by the connected server's GTID counter --- .changeset/mysql-gtid-set-parsing.md | 2 +- .../module-mysql/src/common/ReplicatedGTID.ts | 68 ++++++++++--------- .../src/common/read-executed-gtid.ts | 15 +++- .../src/replication/BinLogStream.ts | 28 ++++---- .../src/replication/zongji/BinLogListener.ts | 60 +++++++++------- .../test/src/ReplicatedGTID.test.ts | 53 +++++++++++---- 6 files changed, 136 insertions(+), 90 deletions(-) diff --git a/.changeset/mysql-gtid-set-parsing.md b/.changeset/mysql-gtid-set-parsing.md index 2a01c5b74..0e9fbbabc 100644 --- a/.changeset/mysql-gtid-set-parsing.md +++ b/.changeset/mysql-gtid-set-parsing.md @@ -2,4 +2,4 @@ '@powersync/service-module-mysql': patch --- -Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. A warning is now logged when multiple server UUIDs are detected in the executed GTID set or on the binlog, since GTID-based LSN ordering across server UUIDs is not reliable. +Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. LSN ordering now follows the connected server's own transaction counter (`@@server_uuid`) instead of the highest counter in the set, so a stale server UUID with a higher counter can no longer hang checkpoints. A warning is logged when a transaction from a different server UUID appears on the binlog (e.g. when connected to a replica), since those are not reliably ordered yet. diff --git a/modules/module-mysql/src/common/ReplicatedGTID.ts b/modules/module-mysql/src/common/ReplicatedGTID.ts index 119d5008a..950cb4e4b 100644 --- a/modules/module-mysql/src/common/ReplicatedGTID.ts +++ b/modules/module-mysql/src/common/ReplicatedGTID.ts @@ -13,6 +13,11 @@ export type ReplicatedGTIDSpecification = { * The (end) position in a BinLog file where this transaction has been replicated in. */ position: BinLogPosition; + /** + * The `@@server_uuid` of the connected server. When set and the raw GTID set contains multiple server + * UUIDs, LSN ordering uses this server's transaction counter rather than the highest counter in the set. + */ + serverUuid?: string; }; export type BinLogGTIDFormat = { @@ -31,8 +36,8 @@ export type BinLogGTIDEvent = { * and position where this GTID could be located. */ export class ReplicatedGTID { - static fromSerialized(comparable: string): ReplicatedGTID { - return new ReplicatedGTID(ReplicatedGTID.deserialize(comparable)); + static fromSerialized(comparable: string, serverUuid?: string): ReplicatedGTID { + return new ReplicatedGTID({ ...ReplicatedGTID.deserialize(comparable), serverUuid }); } private static deserialize(comparable: string): ReplicatedGTIDSpecification { @@ -55,12 +60,13 @@ export class ReplicatedGTID { }; } - static fromBinLogEvent(event: BinLogGTIDEvent) { + static fromBinLogEvent(event: BinLogGTIDEvent, serverUuid?: string) { const { raw_gtid, position } = event; const stringGTID = `${uuid.stringify(raw_gtid.server_id)}:${raw_gtid.transaction_range}`; return new ReplicatedGTID({ raw_gtid: stringGTID, - position + position, + serverUuid }); } @@ -87,26 +93,15 @@ export class ReplicatedGTID { /** * The server UUID of a single-GTID value. For a raw value holding a full GTID set this is only the first - * UUID; use {@link serverIds} for all of them. + * UUID. */ get serverId() { return this.options.raw_gtid.split(':')[0]; } - /** - * All distinct server UUIDs in the raw GTID set. More than one means the set contains transactions from - * multiple servers, which GTID-based LSN ordering does not reliably support. - */ - get serverIds(): string[] { - const ids = new Set(); - for (const uuidSet of this.options.raw_gtid.split(',')) { - const trimmed = uuidSet.trim(); - const separator = trimmed.indexOf(':'); - if (separator > 0) { - ids.add(trimmed.slice(0, separator)); - } - } - return [...ids]; + /** The connected server's `@@server_uuid`, when known. See {@link ReplicatedGTIDSpecification.serverUuid}. */ + get activeServerUuid(): string | undefined { + return this.options.serverUuid; } /** @@ -116,45 +111,52 @@ export class ReplicatedGTID { * The raw GTID can be a full GTID set consisting of multiple comma-separated * (optionally whitespace/newline padded) UUID sets, each of the form * `server_uuid:interval[:interval...]` where an interval is `n` or `n-m`. - * The maximum transaction id across all UUID sets is used for ordering. * - * Note: this assumes the currently writing server has the highest transaction - * counter in the set. If a stale server UUID in the set has a higher counter - * than the active server (e.g. after a restore to a new server), checkpoints - * can be delayed until the active server's counter catches up. + * When the connected server's UUID is known (see {@link ReplicatedGTIDSpecification.serverUuid}) and has + * transactions in the set, ordering uses that server's transaction counter: it is the counter that grows + * with new transactions. Otherwise the maximum counter across all UUID sets is used, which can pin the + * LSN to a stale server's higher counter (e.g. after a restore to a new server) and delay checkpoints + * until the active counter catches up. * * @returns A comparable string in the format * `padded_end_transaction|raw_gtid|binlog_filename|binlog_position` */ get comparable(): string { const { raw, position } = this; + const activeUuid = this.options.serverUuid; - let maxTransactionId = 0; - let hasTransactions = false; + let maxTransactionId: number | null = null; + let activeTransactionId: number | null = null; for (const uuidSet of raw.split(',')) { - const [, ...intervals] = uuidSet.trim().split(':'); + const [serverUuid, ...intervals] = uuidSet.trim().split(':'); + let uuidSetMax: 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)) { - hasTransactions = true; - maxTransactionId = Math.max(maxTransactionId, startId); + uuidSetMax = Math.max(uuidSetMax ?? 0, startId); } if (!Number.isNaN(endId)) { - hasTransactions = true; - maxTransactionId = Math.max(maxTransactionId, endId); + uuidSetMax = Math.max(uuidSetMax ?? 0, endId); } } + if (uuidSetMax == null) { + continue; + } + maxTransactionId = Math.max(maxTransactionId ?? 0, uuidSetMax); + if (serverUuid === activeUuid) { + activeTransactionId = Math.max(activeTransactionId ?? 0, uuidSetMax); + } } // This means no transactions have been executed on the database yet - if (!hasTransactions) { + if (maxTransactionId == null) { return ReplicatedGTID.ZERO.comparable; } - const paddedTransactionId = maxTransactionId.toString().padStart(16, '0'); + const paddedTransactionId = (activeTransactionId ?? maxTransactionId).toString().padStart(16, '0'); return [paddedTransactionId, raw, position.filename, position.offset].join('|'); } diff --git a/modules/module-mysql/src/common/read-executed-gtid.ts b/modules/module-mysql/src/common/read-executed-gtid.ts index 9f60c3362..83da204d3 100644 --- a/modules/module-mysql/src/common/read-executed-gtid.ts +++ b/modules/module-mysql/src/common/read-executed-gtid.ts @@ -2,6 +2,18 @@ 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 connected server, used to pick its transaction counter out of a GTID set + * for LSN ordering. + */ +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 */ @@ -31,7 +43,8 @@ export async function readExecutedGtid(connection: mysqlPromise.Connection): Pro return new ReplicatedGTID({ // The head always points to the next position to start replication from position, - raw_gtid: binlogStatus.Executed_Gtid_Set + raw_gtid: binlogStatus.Executed_Gtid_Set, + serverUuid: await readServerUuid(connection) }); } diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index c5a29c68c..4d3522ca8 100644 --- a/modules/module-mysql/src/replication/BinLogStream.ts +++ b/modules/module-mysql/src/replication/BinLogStream.ts @@ -409,22 +409,18 @@ export class BinLogStream { 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}`); - } - const fromGTID = resume_lsn - ? common.ReplicatedGTID.fromSerialized(resume_lsn) - : await common.readExecutedGtid(connection); - connection.release(); - - const gtidServerIds = fromGTID.serverIds; - if (gtidServerIds.length > 1) { - this.logger.warn( - `The executed GTID set contains multiple server UUIDs: ${gtidServerIds.join(', ')}. ` + - `GTID-based LSN ordering is only reliable for transactions from a single server. ` + - `Checkpoints may stall or be delayed if the active server is not the one with the highest transaction count.` - ); + 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}`); + } + // The server uuid picks the connected server's transaction counter out of GTID sets for LSN ordering. + fromGTID = resume_lsn + ? common.ReplicatedGTID.fromSerialized(resume_lsn, await common.readServerUuid(connection)) + : await common.readExecutedGtid(connection); + } finally { + connection.release(); } if (!this.stopped) { diff --git a/modules/module-mysql/src/replication/zongji/BinLogListener.ts b/modules/module-mysql/src/replication/zongji/BinLogListener.ts index cd9543be3..8797370c4 100644 --- a/modules/module-mysql/src/replication/zongji/BinLogListener.ts +++ b/modules/module-mysql/src/replication/zongji/BinLogListener.ts @@ -101,8 +101,10 @@ export class BinLogListener { // Flag to indicate if are currently in a transaction that involves multiple row mutation events. private isTransactionOpen = false; - // Server UUIDs seen in GTID events, used to warn when the binlog carries transactions from multiple servers. - private seenGtidServerIds = new Set(); + // The @@server_uuid of the connected server, used to order LSNs and to detect foreign transactions. + private currentServerUuid: string | undefined; + // Foreign server UUIDs already warned about, so each one is logged once rather than per transaction. + private warnedForeignServerIds = new Set(); zongji: ZongJi; processingQueue: async.QueueObject; @@ -118,6 +120,7 @@ export class BinLogListener { // 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.currentServerUuid = options.startGTID.activeServerUuid; this.sqlParser = new Parser(); this.processingQueue = this.createProcessingQueue(); this.zongji = this.createZongjiListener(); @@ -295,18 +298,21 @@ 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 + this.currentGTID = common.ReplicatedGTID.fromBinLogEvent( + { + raw_gtid: { + server_id: evt.serverId, + transaction_range: evt.transactionRange + }, + position: { + filename: this.binLogPosition.filename, + offset: evt.nextPosition + } }, - position: { - filename: this.binLogPosition.filename, - offset: evt.nextPosition - } - }); + this.currentServerUuid + ); this.binLogPosition.offset = evt.nextPosition; - this.trackGtidServerId(this.currentGTID.serverId); + this.warnOnForeignServerUuid(this.currentGTID.serverId); await this.eventHandler.onTransactionStart({ timestamp: new Date(evt.timestamp) }); this.logger.info(`Processed GTID event: ${this.currentGTID.comparable}`); break; @@ -387,27 +393,31 @@ export class BinLogListener { this.currentGTID = new common.ReplicatedGTID({ raw_gtid: this.currentGTID.raw, // Copy the position: this.binLogPosition is mutated by subsequent events - position: { ...this.binLogPosition } + position: { ...this.binLogPosition }, + serverUuid: this.currentServerUuid }); return this.currentGTID.comparable; } /** - * Warns when transactions from more than one server UUID appear on the binlog. GTID-based LSN ordering is - * only reliable for a single server UUID, so mixed transactions can stall checkpoints or apply them out of - * order. Warns once per newly seen UUID rather than per transaction. + * Warns when a transaction on the binlog originates from a server other than the connected one, such as + * when the connected server is a replica. LSN ordering follows the connected server's transaction + * counter, so transactions from other server UUIDs are not reliably ordered and can stall checkpoints. + * Warns once per foreign UUID rather than per transaction. */ - private trackGtidServerId(serverId: string): void { - if (this.seenGtidServerIds.has(serverId)) { + private warnOnForeignServerUuid(transactionServerUuid: string): void { + if ( + this.currentServerUuid == null || + transactionServerUuid === this.currentServerUuid || + this.warnedForeignServerIds.has(transactionServerUuid) + ) { return; } - this.seenGtidServerIds.add(serverId); - if (this.seenGtidServerIds.size > 1) { - this.logger.warn( - `Transactions from multiple MySQL server UUIDs detected on the binlog: ${[...this.seenGtidServerIds].join(', ')}. ` + - `GTID-based LSN ordering across different server UUIDs is not supported, and checkpoints may stall or be inconsistent.` - ); - } + this.warnedForeignServerIds.add(transactionServerUuid); + this.logger.warn( + `Detected a transaction from a different MySQL server UUID on the binlog: ${transactionServerUuid} (connected server: ${this.currentServerUuid}). ` + + `LSN ordering follows the connected server's transaction counter, so transactions from other servers are not reliably ordered and checkpoints may stall.` + ); } private async processQueryEvent(event: BinLogQueryEvent): Promise { diff --git a/modules/module-mysql/test/src/ReplicatedGTID.test.ts b/modules/module-mysql/test/src/ReplicatedGTID.test.ts index af2f8631d..d4dffaaca 100644 --- a/modules/module-mysql/test/src/ReplicatedGTID.test.ts +++ b/modules/module-mysql/test/src/ReplicatedGTID.test.ts @@ -140,28 +140,53 @@ describe('ReplicatedGTID', () => { }); }); - describe('serverIds', () => { - test('single UUID', () => { - const gtid = new ReplicatedGTID({ raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17', position: POSITION }); - expect(gtid.serverIds).toEqual(['a7d0ff7b-0c0e-11f0-8b38-566fbaa00004']); + describe('connected server UUID selection', () => { + const ACTIVE = '2e35321d-0c0e-11f0-8b38-566fbaa00004'; + const STALE = '314306f3-ff7b-11ef-a0e0-566fbaa00002'; + + test('uses the connected server counter even when a stale UUID holds a higher one', () => { + // A restore from another server leaves the old UUID with a high counter. Ordering by the set-wide + // maximum would pin the LSN there and hang checkpoints until the active counter catches up. + const gtid = new ReplicatedGTID({ + raw_gtid: `${STALE}:1-42350493,\n${ACTIVE}:1-17`, + position: POSITION, + serverUuid: ACTIVE + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000017'); }); - test('multiple UUIDs in a newline-joined executed set', () => { - const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; - const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); - expect(gtid.serverIds).toEqual(['2e35321d-0c0e-11f0-8b38-566fbaa00004', '314306f3-ff7b-11ef-a0e0-566fbaa00002']); + test('matches the set-wide maximum when the connected server holds it', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: `${ACTIVE}:1-42350493,\n${STALE}:1-2734181`, + position: POSITION, + serverUuid: ACTIVE + }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000042350493'); }); - test('ignores segments without intervals', () => { + test('falls back to the set-wide maximum when the connected server has no transactions in the set', () => { const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17,\ngarbage-no-colon', - position: POSITION + raw_gtid: `${STALE}:1-2734181`, + position: POSITION, + serverUuid: ACTIVE }); - expect(gtid.serverIds).toEqual(['a7d0ff7b-0c0e-11f0-8b38-566fbaa00004']); + expect(gtid.comparable.split('|')[0]).toEqual('0000000002734181'); + }); + + test('single-GTID values from the connected server are unaffected', () => { + const gtid = new ReplicatedGTID({ raw_gtid: `${ACTIVE}:18`, position: POSITION, serverUuid: ACTIVE }); + expect(gtid.comparable.split('|')[0]).toEqual('0000000000000018'); }); - test('empty set has no server ids', () => { - expect(new ReplicatedGTID({ raw_gtid: '', position: POSITION }).serverIds).toEqual([]); + test('fromSerialized produces the same LSN when the same server UUID is provided', () => { + const gtid = new ReplicatedGTID({ + raw_gtid: `${STALE}:1-42350493,\n${ACTIVE}:1-17`, + position: POSITION, + serverUuid: ACTIVE + }); + const deserialized = ReplicatedGTID.fromSerialized(gtid.comparable, ACTIVE); + expect(deserialized.comparable).toEqual(gtid.comparable); + expect(deserialized.activeServerUuid).toEqual(ACTIVE); }); }); }); From b6ac7207bea08b4a1d3c703439750433675da048 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 12:44:38 +0200 Subject: [PATCH 05/11] Refactored GTID parsing and serialization logic in `ReplicatedGTID` for consistency and added validation for single GTID constraints --- .../src/api/MySQLRouteAPIAdapter.ts | 8 +- .../module-mysql/src/common/ReplicatedGTID.ts | 141 +++++------ .../src/common/read-executed-gtid.ts | 67 +++++- .../test/src/ReplicatedGTID.test.ts | 218 +++++++----------- .../test/src/read-executed-gtid.test.ts | 158 +++++++++++++ modules/module-mysql/test/src/util.ts | 25 +- 6 files changed, 388 insertions(+), 229 deletions(-) create mode 100644 modules/module-mysql/test/src/read-executed-gtid.test.ts 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 950cb4e4b..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,25 +9,24 @@ export type BinLogPosition = { }; export type ReplicatedGTIDSpecification = { - raw_gtid: string; /** - * The (end) position in a BinLog file where this transaction has been replicated in. + * 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. */ - position: BinLogPosition; + rawGtid: string; /** - * The `@@server_uuid` of the connected server. When set and the raw GTID set contains multiple server - * UUIDs, LSN ordering uses this server's transaction counter rather than the highest counter in the set. + * The (end) position in a BinLog file where this transaction has been replicated in. */ - serverUuid?: string; + position: BinLogPosition; }; export type BinLogGTIDFormat = { - server_id: Buffer; - transaction_range: number; + serverUuid: Buffer; + transactionId: number; }; export type BinLogGTIDEvent = { - raw_gtid: BinLogGTIDFormat; + rawGtid: BinLogGTIDFormat; position: BinLogPosition; }; @@ -36,23 +36,31 @@ export type BinLogGTIDEvent = { * and position where this GTID could be located. */ export class ReplicatedGTID { - static fromSerialized(comparable: string, serverUuid?: string): ReplicatedGTID { - return new ReplicatedGTID({ ...ReplicatedGTID.deserialize(comparable), serverUuid }); + 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 < 4) { - throw new Error(`Invalid serialized GTID: ${comparable}`); + throw new ReplicationAssertionError(`Invalid serialized GTID: ${comparable}`); } const offset = parseInt(components[3], 10); if (Number.isNaN(offset)) { - throw new Error(`Invalid BinLog offset in serialized GTID: ${comparable}`); + throw new ReplicationAssertionError(`Invalid BinLog offset in serialized GTID: ${comparable}`); } return { - raw_gtid: components[1], + rawGtid: components[1], position: { filename: components[2], offset: offset @@ -60,22 +68,24 @@ export class ReplicatedGTID { }; } - static fromBinLogEvent(event: BinLogGTIDEvent, serverUuid?: string) { - const { raw_gtid, position } = event; - const stringGTID = `${uuid.stringify(raw_gtid.server_id)}:${raw_gtid.transaction_range}`; + static fromBinLogEvent(event: BinLogGTIDEvent) { + const { rawGtid, position } = event; + const stringGTID = `${uuid.stringify(rawGtid.serverUuid)}:${rawGtid.transactionId}`; return new ReplicatedGTID({ - raw_gtid: stringGTID, - position, - serverUuid + rawGtid: stringGTID, + position }); } /** * 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 @@ -85,78 +95,32 @@ 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; } /** - * The server UUID of a single-GTID value. For a raw value holding a full GTID set this is only the first - * UUID. + * The server UUID of the server this transaction originated from */ - get serverId() { - return this.options.raw_gtid.split(':')[0]; - } - - /** The connected server's `@@server_uuid`, when known. See {@link ReplicatedGTIDSpecification.serverUuid}. */ - get activeServerUuid(): string | undefined { - return this.options.serverUuid; + get serverUuid() { + return this.options.rawGtid.split(':')[0]; } /** * Transforms a GTID into a comparable string format, ensuring lexicographical - * order aligns with the GTID's relative age. - * - * The raw GTID can be a full GTID set consisting of multiple comma-separated - * (optionally whitespace/newline padded) UUID sets, each of the form - * `server_uuid:interval[:interval...]` where an interval is `n` or `n-m`. - * - * When the connected server's UUID is known (see {@link ReplicatedGTIDSpecification.serverUuid}) and has - * transactions in the set, ordering uses that server's transaction counter: it is the counter that grows - * with new transactions. Otherwise the maximum counter across all UUID sets is used, which can pin the - * LSN to a stale server's higher counter (e.g. after a restore to a new server) and delay checkpoints - * until the active counter catches up. + * order aligns with the GTID's relative age. This assumes that all GTIDs + * have the same server ID. * * @returns A comparable string in the format * `padded_end_transaction|raw_gtid|binlog_filename|binlog_position` */ get comparable(): string { const { raw, position } = this; - const activeUuid = this.options.serverUuid; - - let maxTransactionId: number | null = null; - let activeTransactionId: number | null = null; - - for (const uuidSet of raw.split(',')) { - const [serverUuid, ...intervals] = uuidSet.trim().split(':'); - let uuidSetMax: 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)) { - uuidSetMax = Math.max(uuidSetMax ?? 0, startId); - } - if (!Number.isNaN(endId)) { - uuidSetMax = Math.max(uuidSetMax ?? 0, endId); - } - } - if (uuidSetMax == null) { - continue; - } - maxTransactionId = Math.max(maxTransactionId ?? 0, uuidSetMax); - if (serverUuid === activeUuid) { - activeTransactionId = Math.max(activeTransactionId ?? 0, uuidSetMax); - } - } - - // This means no transactions have been executed on the database yet - if (maxTransactionId == null) { - return ReplicatedGTID.ZERO.comparable; - } + const [, transactionId] = this.raw.split(':'); - const paddedTransactionId = (activeTransactionId ?? maxTransactionId).toString().padStart(16, '0'); + const paddedTransactionId = transactionId.toString().padStart(16, '0'); return [paddedTransactionId, raw, position.filename, position.offset].join('|'); } @@ -209,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/read-executed-gtid.ts b/modules/module-mysql/src/common/read-executed-gtid.ts index 83da204d3..ce9889d65 100644 --- a/modules/module-mysql/src/common/read-executed-gtid.ts +++ b/modules/module-mysql/src/common/read-executed-gtid.ts @@ -1,10 +1,10 @@ +import { ReplicationAssertionError } from '@powersync/lib-services-framework'; 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 connected server, used to pick its transaction counter out of a GTID set - * for LSN ordering. + * Gets the `@@server_uuid` of the current connected server */ export async function readServerUuid(connection: mysqlPromise.Connection): Promise { const [[result]] = await mysql_utils.retriedQuery({ @@ -40,22 +40,71 @@ 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 + return ReplicatedGTID.ZERO(activeServerUuid); + } + + 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, - serverUuid: await readServerUuid(connection) + 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; + } + } + + throw new ReplicationAssertionError( + `No GTID set found matching Active server UUID: ${activeServerUuid} in GTID sets: ${gtidSets.join(', ')}` + ); +} + +/** + * 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; + } + + 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/test/src/ReplicatedGTID.test.ts b/modules/module-mysql/test/src/ReplicatedGTID.test.ts index d4dffaaca..f182b3229 100644 --- a/modules/module-mysql/test/src/ReplicatedGTID.test.ts +++ b/modules/module-mysql/test/src/ReplicatedGTID.test.ts @@ -1,192 +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('comparable', () => { - test('single UUID with a single range', () => { + describe('single GTID', () => { + test('exposes its raw value, server UUID, and binlog position', () => { const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17', + rawGtid: `${SERVER_UUID}:5`, position: POSITION }); - expect(gtid.comparable).toEqual('0000000000000017|a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17|binlog.000042|1234'); - }); - test('single UUID with a bare transaction id', () => { - const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:5', - position: POSITION - }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000005'); + expect(gtid.raw).toEqual(`${SERVER_UUID}:5`); + expect(gtid.serverUuid).toEqual(SERVER_UUID); + expect(gtid.position).toEqual(POSITION); }); - test('single UUID with multiple intervals uses the maximum transaction id', () => { + test('formats a comparable LSN using the transaction id', () => { const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-5:11-18', + rawGtid: `${SERVER_UUID}:17`, position: POSITION }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000018'); - }); - - test('multiple server UUIDs joined with a newline (gtid_executed format)', () => { - // SHOW MASTER STATUS returns multi-UUID GTID sets joined with ',\n' - const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; - const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); - expect(gtid.comparable).not.toContain('NaN'); - expect(gtid.comparable.split('|')[0]).toEqual('0000000002734181'); - }); - test('multiple server UUIDs where the first UUID holds the maximum', () => { - const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-100,b7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-3', - position: POSITION - }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000100'); + expect(gtid.comparable).toEqual(`0000000000000017|${SERVER_UUID}:17|binlog.000042|1234`); + expect(gtid.toString()).toEqual(gtid.comparable); }); - test('multiple server UUIDs with multi-interval members', () => { + test('normalizes surrounding whitespace', () => { const gtid = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-5:20-30,\nb7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-8', + rawGtid: ` \n\t${SERVER_UUID}:17 \r\n`, position: POSITION }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000030'); - }); - test('ZERO GTID format is stable', () => { - expect(ReplicatedGTID.ZERO.comparable).toEqual('0000000000000000|0:0||0'); + 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('empty GTID set falls back to the ZERO GTID', () => { - const empty = new ReplicatedGTID({ raw_gtid: '', position: POSITION }); - expect(empty.comparable).toEqual(ReplicatedGTID.ZERO.comparable); - - const noRanges = new ReplicatedGTID({ raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004', position: POSITION }); - expect(noRanges.comparable).toEqual(ReplicatedGTID.ZERO.comparable); + 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`); }); + }); - test('unparseable segments are skipped and never produce NaN', () => { - const trailingGarbage = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:1-17,\ngarbage-no-colon', - position: POSITION - }); - expect(trailingGarbage.comparable).not.toContain('NaN'); - expect(trailingGarbage.comparable.split('|')[0]).toEqual('0000000000000017'); - - const garbageInterval = new ReplicatedGTID({ - raw_gtid: 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:abc-def:1-9', - position: POSITION - }); - expect(garbageInterval.comparable).not.toContain('NaN'); - expect(garbageInterval.comparable.split('|')[0]).toEqual('0000000000000009'); + 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 multi-UUID GTID set', () => { - const raw = '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17,\n314306f3-ff7b-11ef-a0e0-566fbaa00002:1-2734181'; - const gtid = new ReplicatedGTID({ raw_gtid: raw, position: POSITION }); + 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(raw); + + 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(); - // Missing binlog offset - expect(() => ReplicatedGTID.fromSerialized('0000000000000001|uuid:1|binlog.000001')).toThrow(); - expect(() => ReplicatedGTID.fromSerialized('0000000000000001|uuid:1|binlog.000001|notanumber')).toThrow(); + 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' + ); }); - }); - describe('LSN ordering', () => { - test('LSNs for the same transaction order by binlog offset', () => { - // Note: the binlog offset is not zero-padded, so lexicographic ordering only holds for - // offsets with the same number of digits. This is sufficient for the checkpoint gate since - // heartbeat keepalive LSNs are byte-identical to the last commit LSN, but is documented - // here as a known limitation of the format (which cannot change for compatibility with - // LSNs already persisted in bucket storage). - const raw = 'a7d0ff7b-0c0e-11f0-8b38-566fbaa00004:18'; - const transactionStart = new ReplicatedGTID({ - raw_gtid: raw, - position: { filename: 'binlog.000042', offset: 157 } - }); - const transactionEnd = new ReplicatedGTID({ - raw_gtid: raw, - position: { filename: 'binlog.000042', offset: 300 } - }); - expect(transactionStart.comparable < transactionEnd.comparable).toBeTruthy(); - }); + test('rejects a serialized GTID set', () => { + const serialized = `0000000000000017|${SERVER_UUID}:1-17|binlog.000042|1234`; - test('correct LSNs order against legacy NaN-corrupted LSNs as documented', () => { - // LSNs produced by the previous multi-UUID parsing bug contain a literal 'NaN' padded transaction id. - const legacyPoisoned = '0000000000000NaN|2e35321d-0c0e-11f0-8b38-566fbaa00004:1-17|binlog.000042|1234'; + expect(() => ReplicatedGTID.fromSerialized(serialized)).toThrow('Expected a single transaction id'); + }); + }); - // Instances with transaction ids >= 1000 sort above the corrupted LSN and self-heal - const highTransaction = new ReplicatedGTID({ - raw_gtid: '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-39489900', + 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(highTransaction.comparable > legacyPoisoned).toBeTruthy(); - // Instances with transaction ids < 1000 still sort below it ('N' > any digit) and require a resync - const lowTransaction = new ReplicatedGTID({ - raw_gtid: '2e35321d-0c0e-11f0-8b38-566fbaa00004:1-999', - position: POSITION - }); - expect(lowTransaction.comparable < legacyPoisoned).toBeTruthy(); + expect(gtid.raw).toEqual(`${SERVER_UUID}:17`); + expect(gtid.position).toEqual(POSITION); + expect(gtid.comparable).toEqual(`0000000000000017|${SERVER_UUID}:17|binlog.000042|1234`); }); }); - describe('connected server UUID selection', () => { - const ACTIVE = '2e35321d-0c0e-11f0-8b38-566fbaa00004'; - const STALE = '314306f3-ff7b-11ef-a0e0-566fbaa00002'; + 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 }); - test('uses the connected server counter even when a stale UUID holds a higher one', () => { - // A restore from another server leaves the old UUID with a high counter. Ordering by the set-wide - // maximum would pin the LSN there and hang checkpoints until the active counter catches up. - const gtid = new ReplicatedGTID({ - raw_gtid: `${STALE}:1-42350493,\n${ACTIVE}:1-17`, - position: POSITION, - serverUuid: ACTIVE - }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000017'); + expect(earlier.comparable < later.comparable).toBeTruthy(); }); - test('matches the set-wide maximum when the connected server holds it', () => { - const gtid = new ReplicatedGTID({ - raw_gtid: `${ACTIVE}:1-42350493,\n${STALE}:1-2734181`, - position: POSITION, - serverUuid: ACTIVE + 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 } }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000042350493'); - }); - - test('falls back to the set-wide maximum when the connected server has no transactions in the set', () => { - const gtid = new ReplicatedGTID({ - raw_gtid: `${STALE}:1-2734181`, - position: POSITION, - serverUuid: ACTIVE + const transactionEnd = new ReplicatedGTID({ + rawGtid, + position: { filename: 'binlog.000042', offset: 300 } }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000002734181'); - }); - test('single-GTID values from the connected server are unaffected', () => { - const gtid = new ReplicatedGTID({ raw_gtid: `${ACTIVE}:18`, position: POSITION, serverUuid: ACTIVE }); - expect(gtid.comparable.split('|')[0]).toEqual('0000000000000018'); - }); - - test('fromSerialized produces the same LSN when the same server UUID is provided', () => { - const gtid = new ReplicatedGTID({ - raw_gtid: `${STALE}:1-42350493,\n${ACTIVE}:1-17`, - position: POSITION, - serverUuid: ACTIVE - }); - const deserialized = ReplicatedGTID.fromSerialized(gtid.comparable, ACTIVE); - expect(deserialized.comparable).toEqual(gtid.comparable); - expect(deserialized.activeServerUuid).toEqual(ACTIVE); + expect(transactionStart.comparable < transactionEnd.comparable).toBeTruthy(); }); }); }); 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..9d6b2a06a --- /dev/null +++ b/modules/module-mysql/test/src/read-executed-gtid.test.ts @@ -0,0 +1,158 @@ +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('throws when the active server is absent from the GTID sets', async () => { + await expect(getLatestActiveGtid([`${STALE_SERVER_UUID}:1-1000`], ACTIVE_SERVER_UUID)).rejects.toThrow( + `No GTID set found matching Active server UUID: ${ACTIVE_SERVER_UUID}` + ); + }); + }); + + 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.comparable).toEqual(`0000000000000000|${ACTIVE_SERVER_UUID}:0||0`); + }); + }); + + 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.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 77fa33348..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 }); } From 04920753baf2013367366aa5f8dda7620f9d774c Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 12:45:11 +0200 Subject: [PATCH 06/11] Add replica check to MySQL source configuration with version-based syntax handling and tests --- .../src/common/check-source-configuration.ts | 15 ++++ .../src/check-source-configuration.test.ts | 70 +++++++++++++++++++ 2 files changed, 85 insertions(+) create mode 100644 modules/module-mysql/test/src/check-source-configuration.test.ts 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/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}`); + } + }); + } +}); From 36817b7ae6a2c16d3b4398e28480fa3849aa5acc Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 12:58:20 +0200 Subject: [PATCH 07/11] Enforce active server UUID consistency in MySQL replication to prevent cross-server transaction processing errors and handle GTID changes more robustly. --- .../src/replication/BinLogStream.ts | 41 +++++-- .../src/replication/zongji/BinLogListener.ts | 100 +++++++++--------- 2 files changed, 80 insertions(+), 61 deletions(-) diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index 4d3522ca8..3f955dfd6 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 }, @@ -354,6 +363,7 @@ export class BinLogStream { try { // If anything errors here, the entire replication process is halted, and // all connections automatically closed, including this one. + await this.setActiveServerUuid(); await this.initReplication(); await this.streamChanges(); this.logger.info('BinLogStream has been shut down'); @@ -363,6 +373,15 @@ export class BinLogStream { } } + private async setActiveServerUuid() { + const connection = await this.connections.getConnection(); + try { + this.activeServerUuid = await common.readServerUuid(connection); + } finally { + connection.release(); + } + } + async initReplication() { const connection = await this.connections.getConnection(); const errors = await common.checkSourceConfiguration(connection); @@ -382,7 +401,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 }, @@ -415,9 +434,8 @@ export class BinLogStream { if (resume_lsn) { this.logger.info(`Existing resume LSN found: ${resume_lsn}`); } - // The server uuid picks the connected server's transaction counter out of GTID sets for LSN ordering. fromGTID = resume_lsn - ? common.ReplicatedGTID.fromSerialized(resume_lsn, await common.readServerUuid(connection)) + ? common.ReplicatedGTID.fromSerialized(resume_lsn) : await common.readExecutedGtid(connection); } finally { connection.release(); @@ -425,7 +443,11 @@ export class BinLogStream { 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({ @@ -434,6 +456,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 8797370c4..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,10 +106,7 @@ export class BinLogListener { // Flag to indicate if are currently in a transaction that involves multiple row mutation events. private isTransactionOpen = false; - // The @@server_uuid of the connected server, used to order LSNs and to detect foreign transactions. - private currentServerUuid: string | undefined; - // Foreign server UUIDs already warned about, so each one is logged once rather than per transaction. - private warnedForeignServerIds = new Set(); + zongji: ZongJi; processingQueue: async.QueueObject; @@ -115,12 +117,9 @@ export class BinLogListener { constructor(public options: BinLogListenerOptions) { this.logger = options.logger ?? defaultLogger; - this.connectionManager = options.connectionManager; - this.eventHandler = options.eventHandler; // 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.currentServerUuid = options.startGTID.activeServerUuid; this.sqlParser = new Parser(); this.processingQueue = this.createProcessingQueue(); this.zongji = this.createZongjiListener(); @@ -128,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 @@ -298,33 +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 - }, - position: { - filename: this.binLogPosition.filename, - offset: evt.nextPosition - } + const transactionGTID = common.ReplicatedGTID.fromBinLogEvent({ + rawGtid: { + serverUuid: evt.serverId, // The server uuid this transaction originated from + transactionId: evt.transactionRange }, - this.currentServerUuid - ); + 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; - this.warnOnForeignServerUuid(this.currentGTID.serverId); + 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}` ); @@ -391,35 +410,12 @@ export class BinLogListener { private advanceCommitPosition(nextPosition: number): string { this.binLogPosition.offset = nextPosition; this.currentGTID = new common.ReplicatedGTID({ - raw_gtid: this.currentGTID.raw, - // Copy the position: this.binLogPosition is mutated by subsequent events - position: { ...this.binLogPosition }, - serverUuid: this.currentServerUuid + rawGtid: this.currentGTID.raw, + position: { ...this.binLogPosition } }); return this.currentGTID.comparable; } - /** - * Warns when a transaction on the binlog originates from a server other than the connected one, such as - * when the connected server is a replica. LSN ordering follows the connected server's transaction - * counter, so transactions from other server UUIDs are not reliably ordered and can stall checkpoints. - * Warns once per foreign UUID rather than per transaction. - */ - private warnOnForeignServerUuid(transactionServerUuid: string): void { - if ( - this.currentServerUuid == null || - transactionServerUuid === this.currentServerUuid || - this.warnedForeignServerIds.has(transactionServerUuid) - ) { - return; - } - this.warnedForeignServerIds.add(transactionServerUuid); - this.logger.warn( - `Detected a transaction from a different MySQL server UUID on the binlog: ${transactionServerUuid} (connected server: ${this.currentServerUuid}). ` + - `LSN ordering follows the connected server's transaction counter, so transactions from other servers are not reliably ordered and checkpoints may stall.` - ); - } - private async processQueryEvent(event: BinLogQueryEvent): Promise { const { query, nextPosition } = event; From f58faa7daa84b6bcba9eb0f5b3a239a78a99b9f6 Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 13:10:47 +0200 Subject: [PATCH 08/11] Updated changeset --- .changeset/mysql-gtid-consistency-fixes.md | 5 +++++ .changeset/mysql-gtid-set-parsing.md | 5 ----- .changeset/mysql-idle-keepalive-lsn.md | 5 ----- 3 files changed, 5 insertions(+), 10 deletions(-) create mode 100644 .changeset/mysql-gtid-consistency-fixes.md delete mode 100644 .changeset/mysql-gtid-set-parsing.md delete mode 100644 .changeset/mysql-idle-keepalive-lsn.md 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/.changeset/mysql-gtid-set-parsing.md b/.changeset/mysql-gtid-set-parsing.md deleted file mode 100644 index 0e9fbbabc..000000000 --- a/.changeset/mysql-gtid-set-parsing.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@powersync/service-module-mysql': patch ---- - -Fix GTID parsing for multi-server-UUID GTID sets. Previously a `gtid_executed` containing multiple server UUIDs (e.g. after a failover or restore) produced a `NaN` transaction id in the comparable LSN, which could permanently block checkpoint creation. GTID sets with multiple intervals per server UUID are now also parsed correctly. LSN ordering now follows the connected server's own transaction counter (`@@server_uuid`) instead of the highest counter in the set, so a stale server UUID with a higher counter can no longer hang checkpoints. A warning is logged when a transaction from a different server UUID appears on the binlog (e.g. when connected to a replica), since those are not reliably ordered yet. diff --git a/.changeset/mysql-idle-keepalive-lsn.md b/.changeset/mysql-idle-keepalive-lsn.md deleted file mode 100644 index 48172cfe2..000000000 --- a/.changeset/mysql-idle-keepalive-lsn.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@powersync/service-module-mysql': patch ---- - -Fix checkpoints stalling on idle MySQL servers. Heartbeat keepalives now report the LSN of the last committed transaction instead of the transaction start position, which previously blocked checkpoint creation ("Waiting before creating checkpoint" logged every ~30s) until the next transaction arrived. From e3ca1d48329588393ed78e939c5e69848ae8fcea Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 13:30:10 +0200 Subject: [PATCH 09/11] Ensure active server UUID is always set --- .../src/replication/BinLogStream.ts | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/modules/module-mysql/src/replication/BinLogStream.ts b/modules/module-mysql/src/replication/BinLogStream.ts index 3f955dfd6..7274a46fc 100644 --- a/modules/module-mysql/src/replication/BinLogStream.ts +++ b/modules/module-mysql/src/replication/BinLogStream.ts @@ -363,7 +363,6 @@ export class BinLogStream { try { // If anything errors here, the entire replication process is halted, and // all connections automatically closed, including this one. - await this.setActiveServerUuid(); await this.initReplication(); await this.streamChanges(); this.logger.info('BinLogStream has been shut down'); @@ -373,16 +372,19 @@ export class BinLogStream { } } - private async setActiveServerUuid() { - const connection = await this.connections.getConnection(); - try { - this.activeServerUuid = await common.readServerUuid(connection); - } finally { - connection.release(); + 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(); @@ -425,6 +427,7 @@ export class BinLogStream { } async streamChanges() { + await this.ensureActiveServerUuid(); const serverId = createRandomServerId(this.storage.replicationStreamId); const connection = await this.connections.getConnection(); From ce722ecb00b36471cce004e0318f86bd6a5d67cb Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 13:52:54 +0200 Subject: [PATCH 10/11] Handle GTID reset for a restored mysql server with a different UUID --- .../src/common/read-executed-gtid.ts | 5 +---- .../test/src/read-executed-gtid.test.ts | 19 ++++++++++++++++--- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/modules/module-mysql/src/common/read-executed-gtid.ts b/modules/module-mysql/src/common/read-executed-gtid.ts index ce9889d65..30cc21844 100644 --- a/modules/module-mysql/src/common/read-executed-gtid.ts +++ b/modules/module-mysql/src/common/read-executed-gtid.ts @@ -1,4 +1,3 @@ -import { ReplicationAssertionError } from '@powersync/lib-services-framework'; import mysqlPromise from 'mysql2/promise'; import * as mysql_utils from '../utils/mysql-utils.js'; import { ReplicatedGTID } from './ReplicatedGTID.js'; @@ -76,9 +75,7 @@ export async function getLatestActiveGtid(gtidSets: string[], activeServerUuid: } } - throw new ReplicationAssertionError( - `No GTID set found matching Active server UUID: ${activeServerUuid} in GTID sets: ${gtidSets.join(', ')}` - ); + return `${activeServerUuid}:0`; } /** diff --git a/modules/module-mysql/test/src/read-executed-gtid.test.ts b/modules/module-mysql/test/src/read-executed-gtid.test.ts index 9d6b2a06a..fe0d99a22 100644 --- a/modules/module-mysql/test/src/read-executed-gtid.test.ts +++ b/modules/module-mysql/test/src/read-executed-gtid.test.ts @@ -27,9 +27,9 @@ describe('read-executed-gtid', () => { expect(gtid).toEqual(`${ACTIVE_SERVER_UUID}:42`); }); - test('throws when the active server is absent from the GTID sets', async () => { - await expect(getLatestActiveGtid([`${STALE_SERVER_UUID}:1-1000`], ACTIVE_SERVER_UUID)).rejects.toThrow( - `No GTID set found matching Active server UUID: ${ACTIVE_SERVER_UUID}` + 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` ); }); }); @@ -73,6 +73,19 @@ describe('read-executed-gtid', () => { expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:0`); expect(gtid.comparable).toEqual(`0000000000000000|${ACTIVE_SERVER_UUID}:0||0`); }); + + 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', () => { From 8b5cf9bdd074502261777a8a5b7387be1273b5aa Mon Sep 17 00:00:00 2001 From: Roland Teichert Date: Fri, 7 Aug 2026 14:02:35 +0200 Subject: [PATCH 11/11] Improve GTID initialization logic --- .../src/common/read-executed-gtid.ts | 15 +++++++++++++-- .../test/src/read-executed-gtid.test.ts | 19 ++++++++++++++++++- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/modules/module-mysql/src/common/read-executed-gtid.ts b/modules/module-mysql/src/common/read-executed-gtid.ts index 30cc21844..1be9a6635 100644 --- a/modules/module-mysql/src/common/read-executed-gtid.ts +++ b/modules/module-mysql/src/common/read-executed-gtid.ts @@ -43,8 +43,12 @@ export async function readExecutedGtid(connection: mysqlPromise.Connection): Pro const executedGtidSet = binlogStatus.Executed_Gtid_Set.trim(); if (executedGtidSet.length === 0) { - // New server with no transactions executed yet - return ReplicatedGTID.ZERO(activeServerUuid); + // 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(','); @@ -97,6 +101,13 @@ export async function isGtidPositionStillAvailable( 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`, diff --git a/modules/module-mysql/test/src/read-executed-gtid.test.ts b/modules/module-mysql/test/src/read-executed-gtid.test.ts index fe0d99a22..6a09410be 100644 --- a/modules/module-mysql/test/src/read-executed-gtid.test.ts +++ b/modules/module-mysql/test/src/read-executed-gtid.test.ts @@ -71,7 +71,8 @@ describe('read-executed-gtid', () => { const gtid = await readExecutedGtid(connection); expect(gtid.raw).toEqual(`${ACTIVE_SERVER_UUID}:0`); - expect(gtid.comparable).toEqual(`0000000000000000|${ACTIVE_SERVER_UUID}:0||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 () => { @@ -115,6 +116,22 @@ describe('read-executed-gtid', () => { 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']