Coverage for app/services/reconciler.py: 25.00%
104 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1"""Background reconciler for stuck Celery tasks.
3The Celery event listener is the primary path for syncing task state
4into the DB, but events can be lost (backend restart, dropped message,
5worker death mid-handler), leaving tasks stuck in PENDING/RUNNING.
7This reconciler runs as a coroutine inside the FastAPI lifespan. Every
830 seconds it:
101. Selects all tasks in PENDING or RUNNING.
112. Reconciles each against its Celery AsyncResult.
123. Flips tasks with ``celeryTaskId IS NULL`` older than the grace
13 window to FAILED (dispatch never completed).
15Multi-instance safety: a Postgres session-scoped advisory lock gates
16each pass, so only one FastAPI process runs the reconcile body per
17tick. The pass is idempotent regardless.
18"""
19from __future__ import annotations
21import asyncio
22import json
23import logging
24from datetime import timedelta
26from celery.result import AsyncResult
27from sqlalchemy import text
28from sqlalchemy.orm import Session
30from app.celery_app import celery_app
31from app.crud import locks as crud_locks
32from app.crud import tasks as crud_tasks
33from app.database import SessionLocal
34from app.models import Task, TaskStatus
35from app.utils.time import utcnow
37logger = logging.getLogger(__name__)
40# Constant key for the advisory lock. Distinct from the per-user lock
41# helper's hashtext()-derived range.
42RECONCILER_ADVISORY_LOCK_KEY = 4242424242
44RECONCILE_INTERVAL_SECONDS = 30
46# Grace window for a celery_id-NULL task (dispatch committed the row
47# but send_task never stamped an id). Older than this → fair game.
48DISPATCH_GRACE_SECONDS = 60
50# Grace window for a Celery PENDING state with a known celery_id (in
51# RabbitMQ but no worker picked it up). Older → treated as lost.
52CELERY_PENDING_GRACE_SECONDS = 3600
55async def run_reconciler() -> None:
56 """Top-level coroutine started from the FastAPI lifespan."""
57 logger.info(
58 "Reconciler loop starting (interval=%ss)",
59 RECONCILE_INTERVAL_SECONDS,
60 )
61 try:
62 while True:
63 try:
64 await asyncio.to_thread(_reconcile_once)
65 except asyncio.CancelledError:
66 raise
67 except Exception:
68 logger.exception("Reconciler pass failed; will retry next tick")
69 await asyncio.sleep(RECONCILE_INTERVAL_SECONDS)
70 except asyncio.CancelledError:
71 logger.info("Reconciler loop cancelled — shutting down")
72 raise
75def _reconcile_once() -> None:
76 """One reconciliation pass. Sync because SessionLocal is sync."""
77 db: Session = SessionLocal()
78 try:
79 if not _try_acquire_lock(db):
80 logger.debug("Reconciler skipped: another instance holds the lock")
81 return
83 try:
84 stuck = (
85 db.query(Task)
86 .filter(Task.status.in_((TaskStatus.PENDING, TaskStatus.RUNNING)))
87 .all()
88 )
89 if stuck:
90 logger.debug("Reconciler examining %d stuck task(s)", len(stuck))
91 for task in stuck:
92 try:
93 _reconcile_task(db, task)
94 except Exception:
95 logger.exception(
96 "Failed to reconcile task %s — continuing",
97 task.taskId,
98 )
99 finally:
100 _release_lock(db)
101 finally:
102 db.close()
105def _try_acquire_lock(db: Session) -> bool:
106 row = db.execute(
107 text("SELECT pg_try_advisory_lock(:k)"),
108 {"k": RECONCILER_ADVISORY_LOCK_KEY},
109 ).scalar()
110 # The first SELECT opens an implicit TX; commit it so the read
111 # doesn't tie up the connection. (Lock is session-scoped.)
112 db.commit()
113 return bool(row)
116def _release_lock(db: Session) -> None:
117 db.execute(
118 text("SELECT pg_advisory_unlock(:k)"),
119 {"k": RECONCILER_ADVISORY_LOCK_KEY},
120 )
121 db.commit()
124def _reconcile_task(db: Session, task: Task) -> None:
125 now = utcnow()
127 # Per-deployment advisory lock — serialises against the request
128 # handlers so the reconciler's "stuck task → FAILED" decision can't
129 # race a handler's status read. Xact-scoped; released on the first
130 # ``update_task`` commit below.
131 crud_locks.acquire_deployment_xact_lock(db, task.deploymentId)
133 if task.celeryTaskId is None:
134 age = now - task.created_at if task.created_at else timedelta(0)
135 if age >= timedelta(seconds=DISPATCH_GRACE_SECONDS):
136 logger.warning(
137 "Marking task %s FAILED: celery_id NULL after %ss",
138 task.taskId,
139 int(age.total_seconds()),
140 )
141 crud_tasks.update_task(db, task.taskId, {
142 "status": TaskStatus.FAILED,
143 "finished_at": now,
144 "logs": "Reconciler: Celery dispatch did not complete (no task id stamped)",
145 })
146 return
148 async_result = AsyncResult(task.celeryTaskId, app=celery_app)
149 state = async_result.state # PENDING / STARTED / SUCCESS / FAILURE / REVOKED ...
151 if state == "SUCCESS":
152 logs_str, tf_state_str, outputs_str = _extract_success_payload(async_result.result)
153 crud_tasks.update_task(db, task.taskId, {
154 "status": TaskStatus.SUCCESS,
155 "finished_at": now,
156 "logs": logs_str,
157 "tf_state": tf_state_str,
158 "outputs": outputs_str,
159 })
160 logger.info("Reconciled task %s -> SUCCESS", task.taskId)
162 elif state == "FAILURE":
163 crud_tasks.update_task(db, task.taskId, {
164 "status": TaskStatus.FAILED,
165 "finished_at": now,
166 "logs": _extract_failure_logs(async_result),
167 })
168 logger.info("Reconciled task %s -> FAILED", task.taskId)
170 elif state == "REVOKED":
171 crud_tasks.update_task(db, task.taskId, {
172 "status": TaskStatus.CANCELLED,
173 "finished_at": now,
174 })
175 logger.info("Reconciled task %s -> CANCELLED", task.taskId)
177 elif state == "STARTED" and task.status == TaskStatus.PENDING:
178 # task-started event was lost; promote PENDING -> RUNNING.
179 crud_tasks.update_task(db, task.taskId, {
180 "status": TaskStatus.RUNNING,
181 "started_at": task.started_at or now,
182 })
183 logger.info("Reconciled task %s -> RUNNING (task-started event missed)", task.taskId)
185 elif state == "PENDING":
186 age = now - task.created_at if task.created_at else timedelta(0)
187 if age >= timedelta(seconds=CELERY_PENDING_GRACE_SECONDS):
188 logger.warning(
189 "Marking task %s FAILED: stuck in Celery PENDING for %ss",
190 task.taskId,
191 int(age.total_seconds()),
192 )
193 crud_tasks.update_task(db, task.taskId, {
194 "status": TaskStatus.FAILED,
195 "finished_at": now,
196 "logs": "Reconciler: task stuck in Celery PENDING beyond grace window",
197 })
200def _extract_success_payload(result):
201 """Mirror the success-event handler's shape so reconciled rows
202 look identical to event-driven rows."""
203 if not isinstance(result, dict):
204 return None, None, None
205 logs_data = result.get("logs")
206 tf_state = result.get("tf_state")
207 outputs = result.get("terraform_outputs")
208 logs_str = (
209 json.dumps(logs_data, ensure_ascii=False) if isinstance(logs_data, list)
210 else logs_data if isinstance(logs_data, str)
211 else None
212 )
213 tf_state_str = (
214 tf_state if isinstance(tf_state, str)
215 else (json.dumps(tf_state) if tf_state else None)
216 )
217 outputs_str = (
218 outputs if isinstance(outputs, str)
219 else (json.dumps(outputs) if outputs else None)
220 )
221 return logs_str, tf_state_str, outputs_str
224def _extract_failure_logs(async_result: AsyncResult) -> str:
225 try:
226 info = async_result.info
227 if isinstance(info, BaseException):
228 return f"Task failed: {type(info).__name__}: {str(info)[:500]}"
229 return f"Task failed: {str(info)[:500]}"
230 except Exception:
231 return "Task failed (reconciler could not retrieve traceback)"