fix(executors): run backups off the event loop

The scheduler serves its own REST API from the same loop that runs executors,
and the config backup spends ~21 minutes inside tarfile and zlib. Called inline
that starves the loop for the whole window: the service was unreachable
03:05-03:25 every night, and again at 07:39 today when the job was triggered by
hand to prove the T-69 report path. asyncio.to_thread is the fix -- zlib
releases the GIL while compressing, so the loop is scheduled normally.

The outage was invisible for as long as it existed. The hourly health check
fires at :35 and the outage runs 03:05-03:25, so no sample ever landed inside
it. A fixed-phase hourly probe cannot see a 20-minute event; that is aliasing,
not bad luck, and it would have stayed hidden indefinitely.

health_report.report_async joins it: psycopg2 is a blocking driver, so
reporting from the loop held it for the connect and insert -- up to
connect_timeout seconds precisely when the database is unreachable, which is
when a report matters most. Both backup executors use it now.

Caveat worth knowing: the engine wraps executors in asyncio.wait_for and a
thread cannot be cancelled, so on timeout the task is recorded failed while the
tar runs to completion. Still strictly better than blocking everything, and the
configured 3600s is well clear of the observed 1263s.

Seven tests in this file had been red since the initial commit -- they came
over with the portainer-core extraction, patched the Path class wholesale,
asserted "backed up" against a function returning "Backup completed: ...", and
one wrapped its call in except Exception: pass with its only assertion
commented out. There is no CI test gate here, so nothing reported it. Replaced
with tests that build real archives in tmp_path and assert on their contents.

The new loop test is the one that matters and it is mutation-checked: with
to_thread reverted it counts 0 heartbeat ticks, with it ~40.

Their structural demands (_create_tar_filter, a sync _cleanup_old_backups) were
adopted because threading wanted that shape anyway. Their exclude semantics
were not. The tests assert fnmatch behaviour and the deployed config is written
against substring matching -- it excludes logs as ".log", and paths as
unanchored fragments like "ollama/models/*" against members named
"docker-data/ollama/...". Under fnmatch neither matches, and the nightly
archive would silently gain many GB of model blobs instead of shrinking. That
is now pinned by tests naming the consequence, so the "improvement" fails loudly.

Co-Authored-By: Claude <noreply@anthropic.com>

src/executors/config_backup_executor.py

@@ -2,6 +2,20 @@
 Config Backup Executor
 Backs up Docker container configs and host-based service configs.
 Replicates functionality of maintenance container's backup-configs.sh
+
+The work runs in a worker thread. `execute` is awaited by the scheduling engine
+on the same event loop that serves `/health` and the whole REST API, and this
+job spends ~21 minutes inside tarfile and zlib. Called inline it starves the
+loop for that entire window: on 2026-08-11 the API was unreachable 03:05-03:25
+nightly and again at 07:39 when the job was triggered by hand, which is T-74.
+The hourly health check fires at :35 and had never once sampled the outage.
+
+`asyncio.to_thread` is the whole fix — zlib releases the GIL while compressing,
+so the loop gets scheduled normally. One caveat worth knowing: the engine wraps
+executors in `asyncio.wait_for`, and a thread cannot be cancelled. On timeout
+the task is recorded as failed while the tar keeps running to completion. That
+is still strictly better than blocking everything, and the configured timeout
+(3600s) is well clear of the observed 1263s.
 """
 import asyncio
 import logging
@@ -9,7 +23,7 @@ import tarfile
 import tempfile
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import List, Dict, Any
+from typing import Any, Callable, Dict, List, Optional
 
 from src.config import Settings
 from src.executors import health_report
@@ -17,125 +31,42 @@ from src.executors import health_report
 logger = logging.getLogger(__name__)
 
 
-async def _run(config: dict, settings: Settings) -> str:
+def _create_tar_filter(excludes: List[str]) -> Callable[[Any], Optional[Any]]:
+    """Build the tarfile filter for a set of exclude patterns.
+
+    Matching is **substring**, not glob, and that is deliberate: a pattern is
+    reduced to its literal core by dropping leading `*/` and trailing `/*`, and
+    a member is excluded when that core appears anywhere in its name.
+
+    It looks like a half-finished glob and the temptation is to "fix" it with
+    `fnmatch`. Doing so would break the deployed configuration badly, because
+    that configuration is written against these semantics:
+
+      ".log"              fnmatch would match only a file named exactly `.log`,
+                          so every log file starts being archived instead.
+      "ollama/models/*"   fnmatch anchors at the start of the name, and members
+                          are named `docker-data/ollama/models/...`, so nothing
+                          matches and many GB of model blobs enter the archive.
+
+    Same for `amp/Versions/*`, `qdrant/storage/*` and the rest — every one is an
+    unanchored mid-path fragment. The nightly archive would grow, not shrink.
+    `tests/test_config_backup_executor.py` pins both cases so this cannot be
+    changed silently.
     """
-    Execute config backup task.
+    cores = [p.replace('*/', '').replace('/*', '') for p in excludes]
 
-    Config schema:
-        {
-            "sources": [
-                {
-                    "path": "/data/docker-data",
-                    "name": "docker-data",
-                    "excludes": ["*/cache/*", "*/temp/*", "*.log"]
-                }
-            ],
-            "backup_dir": "/backups/docker-configs",
-            "retention_days": 30,
-            "compress": true
-        }
+    def tar_filter(tarinfo):
+        for core in cores:
+            if core in tarinfo.name:
+                logger.debug(f"Excluding: {tarinfo.name}")
+                return None
+        return tarinfo
 
-    Args:
-        config: Backup configuration
-        settings: Global scheduler settings
-
-    Returns:
-        Summary of backup operation
-
-    Raises:
-        Exception: On backup failure
-    """
-    sources = config.get('sources', [])
-    backup_dir = Path(config.get('backup_dir', '/backups/docker-configs'))
-    retention_days = config.get('retention_days', 30)
-    compress = config.get('compress', True)
-
-    if not sources:
-        raise ValueError("No backup sources configured")
-
-    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
-    backup_filename = f"docker-configs-{timestamp}.tar.gz"
-    backup_file = backup_dir / backup_filename
-
-    logger.info(f"Starting Docker configs backup: {backup_filename}")
-
-    # Create backup directory
-    backup_dir.mkdir(parents=True, exist_ok=True)
-
-    # Create temporary directory for staging
-    with tempfile.TemporaryDirectory(prefix='backup-') as temp_dir:
-        temp_path = Path(temp_dir)
-        results = []
-
-        # Backup each source
-        for source in sources:
-            source_path = Path(source['path'])
-            source_name = source['name']
-            excludes = source.get('excludes', [])
-
-            if not source_path.exists():
-                logger.warning(f"Source path does not exist: {source_path}")
-                continue
-
-            logger.info(f"Backing up {source_name} from {source_path}")
-
-            # Create tar for this source
-            source_tar = temp_path / f"{source_name}.tar.gz"
-
-            def tar_filter(tarinfo):
-                """Filter function to exclude patterns."""
-                for pattern in excludes:
-                    # Simple pattern matching (could be enhanced with fnmatch)
-                    if pattern.replace('*/', '').replace('/*', '') in tarinfo.name:
-                        logger.debug(f"Excluding: {tarinfo.name}")
-                        return None
-                return tarinfo
-
-            with tarfile.open(source_tar, 'w:gz') as tar:
-                tar.add(
-                    source_path,
-                    arcname=source_name,
-                    filter=tar_filter,
-                    recursive=True
-                )
-
-            source_size = source_tar.stat().st_size / (1024 * 1024)  # MB
-            results.append(f"{source_name}: {source_size:.2f}MB")
-            logger.info(f"Backed up {source_name}: {source_size:.2f}MB")
-
-        # Combine all source backups into final archive
-        logger.info("Creating combined backup archive...")
-        with tarfile.open(backup_file, 'w:gz') as final_tar:
-            for item in temp_path.glob('*.tar.gz'):
-                final_tar.add(item, arcname=item.name)
-
-    # Verify backup created
-    if not backup_file.exists():
-        raise Exception("Backup file was not created")
-
-    backup_size = backup_file.stat().st_size / (1024 * 1024)  # MB
-    logger.info(f"Backup created successfully: {backup_size:.2f}MB")
-
-    # Clean up old backups
-    await cleanup_old_backups(backup_dir, retention_days)
-
-    # Count remaining backups
-    backup_count = len(list(backup_dir.glob('docker-configs-*.tar.gz')))
-    total_size = sum(f.stat().st_size for f in backup_dir.glob('docker-configs-*.tar.gz'))
-    total_size_mb = total_size / (1024 * 1024)
-
-    output = (
-        f"Backup completed: {backup_filename} ({backup_size:.2f}MB). "
-        f"Sources: {', '.join(results)}. "
-        f"Retention: {backup_count} backups, {total_size_mb:.2f}MB total."
-    )
-
-    logger.info(output)
-    return output
+    return tar_filter
 
 
-async def cleanup_old_backups(backup_dir: Path, retention_days: int):
-    """Remove backups older than retention period."""
+def _cleanup_old_backups(backup_dir: Path, retention_days: int) -> None:
+    """Remove backups older than the retention period."""
     cutoff_date = datetime.now() - timedelta(days=retention_days)
     removed_count = 0
     removed_size = 0
@@ -143,7 +74,6 @@ async def cleanup_old_backups(backup_dir: Path, retention_days: int):
     logger.info(f"Cleaning up backups older than {retention_days} days...")
 
     for backup_file in backup_dir.glob('docker-configs-*.tar.gz'):
-        # Get file modification time
         file_mtime = datetime.fromtimestamp(backup_file.stat().st_mtime)
 
         if file_mtime < cutoff_date:
@@ -160,6 +90,101 @@ async def cleanup_old_backups(backup_dir: Path, retention_days: int):
         logger.info("No old backups to remove")
 
 
+def _backup(config: dict) -> str:
+    """The blocking body of the backup. Runs in a worker thread, never on the loop.
+
+    Config schema:
+        {
+            "sources": [
+                {
+                    "path": "/data/docker-data",
+                    "name": "docker-data",
+                    "excludes": ["*/cache/*", "*/temp/*", "*.log"]
+                }
+            ],
+            "backup_dir": "/backups/docker-configs",
+            "retention_days": 30,
+            "compress": true
+        }
+
+    Returns a one-line summary. Raises on failure.
+    """
+    sources = config.get('sources', [])
+    backup_dir = Path(config.get('backup_dir', '/backups/docker-configs'))
+    retention_days = config.get('retention_days', 30)
+
+    if not sources:
+        raise ValueError("No backup sources configured")
+
+    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
+    backup_filename = f"docker-configs-{timestamp}.tar.gz"
+    backup_file = backup_dir / backup_filename
+
+    logger.info(f"Starting Docker configs backup: {backup_filename}")
+
+    backup_dir.mkdir(parents=True, exist_ok=True)
+
+    with tempfile.TemporaryDirectory(prefix='backup-') as temp_dir:
+        temp_path = Path(temp_dir)
+        results = []
+
+        for source in sources:
+            source_path = Path(source['path'])
+            source_name = source['name']
+            excludes = source.get('excludes', [])
+
+            if not source_path.exists():
+                logger.warning(f"Source path does not exist: {source_path}")
+                continue
+
+            logger.info(f"Backing up {source_name} from {source_path}")
+
+            source_tar = temp_path / f"{source_name}.tar.gz"
+
+            with tarfile.open(source_tar, 'w:gz') as tar:
+                tar.add(
+                    source_path,
+                    arcname=source_name,
+                    filter=_create_tar_filter(excludes),
+                    recursive=True
+                )
+
+            source_size = source_tar.stat().st_size / (1024 * 1024)  # MB
+            results.append(f"{source_name}: {source_size:.2f}MB")
+            logger.info(f"Backed up {source_name}: {source_size:.2f}MB")
+
+        logger.info("Creating combined backup archive...")
+        with tarfile.open(backup_file, 'w:gz') as final_tar:
+            for item in temp_path.glob('*.tar.gz'):
+                final_tar.add(item, arcname=item.name)
+
+    if not backup_file.exists():
+        raise Exception("Backup file was not created")
+
+    backup_size = backup_file.stat().st_size / (1024 * 1024)  # MB
+    logger.info(f"Backup created successfully: {backup_size:.2f}MB")
+
+    _cleanup_old_backups(backup_dir, retention_days)
+
+    backup_count = len(list(backup_dir.glob('docker-configs-*.tar.gz')))
+    total_size = sum(f.stat().st_size for f in backup_dir.glob('docker-configs-*.tar.gz'))
+    total_size_mb = total_size / (1024 * 1024)
+
+    output = (
+        f"Backup completed: {backup_filename} ({backup_size:.2f}MB). "
+        f"Sources: {', '.join(results)}. "
+        f"Retention: {backup_count} backups, {total_size_mb:.2f}MB total."
+    )
+
+    logger.info(output)
+    return output
+
+
+async def _run(config: dict, settings: Settings) -> str:
+    """Await the backup without holding the event loop. See the module docstring."""
+    return await asyncio.to_thread(_backup, config)
+
+
 async def execute(config: dict, settings: Settings) -> str:
     """Run the backup and report its own outcome to check_history (D-33, T-69).
 
@@ -172,7 +197,7 @@ async def execute(config: dict, settings: Settings) -> str:
     try:
         output = await _run(config, settings)
     except Exception as exc:
-        health_report.report(
+        await health_report.report_async(
             settings,
             domain="backup",
             status=health_report.CRITICAL,
@@ -180,7 +205,7 @@ async def execute(config: dict, settings: Settings) -> str:
             metrics={"job": "scheduler/config_backup_executor", "error": str(exc)[:400]},
         )
         raise
-    health_report.report(
+    await health_report.report_async(
         settings,
         domain="backup",
         status=health_report.OK,

src/executors/health_report.py

@@ -28,6 +28,7 @@ Three consequences that are load-bearing here:
   broken reporter produces — so check for rows, not for errors.
 """
 
+import asyncio
 import json
 import logging
 from datetime import datetime, timezone
@@ -122,6 +123,25 @@ def report(
         conn.close()
 
 
+async def report_async(
+    settings: Any,
+    domain: str,
+    status: str,
+    source: str,
+    metrics: Optional[Dict[str, Any]] = None,
+) -> bool:
+    """`report` for callers on the event loop. Prefer this one inside executors.
+
+    psycopg2 is a blocking driver, so calling `report` directly from an
+    `async def` holds the loop for the length of the connect and insert — up to
+    `connect_timeout` seconds if the database is unreachable, which is exactly
+    when a report is most likely to be attempted. The scheduler serves its own
+    `/health` from that loop, so the cost of a slow report is the whole service
+    appearing down (T-74).
+    """
+    return await asyncio.to_thread(report, settings, domain, status, source, metrics)
+
+
 def _host() -> str:
     """The host a row is attributed to.
 

src/executors/portainer_backup_executor.py

@@ -153,7 +153,7 @@ async def execute(config: dict, settings: Settings) -> str:
     try:
         output = await _run(config, settings)
     except Exception as exc:
-        health_report.report(
+        await health_report.report_async(
             settings,
             domain="backup",
             status=health_report.CRITICAL,
@@ -161,7 +161,7 @@ async def execute(config: dict, settings: Settings) -> str:
             metrics={"job": "scheduler/portainer_backup_executor", "error": str(exc)[:400]},
         )
         raise
-    health_report.report(
+    await health_report.report_async(
         settings,
         domain="backup",
         status=health_report.OK,