Post-merge updates for MySQL.

Author: rkistner

modules/module-mysql/src/replication/BinLogStream.ts

@@ -484,7 +484,19 @@ AND table_type = 'BASE TABLE';`,
       },
       onCommit: async (lsn: string) => {
         this.metrics.getCounter(ReplicationMetric.TRANSACTIONS_REPLICATED).add(1);
-        await batch.commit(lsn);
+        const didCommit = await batch.commit(lsn, { oldestUncommittedChange: this.oldestUncommittedChange });
+        if (didCommit) {
+          this.oldestUncommittedChange = null;
+          this.isStartingReplication = false;
+        }
+      },
+      onTransactionStart: async (options) => {
+        if (this.oldestUncommittedChange == null) {
+          this.oldestUncommittedChange = options.timestamp;
+        }
+      },
+      onRotate: async () => {
+        this.isStartingReplication = false;
       }
     };
   }

modules/module-mysql/src/replication/zongji/BinLogListener.ts

@@ -12,6 +12,8 @@ const MAX_QUEUE_PAUSE_TIME_MS = 45_000;
 export type Row = Record<string, any>;
 
 export interface BinLogEventHandler {
+  onTransactionStart: (options: { timestamp: Date }) => Promise<void>;
+  onRotate: () => Promise<void>;
   onWrite: (rows: Row[], tableMap: TableMapEntry) => Promise<void>;
   onUpdate: (rowsAfter: Row[], rowsBefore: Row[], tableMap: TableMapEntry) => Promise<void>;
   onDelete: (rows: Row[], tableMap: TableMapEntry) => Promise<void>;
@@ -196,10 +198,12 @@ export class BinLogListener {
               offset: evt.nextPosition
             }
           });
+          await this.eventHandler.onTransactionStart({ timestamp: new Date(evt.timestamp) });
           break;
         case zongji_utils.eventIsRotation(evt):
           this.binLogPosition.filename = evt.binlogName;
           this.binLogPosition.offset = evt.position;
+          await this.eventHandler.onRotate();
           break;
         case zongji_utils.eventIsWriteMutation(evt):
           await this.eventHandler.onWrite(evt.rows, evt.tableMap[evt.tableId]);