core: services: recoder_extractor: Add mcap recover

Signed-off-by: Patrick José Pereira <patrickelectric@gmail.com>

Author: patrickelectric

core/services/recorder_extractor/main.py

@@ -3,6 +3,7 @@
 import asyncio
 import contextlib
 import logging
+import shutil
 from functools import wraps
 from io import BytesIO
 from pathlib import Path
@@ -101,6 +102,61 @@ def parse_duration_ns(discover_output: str) -> int:
     return duration_ns
 
 
+async def check_and_recover_mcap(mcap_path: Path) -> None:
+    """
+    Check if mcap binary is available, run mcap doctor on the file,
+    and if it fails, run mcap recover to fix the file.
+    """
+    # Check if mcap binary exists
+    mcap_binary = shutil.which("mcap")
+    if not mcap_binary:
+        logger.warning("mcap binary not found, skipping doctor/recover check")
+        return
+
+    # Ensure path is absolute and exists
+    if not mcap_path.exists():
+        logger.warning(f"MCAP file not found: {mcap_path}")
+        return
+
+    logger.info(f"Running mcap doctor on {mcap_path}")
+    # Run mcap doctor
+    doctor_cmd = ["mcap", "doctor", str(mcap_path)]
+    doctor_proc = await asyncio.create_subprocess_exec(
+        *doctor_cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        text=False,
+    )
+    stdout_bytes, stderr_bytes = await doctor_proc.communicate()
+    stdout = stdout_bytes.decode("utf-8", "ignore")
+    stderr = stderr_bytes.decode("utf-8", "ignore")
+
+    if doctor_proc.returncode == 0:
+        logger.info(f"mcap doctor passed for {mcap_path}: {stdout.strip()}")
+        return
+
+    logger.error(f"mcap doctor failed for {mcap_path} (code={doctor_proc.returncode}): {stderr}")
+    logger.info(f"Running mcap recover to fix {mcap_path}")
+    # Run mcap recover to replace the file
+    recover_cmd = ["mcap", "recover", str(mcap_path), "-o", str(mcap_path)]
+    recover_proc = await asyncio.create_subprocess_exec(
+        *recover_cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        text=False,
+    )
+    recover_stdout_bytes, recover_stderr_bytes = await recover_proc.communicate()
+    recover_stdout = recover_stdout_bytes.decode("utf-8", "ignore")
+    recover_stderr = recover_stderr_bytes.decode("utf-8", "ignore")
+
+    if recover_proc.returncode != 0:
+        logger.error(
+            f"mcap recover failed for {mcap_path} (code={recover_proc.returncode}): {recover_stderr}",
+        )
+    else:
+        logger.info(f"mcap recover completed for {mcap_path}: {recover_stdout.strip()}")
+
+
 @cached()
 async def build_thumbnail_bytes(path: Path) -> bytes:
     """
@@ -187,6 +243,9 @@ async def extract_mcap_recordings() -> None:
                     logger.info(f"Skipping MCAP extract, file in use: {mcap_path}")
                     continue
 
+                # Check and recover MCAP file if mcap binary is available
+                await check_and_recover_mcap(mcap_path)
+
                 command = [
                     "mcap-foxglove-video-extract",
                     str(mcap_path),