Fix Celery task result Redis connection errors (v2.5.53)

Add safe_task_ready() and safe_task_get() helper functions with retry
logic to handle Redis connection errors in Celery's internal backend.

The previous fix (v2.5.51) addressed application-level Redis connections
but Celery's task.ready() and task.get() methods use their own internal
Redis connection which can still fail with connection reset errors.

This fix wraps those Celery methods with retry logic and graceful
error handling to prevent integrity scan crashes.

Author: ttlequals0

CHANGELOG.MD

@@ -5,6 +5,23 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0).
 
+## [2.5.53] - 2026-01-01
+
+### Bug Fixes
+
+- **Fix Celery task result Redis connection errors**: Integrity scan crashed when checking task results
+  - Root cause: Celery's `task.ready()` and `task.get()` use their own internal Redis connection, separate from our fixed Redis client
+  - When Celery's Redis connection gets reset, task result checking fails with `ConnectionResetError`
+  - Error message: `'Connection' object has no attribute 'register_connect_callback'`
+  - **Solution**: Added wrapper functions with retry logic for Celery task result operations
+    - `safe_task_ready()`: Safely check task status with 3 retries and 0.5s delay
+    - `safe_task_get()`: Safely get task result with 3 retries and 0.5s delay
+    - Both functions catch `redis.ConnectionError`, `redis.TimeoutError`, `ConnectionResetError`, and `AttributeError`
+    - On transient failure, retries before giving up
+  - Files affected: `pixelprobe/services/maintenance_service.py`
+
+---
+
 ## [2.5.52] - 2026-01-01
 
 ### Bug Fixes

pixelprobe/services/maintenance_service.py

@@ -17,6 +17,70 @@
 
 logger = logging.getLogger(__name__)
 
+
+def safe_task_ready(task, max_retries=3, retry_delay=0.5):
+    """
+    Safely check if a Celery task is ready with retry logic for Redis connection errors.
+
+    Celery's task.ready() uses its internal Redis connection which can get reset.
+    This wrapper adds retry logic to handle transient connection failures.
+
+    Args:
+        task: Celery AsyncResult object
+        max_retries: Number of retry attempts
+        retry_delay: Delay between retries in seconds
+
+    Returns:
+        bool: True if task is ready, False if not ready or on persistent error
+    """
+    import redis
+
+    for attempt in range(max_retries):
+        try:
+            return task.ready()
+        except (redis.ConnectionError, redis.TimeoutError, ConnectionResetError, AttributeError) as e:
+            if attempt < max_retries - 1:
+                logger.warning(f"Redis connection error checking task status (attempt {attempt + 1}/{max_retries}): {e}")
+                time.sleep(retry_delay)
+            else:
+                logger.error(f"Failed to check task status after {max_retries} attempts: {e}")
+                # Return False to keep task in active list and retry later
+                return False
+        except Exception as e:
+            logger.error(f"Unexpected error checking task status: {e}")
+            return False
+    return False
+
+
+def safe_task_get(task, timeout=1, max_retries=3, retry_delay=0.5):
+    """
+    Safely get a Celery task result with retry logic for Redis connection errors.
+
+    Args:
+        task: Celery AsyncResult object
+        timeout: Timeout for getting the result
+        max_retries: Number of retry attempts
+        retry_delay: Delay between retries in seconds
+
+    Returns:
+        Result dict or None on error
+    """
+    import redis
+
+    for attempt in range(max_retries):
+        try:
+            return task.get(timeout=timeout)
+        except (redis.ConnectionError, redis.TimeoutError, ConnectionResetError, AttributeError) as e:
+            if attempt < max_retries - 1:
+                logger.warning(f"Redis connection error getting task result (attempt {attempt + 1}/{max_retries}): {e}")
+                time.sleep(retry_delay)
+            else:
+                logger.error(f"Failed to get task result after {max_retries} attempts: {e}")
+                raise
+        except Exception:
+            raise
+    return None
+
 class MaintenanceService:
     """Service for maintenance operations like cleanup and file monitoring"""
 
@@ -326,9 +390,9 @@ def _run_cleanup(self, cleanup_id, file_paths=None, schedule_id=None):
                 # Collect completed tasks and free up slots
                 still_active = []
                 for task in active_tasks:
-                    if task.ready():
+                    if safe_task_ready(task):
                         try:
-                            check_result = task.get(timeout=1)
+                            check_result = safe_task_get(task, timeout=1)
                             total_files_processed += 1
 
                             if not check_result.get('exists'):
@@ -717,9 +781,9 @@ def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=No
                 still_active = []
                 for task_info in active_tasks:
                     task, file_size = task_info['task'], task_info['size']
-                    if task.ready():
+                    if safe_task_ready(task):
                         try:
-                            result = task.get(timeout=1)
+                            result = safe_task_get(task, timeout=1)
                             total_files_processed += 1
 
                             # Update last integrity check timestamp for this file

version.py

@@ -4,7 +4,7 @@
 # Default version - this is the single source of truth
 
 
-_DEFAULT_VERSION = '2.5.52'
+_DEFAULT_VERSION = '2.5.53'
 
 
 # Allow override via environment variable for CI/CD, but default to the hardcoded version