diff --git a/CHANGELOG.md b/CHANGELOG.md index a716c3e..3cf3a4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `GET /processing-runs/{id}` (was: `GET /processing-runs/processing-runs/{id}`) - `GET /processing-runs/{id}/logs` (was: `GET /processing-runs/processing-runs/{id}/logs`) - **`NotificationConfigCreate` schema test failure**: `NotificationConfigBase.name` was a required field (`...`) but the unit test and the database column both use a default of `"My Notification"`. Changed the Pydantic field to `default="My Notification"` to match the DB default and allow callers to omit the field. +- **`SyntaxWarning` at startup**: Fixed invalid escape sequence `\S` in a docstring in `mail_processor.py` (changed to `\\S`). In Python 3.12+ this emits a `SyntaxWarning` and will become a `SyntaxError` in a future Python version. +- **Processing runs stuck in "running" state**: Fixed three related bugs in `tasks.py` that caused `ProcessingRun` records to remain in the `running` state indefinitely: + 1. The exception handler now calls `await db.rollback()` before attempting to write the `failed` status, ensuring the SQLAlchemy session is in a clean state even when the original exception occurred during a DB flush/commit. + 2. The error-handler `await db.commit()` is now wrapped in its own `try/except` so a commit failure inside the handler no longer propagates silently and leaves the run as `running`. + 3. `account.last_check_at` is now updated in the error path, throttling re-dispatch by `process_all_enabled_accounts` and preventing a cascade of new `running` runs on every scheduler tick. +- **Stale "running" run cleanup**: `cleanup_old_logs` now marks any `ProcessingRun` that has been in the `running` state for longer than 35 minutes (Celery hard time-limit is 30 min) as `failed` with an explanatory message. This recovers runs left behind by OOM kills, container restarts, or other SIGKILL events. ### Added - **Semantic Release** (`release.yml`): Automated versioning and GitHub Release creation on every push to `main` using `python-semantic-release`. Reads conventional-commit prefixes (`feat:`, `fix:`, etc.) to determine the next version and updates `CHANGELOG.md`. diff --git a/backend/app/services/mail_processor.py b/backend/app/services/mail_processor.py index 144e6a5..dccca9b 100644 --- a/backend/app/services/mail_processor.py +++ b/backend/app/services/mail_processor.py @@ -263,7 +263,7 @@ class MailProcessor: async def _fetch_imap_emails( self, max_count: int, already_seen_uids: Set[str] ) -> Tuple[List[bytes], List[str]]: - """Fetch emails via IMAP, marking each message \Seen to prevent re-fetch.""" + """Fetch emails via IMAP, marking each message \\Seen to prevent re-fetch.""" emails: List[bytes] = [] new_uids: List[str] = [] diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index 1db3cb6..1848958 100644 --- a/backend/app/workers/tasks.py +++ b/backend/app/workers/tasks.py @@ -400,7 +400,13 @@ async def process_mail_account(account_id: int): task_name="process_mail_account" ).observe(_task_duration) - # Mark run as failed + # Mark run as failed – roll back any pending/broken transaction first + # so the session is in a clean state before we write the failure status. + try: + await db.rollback() + except Exception as rb_exc: + logger.warning(f"Rollback failed during error handler: {rb_exc}") + if "run" in locals(): run.status = "failed" # type: ignore[assignment] run.error_message = str(e) # type: ignore[assignment] @@ -414,6 +420,9 @@ async def process_mail_account(account_id: int): account.status = AccountStatus.ERROR # type: ignore[assignment] account.last_error_at = datetime.now(timezone.utc) # type: ignore[assignment] account.last_error_message = str(e) # type: ignore[assignment] + # Always update last_check_at so process_all_enabled_accounts + # throttles re-dispatch instead of queuing a new task every cycle. + account.last_check_at = datetime.now(timezone.utc) # type: ignore[assignment] # Notify user about the error try: @@ -429,7 +438,13 @@ async def process_mail_account(account_id: int): f"Failed to send error notification: {notify_exc}" ) - await db.commit() + try: + await db.commit() + except Exception as commit_exc: + logger.error( + f"Failed to persist failed status for run of account " + f"{account_id}: {commit_exc}" + ) @celery_app.task(base=AsyncTask, name="app.workers.tasks.process_all_enabled_accounts") @@ -492,6 +507,8 @@ async def process_all_enabled_accounts(): async def cleanup_old_logs(days_to_keep: int = 30): """ Clean up old processing logs, runs, and downloaded message ID records. + Also marks stale "running" runs (older than the Celery task time-limit) + as "failed" to recover from worker crashes or SIGKILL events. Args: days_to_keep: Number of days of data to retain @@ -501,6 +518,32 @@ async def cleanup_old_logs(days_to_keep: int = 30): try: cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_to_keep) + # Mark stale "running" runs as "failed". + # A run is considered stale when it has been in the "running" state + # for longer than the Celery hard time limit (30 min) plus a small + # buffer – meaning the worker was likely killed before it could write + # the final status (OOM kill, container restart, SIGKILL, etc.). + stale_threshold = datetime.now(timezone.utc) - timedelta(minutes=35) + stale_result = await db.execute( + select(ProcessingRun).where( + ProcessingRun.status == "running", + ProcessingRun.started_at < stale_threshold, + ) + ) + stale_runs = stale_result.scalars().all() + for stale_run in stale_runs: + stale_run.status = "failed" # type: ignore[assignment] + stale_run.error_message = ( # type: ignore[assignment] + "Run timed out or worker was killed before completion" + ) + stale_run.completed_at = datetime.now(timezone.utc) # type: ignore[assignment] + stale_run.duration_seconds = ( # type: ignore[assignment] + stale_run.completed_at - _as_utc(stale_run.started_at) + ).total_seconds() + + if stale_runs: + logger.info(f"Marked {len(stale_runs)} stale processing runs as failed") + # Delete old processing runs result = await db.execute( select(ProcessingRun).where(ProcessingRun.started_at < cutoff_date)