diff --git a/app/celery_worker.py b/app/celery_worker.py index 5531a86e..625884d0 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -33,6 +33,7 @@ from app.tasks.upload_to_s3 import upload_to_s3 # noqa: F401 from app.tasks.upload_to_sftp import upload_to_sftp # noqa: F401 from app.tasks.upload_to_webdav import upload_to_webdav # noqa: F401 from app.tasks.uptime_kuma_tasks import ping_uptime_kuma # noqa: F401 +from app.tasks.monitor_stalled_steps import monitor_stalled_steps # noqa: F401 celery.conf.task_routes = { "app.tasks.*": {"queue": "default"}, @@ -79,6 +80,12 @@ celery.conf.beat_schedule = { "schedule": crontab(hour="0", minute="0"), # Midnight "options": {"expires": 3600}, # 1 hour expiry }, + # Monitor for stalled processing steps every minute + "monitor-stalled-steps": { + "task": "app.tasks.monitor_stalled_steps.monitor_stalled_steps", + "schedule": crontab(minute="*/1"), # Every minute + "options": {"expires": 55}, # Must complete within 55 seconds + }, } # Remove None entries from beat_schedule diff --git a/app/config.py b/app/config.py index e54d160f..c96099ae 100644 --- a/app/config.py +++ b/app/config.py @@ -189,6 +189,12 @@ class Settings(BaseSettings): description="Show the 'Check for Duplicates' step in processing history. If False, the check is still performed but not displayed. Default: True.", ) + # Processing step timeout - prevents files from getting stuck in "in_progress" state + step_timeout: int = Field( + default=600, + description="Timeout in seconds for processing steps. If a step is 'in_progress' for longer than this, it will be marked as failed. Default: 600 seconds (10 minutes).", + ) + # Security Headers Configuration (see SECURITY_AUDIT.md and docs/DeploymentGuide.md) # Disabled by default since most deployments use a reverse proxy (Traefik, Nginx, etc.) # that already adds these headers. Enable if deploying directly without a reverse proxy. diff --git a/app/tasks/monitor_stalled_steps.py b/app/tasks/monitor_stalled_steps.py new file mode 100644 index 00000000..e63ab303 --- /dev/null +++ b/app/tasks/monitor_stalled_steps.py @@ -0,0 +1,54 @@ +""" +Periodic task to detect and recover from stalled processing steps. + +This task runs periodically (every minute by default) to find any processing steps +that have been stuck in "in_progress" state for too long and mark them as failed. +""" + +import logging +from datetime import datetime + +from app.celery_app import celery +from app.database import SessionLocal +from app.utils.step_timeout import mark_stalled_steps_as_failed + +logger = logging.getLogger(__name__) + + +@celery.task(name="app.tasks.monitor_stalled_steps.monitor_stalled_steps") +def monitor_stalled_steps(): + """ + Periodic task to detect and mark stalled processing steps as failed. + + This task: + 1. Connects to the database + 2. Finds any in-progress steps that exceeded the timeout + 3. Marks them as failed with a timeout error message + 4. Logs the recovery action + + This helps prevent files from getting stuck in "pending" state when + processing crashes or hangs without proper error handling. + + Scheduled to run every minute via Celery Beat. + """ + try: + with SessionLocal() as db: + stalled_count = mark_stalled_steps_as_failed(db) + + if stalled_count > 0: + logger.warning( + f"[{datetime.utcnow().isoformat()}] " + f"Recovered {stalled_count} stalled step(s). " + f"Marked as failed due to timeout." + ) + else: + logger.debug( + f"[{datetime.utcnow().isoformat()}] " + f"No stalled steps found." + ) + + return {"recovered": stalled_count} + + except Exception as e: + logger.error(f"Error in monitor_stalled_steps task: {e}", exc_info=True) + return {"error": str(e), "recovered": 0} diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index cb645d79..7279a9c8 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -110,8 +110,13 @@ def process_document(self, original_local_file: str, original_filename: str = No ) new_record = existing_record else: + # Check for duplicate only if this is a new file (not reprocessing) + # IMPORTANT: Only consider it a duplicate if it matches a DIFFERENT file existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none() - if existing and settings.enable_deduplication: + + # A file is only a duplicate if it matches a different file's hash + # (not its own hash when reprocessing) + if existing and existing.id != file_id and settings.enable_deduplication: logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") # Log the deduplication result without creating a new database record # This avoids UNIQUE constraint violations on filehash diff --git a/app/utils/step_timeout.py b/app/utils/step_timeout.py new file mode 100644 index 00000000..484a2a45 --- /dev/null +++ b/app/utils/step_timeout.py @@ -0,0 +1,114 @@ +""" +Step timeout detection and handling. + +Monitors in-progress steps and marks them as failed if they exceed a timeout threshold. +This prevents files from getting stuck in "pending" state when processing crashes. +""" + +import logging +from datetime import datetime, timedelta +from typing import Optional + +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import FileProcessingStep + +logger = logging.getLogger(__name__) + +# Default timeout for a step (in seconds) +DEFAULT_STEP_TIMEOUT = 600 # 10 minutes + + +def get_step_timeout() -> int: + """ + Get the step timeout configuration from settings. + + Returns: + Timeout in seconds (default: 600 seconds / 10 minutes) + """ + # Could be made configurable via settings in the future + return getattr(settings, "step_timeout", DEFAULT_STEP_TIMEOUT) + + +def mark_stalled_steps_as_failed( + db: Session, + timeout_seconds: Optional[int] = None, + file_id: Optional[int] = None +) -> int: + """ + Find and mark any in-progress steps that have exceeded the timeout as failed. + + This function: + 1. Queries for all in-progress steps + 2. Checks if they've been running longer than the timeout + 3. Marks them as failed with a timeout error message + 4. Sets their completed_at timestamp + + Args: + db: SQLAlchemy session + timeout_seconds: Timeout duration in seconds (default: from settings) + file_id: Optional file ID to check only that file's steps + + Returns: + Number of steps marked as failed + """ + if timeout_seconds is None: + timeout_seconds = get_step_timeout() + + now = datetime.utcnow() + cutoff_time = now - timedelta(seconds=timeout_seconds) + + # Query for stalled steps + query = db.query(FileProcessingStep).filter( + FileProcessingStep.status == "in_progress", + FileProcessingStep.started_at <= cutoff_time, # Started before cutoff + FileProcessingStep.started_at.isnot(None), # Has a start time + ) + + if file_id is not None: + query = query.filter(FileProcessingStep.file_id == file_id) + + stalled_steps = query.all() + + if not stalled_steps: + return 0 + + logger.warning( + f"Found {len(stalled_steps)} stalled step(s) that exceeded " + f"{timeout_seconds}s timeout. Marking as failed." + ) + + count = 0 + for step in stalled_steps: + step.status = "failure" + step.completed_at = now + step.error_message = ( + f"Step timeout after {timeout_seconds} seconds. " + f"Processing did not complete. Started at {step.started_at}, " + f"timeout triggered at {now}." + ) + count += 1 + + logger.error( + f"[File {step.file_id}] Step '{step.step_name}' marked as failed due to timeout. " + f"Started: {step.started_at}, Timeout at: {now}" + ) + + db.commit() + return count + + +def check_and_recover_stalled_file(db: Session, file_id: int) -> bool: + """ + Check if a specific file has stalled steps and recover by marking them as failed. + + Args: + db: SQLAlchemy session + file_id: File ID to check + + Returns: + True if stalled steps were found and marked as failed, False otherwise + """ + count = mark_stalled_steps_as_failed(db, file_id=file_id) + return count > 0