gitea-admin/gh-christianlouis-inboxconverge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #104 from christianlouis/copilot/update-processing-log-entries

Browse Source 
Fix processing log entries: stale runs, error handler crash, and duration display bug
This commit is contained in:
Christian Krakau-Louis
2026-03-28 12:30:39 +01:00
committed by GitHub
parent f821cadbbf 88fac1d8dd
commit 526824de05
3 changed files with 61 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/workers/tasks.py
+41 -3
View File
@@ -90,10 +90,15 @@ async def process_mail_account(account_id: int):
logger.warning(f"Account {account_id} not found or disabled")
return
# Capture start time in a local variable so the error handler can
# compute duration_seconds without touching the (expired) ORM
# attribute after a session rollback.
_run_started_at = datetime.now(timezone.utc)
# Create processing run
run = ProcessingRun(
mail_account_id=account.id,
started_at=datetime.now(timezone.utc),
started_at=_run_started_at,
status="running",
)
db.add(run)
@@ -189,6 +194,11 @@ async def process_mail_account(account_id: int):
)
run.status = "failed" # type: ignore[assignment]
run.error_message = "No delivery method configured (SMTP credentials missing and Gmail API not set up)" # type: ignore[assignment]
run.completed_at = datetime.now(timezone.utc) # type: ignore[assignment]
run.duration_seconds = ( # type: ignore[assignment]
run.completed_at - _run_started_at
).total_seconds()
account.last_check_at = datetime.now(timezone.utc) # type: ignore[assignment]
await db.commit()
return
@@ -411,8 +421,10 @@ async def process_mail_account(account_id: int):
run.status = "failed" # type: ignore[assignment]
run.error_message = str(e) # type: ignore[assignment]
run.completed_at = datetime.now(timezone.utc) # type: ignore[assignment]
run.duration_seconds = (
run.completed_at - _as_utc(run.started_at)
# Use the locally-captured start time to avoid accessing an
# expired ORM attribute after the session rollback above.
run.duration_seconds = ( # type: ignore[assignment]
run.completed_at - _run_started_at
).total_seconds()
# Update account error status
@@ -456,6 +468,32 @@ async def process_all_enabled_accounts():
_task_start = time.monotonic()
async with async_session_maker() as db:
try:
# Mark stale "running" runs as failed. A run is considered stale
# when it has been in the "running" state longer than the Celery
# hard time limit (30 min) plus a small buffer – this recovers from
# worker crashes or SIGKILL events faster than waiting for the
# daily cleanup_old_logs task.
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:
await db.commit()
logger.info(f"Marked {len(stale_runs)} stale processing runs as failed")
# Fetch all enabled accounts regardless of operational status so
# that accounts in ERROR state are retried automatically.
result = await db.execute(