Implement binlog recovery utilities and integrate with replicators

- Added a new module for handling MySQL binlog corruption (Error 1236) with automatic recovery functionality.
- Integrated recovery logic into both DbReplicatorRealtime and BinlogReplicator to streamline error handling and process restart.
- Updated .gitignore to exclude the binlog directory instead of files for better management.

Author: jaredmdobson

.gitignore

@@ -1,7 +1,7 @@
 __pycache__
 .idea/
 config.yaml
-binlog*
+binlog/
 *cmake_build*
 monitoring.log
 .DS_Store

mysql_ch_replicator/binlog_recovery.py

@@ -0,0 +1,49 @@
+"""
+Shared binlog recovery utilities for handling MySQL Error 1236 (binlog corruption).
+"""
+import os
+import shutil
+from logging import getLogger
+
+logger = getLogger(__name__)
+
+
+def recover_from_binlog_corruption(binlog_dir: str, error: Exception) -> None:
+    """
+    Recover from MySQL Error 1236 (binlog corruption) by deleting the corrupted
+    binlog directory and raising an exception to trigger process restart.
+
+    Args:
+        binlog_dir: Path to the binlog directory to delete
+        error: The original OperationalError that triggered recovery
+
+    Raises:
+        RuntimeError: Always raised to trigger process restart after cleanup
+
+    This function:
+    1. Logs the error and recovery attempt
+    2. Deletes the corrupted binlog directory
+    3. Raises RuntimeError to exit the process cleanly
+    4. ProcessRunner will automatically restart the process
+    5. On restart, replication resumes from a fresh state
+    """
+    logger.error(f"[binlogrepl] operational error (1236, 'Could not find first log file name in binary log index file')")
+    logger.error(f"[binlogrepl] Full error: {error}")
+    logger.info("[binlogrepl] Error 1236 detected - attempting automatic recovery")
+
+    # Delete the corrupted binlog directory to force fresh start
+    if os.path.exists(binlog_dir):
+        logger.warning(f"[binlogrepl] Deleting corrupted binlog directory: {binlog_dir}")
+        try:
+            shutil.rmtree(binlog_dir)
+            logger.info(f"[binlogrepl] Successfully deleted binlog directory: {binlog_dir}")
+        except Exception as delete_error:
+            logger.error(f"[binlogrepl] Failed to delete binlog directory: {delete_error}", exc_info=True)
+            raise RuntimeError("Failed to delete corrupted binlog directory") from delete_error
+    else:
+        logger.warning(f"[binlogrepl] Binlog directory does not exist: {binlog_dir}")
+
+    # Exit process cleanly to trigger automatic restart by runner
+    logger.info("[binlogrepl] Exiting process for automatic restart by runner")
+    logger.info("[binlogrepl] The runner will automatically restart this process")
+    raise RuntimeError("Binlog corruption detected (Error 1236) - restarting for recovery") from error

mysql_ch_replicator/binlog_replicator.py

@@ -12,6 +12,7 @@
 
 from pymysql.err import OperationalError
 
+from .binlog_recovery import recover_from_binlog_corruption
 from .config import BinlogReplicatorSettings, Settings
 from .pymysqlreplication import BinLogStreamReader
 from .pymysqlreplication.event import QueryEvent
@@ -617,6 +618,11 @@ def run(self):
                     time.sleep(BinlogReplicator.READ_LOG_INTERVAL)
 
             except OperationalError as e:
+                # Check if this is Error 1236 (binlog corruption) - needs automatic recovery
+                if e.args[0] == 1236:
+                    recover_from_binlog_corruption(self.replicator_settings.data_dir, e)
+
+                # For other operational errors, log and retry
                 logger.error(f"operational error {str(e)}", exc_info=True)
                 time.sleep(15)
             except Exception as e:

mysql_ch_replicator/db_replicator_realtime.py

@@ -1,12 +1,12 @@
 import json
 import os
-import shutil
 import time
 from collections import defaultdict
 from logging import getLogger
 
 import pymysql.err
 
+from .binlog_recovery import recover_from_binlog_corruption
 from .binlog_replicator import EventType, LogEvent
 from .common import Status
 from .converter import strip_sql_comments
@@ -79,34 +79,12 @@ def run_realtime_replication(self):
             except pymysql.err.OperationalError as e:
                 # Check if this is the binlog index file corruption error (Error 1236)
                 if e.args[0] == 1236:
-                    logger.error(
-                        "[binlogrepl] operational error (1236, 'Could not find first log file name in binary log index file')"
-                    )
-                    logger.error(f"[binlogrepl] Full error: {e}")
-                    logger.info("[binlogrepl] Attempting automatic recovery...")
-
                     # Get binlog directory path for this database
                     binlog_dir = os.path.join(
                         self.replicator.config.binlog_replicator.data_dir,
                         self.replicator.database
                     )
-
-                    # Delete the corrupted binlog directory
-                    if os.path.exists(binlog_dir):
-                        logger.warning(f"[binlogrepl] Deleting corrupted binlog directory: {binlog_dir}")
-                        try:
-                            shutil.rmtree(binlog_dir)
-                            logger.info(f"[binlogrepl] Successfully deleted binlog directory: {binlog_dir}")
-                        except Exception as delete_error:
-                            logger.error(f"[binlogrepl] Failed to delete binlog directory: {delete_error}", exc_info=True)
-                            raise RuntimeError("Failed to delete corrupted binlog directory") from delete_error
-                    else:
-                        logger.warning(f"[binlogrepl] Binlog directory does not exist: {binlog_dir}")
-
-                    # Exit process cleanly to trigger automatic restart by runner
-                    logger.info("[binlogrepl] Exiting process for automatic restart by runner")
-                    logger.info("[binlogrepl] The runner will automatically restart this process")
-                    raise RuntimeError("Binlog corruption detected (Error 1236) - restarting for recovery") from e
+                    recover_from_binlog_corruption(binlog_dir, e)
                 else:
                     # Re-raise other OperationalErrors
                     logger.error(f"[binlogrepl] Unhandled OperationalError: {e}", exc_info=True)