[MySQL] Fix idle keepalive LSN stall and multi-server-UUID GTID parsing

.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.

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) {

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,20 +9,24 @@ export type BinLogPosition = {
 };
 
 export type ReplicatedGTIDSpecification = {
-  raw_gtid: string;
+  /**
+   * The raw Global Transaction ID. This is of the format `server_uuid:transaction_id`.
+   * Must be a single GTID — not a GTID set (multiple UUIDs) or interval range.
+   */
+  rawGtid: string;
   /**
    * The (end) position in a BinLog file where this transaction has been replicated in.
    */
   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;
 };
 
@@ -31,40 +36,56 @@ export type BinLogGTIDEvent = {
  * and position where this GTID could be located.
  */
 export class ReplicatedGTID {
+  private options: ReplicatedGTIDSpecification;
+
+  constructor(options: ReplicatedGTIDSpecification) {
+    const rawGtid = options.rawGtid.trim();
+    assertSingleGtid(rawGtid);
+    this.options = { ...options, rawGtid };
+  }
+
   static fromSerialized(comparable: string): ReplicatedGTID {
     return new ReplicatedGTID(ReplicatedGTID.deserialize(comparable));
   }
 
   private static deserialize(comparable: string): ReplicatedGTIDSpecification {
     const components = comparable.split('|');
-    if (components.length < 3) {
-      throw new Error(`Invalid serialized GTID: ${comparable}`);
+    if (components.length < 4) {
+      throw new ReplicationAssertionError(`Invalid serialized GTID: ${comparable}`);
+    }
+
+    const offset = parseInt(components[3], 10);
+    if (Number.isNaN(offset)) {
+      throw new ReplicationAssertionError(`Invalid BinLog offset in serialized GTID: ${comparable}`);
     }
 
     return {
-      raw_gtid: components[1],
+      rawGtid: components[1],
       position: {
         filename: components[2],
-        offset: parseInt(components[3])
+        offset: offset
       } satisfies BinLogPosition
     };
   }
 
   static fromBinLogEvent(event: BinLogGTIDEvent) {
-    const { raw_gtid, position } = event;
-    const stringGTID = `${uuid.stringify(raw_gtid.server_id)}:${raw_gtid.transaction_range}`;
+    const { rawGtid, position } = event;
+    const stringGTID = `${uuid.stringify(rawGtid.serverUuid)}:${rawGtid.transactionId}`;
     return new ReplicatedGTID({
-      raw_gtid: stringGTID,
+      rawGtid: stringGTID,
       position
     });
   }
 
   /**
    * Special case for the zero GTID which means no transactions have been executed.
    */
-  static ZERO = new ReplicatedGTID({ raw_gtid: '0:0', position: { filename: '', offset: 0 } });
-
-  constructor(protected options: ReplicatedGTIDSpecification) {}
+  static ZERO(serverUuid: string): ReplicatedGTID {
+    return new ReplicatedGTID({
+      rawGtid: `${serverUuid}:0`,
+      position: { filename: '', offset: 0 }
+    });
+  }
 
   /**
    * Get the BinLog position of this replicated GTID event
@@ -74,14 +95,17 @@ export class ReplicatedGTID {
   }
 
   /**
-   * Get the raw Global Transaction ID. This of the format `server_id:transaction_ranges`
+   * Get the raw Global Transaction ID. This is of the format `server_uuid:transaction_id`
    */
   get raw() {
-    return this.options.raw_gtid;
+    return this.options.rawGtid;
   }
 
-  get serverId() {
-    return this.options.raw_gtid.split(':')[0];
+  /**
+   * The server UUID of the server this transaction originated from
+   */
+  get serverUuid() {
+    return this.options.rawGtid.split(':')[0];
   }
 
   /**
@@ -94,21 +118,9 @@ export class ReplicatedGTID {
    */
   get comparable(): string {
     const { raw, position } = this;
-    const [, transactionRanges] = this.raw.split(':');
+    const [, transactionId] = this.raw.split(':');
 
-    // This means no transactions have been executed on the database yet
-    if (!transactionRanges) {
-      return ReplicatedGTID.ZERO.comparable;
-    }
-
-    let maxTransactionId = 0;
-
-    for (const range of transactionRanges.split(',')) {
-      const [start, end] = range.split('-');
-      maxTransactionId = Math.max(maxTransactionId, parseInt(start, 10), parseInt(end || start, 10));
-    }
-
-    const paddedTransactionId = maxTransactionId.toString().padStart(16, '0');
+    const paddedTransactionId = transactionId.toString().padStart(16, '0');
     return [paddedTransactionId, raw, position.filename, position.offset].join('|');
   }
 
@@ -161,3 +173,24 @@ export class ReplicatedGTID {
     );
   }
 }
+
+/**
+ * Asserts that the given gtid string is a single GTID of the form `server_uuid:transaction_id`,
+ * not a GTID set such as `uuid:1-17` or `uuid1:1,uuid2:2`.
+ */
+function assertSingleGtid(gtid: string): void {
+  // GTID sets join UUID sets with commas (often with newlines: `,\n`).
+  if (gtid.includes(',') || gtid.includes('\n')) {
+    throw new ReplicationAssertionError(`Expected a single GTID (server_uuid:transaction_id), got a GTID set: ${gtid}`);
+  }
+
+  const parts = gtid.split(':');
+  if (parts.length !== 2 || parts[0].length === 0 || parts[1].length === 0) {
+    throw new ReplicationAssertionError(`Expected a single GTID (server_uuid:transaction_id), got: ${gtid}`);
+  }
+
+  // Intervals use `n-m`; a single transaction id must be a non-negative integer.
+  if (!/^\d+$/.test(parts[1])) {
+    throw new ReplicationAssertionError(`Expected a single transaction id, got: ${gtid}`);
+  }
+}

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<string[]> {
   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: `

modules/module-mysql/src/common/read-executed-gtid.ts

@@ -2,6 +2,17 @@ import mysqlPromise from 'mysql2/promise';
 import * as mysql_utils from '../utils/mysql-utils.js';
 import { ReplicatedGTID } from './ReplicatedGTID.js';
 
+/**
+ * Gets the `@@server_uuid` of the current connected server
+ */
+export async function readServerUuid(connection: mysqlPromise.Connection): Promise<string> {
+  const [[result]] = await mysql_utils.retriedQuery({
+    connection,
+    query: `SELECT @@server_uuid AS server_uuid`
+  });
+  return result.server_uuid;
+}
+
 /**
  * Gets the current master HEAD GTID
  */
@@ -28,21 +39,80 @@ export async function readExecutedGtid(connection: mysqlPromise.Connection): Pro
     offset: parseInt(binlogStatus.Position)
   };
 
+  const activeServerUuid = await readServerUuid(connection);
+  const executedGtidSet = binlogStatus.Executed_Gtid_Set.trim();
+
+  if (executedGtidSet.length === 0) {
+    // New server with no transactions executed yet. Keep the current binlog
+    // coordinate so this synthetic GTID can still be validated after a restart.
+    return new ReplicatedGTID({
+      rawGtid: `${activeServerUuid}:0`,
+      position
+    });
+  }
+
+  const gtidSets = executedGtidSet.split(',');
+  const latestActiveGtid = await getLatestActiveGtid(gtidSets, activeServerUuid);
   return new ReplicatedGTID({
-    // The head always points to the next position to start replication from
-    position,
-    raw_gtid: binlogStatus.Executed_Gtid_Set
+    rawGtid: latestActiveGtid,
+    position
   });
 }
 
-export async function isBinlogStillAvailable(
+export async function getLatestActiveGtid(gtidSets: string[], activeServerUuid: string): Promise<string> {
+  for (const gtidSet of gtidSets) {
+    const [serverUuid, ...intervals] = gtidSet.trim().split(':');
+    if (serverUuid === activeServerUuid) {
+      let maxTransactionId: number | null = null;
+      for (const interval of intervals) {
+        const [start, end] = interval.split('-');
+        const startId = parseInt(start, 10);
+        const endId = end !== undefined ? parseInt(end, 10) : startId;
+        if (!Number.isNaN(startId)) {
+          maxTransactionId = Math.max(maxTransactionId ?? 0, startId);
+        }
+        if (!Number.isNaN(endId)) {
+          maxTransactionId = Math.max(maxTransactionId ?? 0, endId);
+        }
+      }
+      return activeServerUuid + ':' + maxTransactionId;
+    }
+  }
+
+  return `${activeServerUuid}:0`;
+}
+
+/**
+ * Checks that a stored resume GTID is still part of the server's executed history and that its
+ * binlog coordinate is still readable. This detects source rewinds where a restored server keeps
+ * the same UUID or recreates a binlog with the same filename but a shorter length.
+ */
+export async function isGtidPositionStillAvailable(
   connection: mysqlPromise.Connection,
-  binlogFile: string
+  gtid: ReplicatedGTID
 ): Promise<boolean> {
   const [logFiles] = await mysql_utils.retriedQuery({
     connection,
     query: `SHOW BINARY LOGS;`
   });
+  const logFile = logFiles.find((file) => file['Log_name'] == gtid.position.filename);
+
+  if (!logFile || Number(logFile['File_size']) < gtid.position.offset) {
+    return false;
+  }
+
+  // Transaction zero is PowerSync's synthetic position before the first
+  // transaction from this server UUID. It is not valid MySQL GTID_SET syntax,
+  // so its availability is determined by the binlog coordinate above.
+  if (gtid.raw.split(':')[1] === '0') {
+    return true;
+  }
+
+  const [[result]] = await mysql_utils.retriedQuery({
+    connection,
+    query: `SELECT GTID_SUBSET(?, @@GLOBAL.gtid_executed) AS is_executed`,
+    params: [gtid.raw]
+  });
 
-  return logFiles.some((f) => f['Log_name'] == binlogFile);
+  return result.is_executed === 1;
 }