"""Enfilage, lecture et progrès des jobs (spec 2026-08-31 « socle jobs » §1 et §3). Jamais de SELECT * : l'ORM émet la liste explicite des colonnes mappées. Toute lecture exposée à l'api est filtrée par `org_id` ; le worker, lui, est interne et traite toutes les organisations (il n'utilise pas `get_job`). """ import logging import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import set_committed_value from app.core.errors import AppError from app.jobs import registry from app.jobs.models import Job logger = logging.getLogger(__name__) # Course résiduelle bornée (spec §1) : le gagnant peut finir — et relâcher sa clé — entre # le conflit d'insertion et la relecture. On ré-insère au lieu de rendre une erreur. MAX_ENQUEUE_ATTEMPTS = 3 async def _get(session: AsyncSession, job_id: uuid.UUID) -> Job: job = await session.scalar(select(Job).where(Job.id == job_id)) assert job is not None # la ligne vient d'être insérée dans cette transaction return job async def _find_by_dedup_key(session: AsyncSession, org_id: str, dedup_key: str) -> Job | None: # Filtré par org_id : la clé est unique *globalement*, donc une clé mal construite qui # collisionne entre deux organisations ne doit jamais rendre le job de l'autre org. return await session.scalar(select(Job).where(Job.dedup_key == dedup_key, Job.org_id == org_id)) async def enqueue( session: AsyncSession, org_id: str, type_: str, *, payload: dict[str, Any] | None = None, dedup_key: str | None = None, ) -> Job: """Enfile un job, ou rend celui qui occupe déjà la `dedup_key` (anti-empilement). Sans `dedup_key`, l'insertion est toujours acceptée : la colonne est nullable et Postgres considère les NULL comme distincts dans un index UNIQUE. **Contrat : cette fonction committe la session de l'appelant** — le job doit être visible du worker, qui est un autre process. Un handler d'api qui écrit sa propre ligne (par exemple `imports`) puis enfile ne peut donc pas rendre les deux atomiques : le premier commit publie aussi son travail en cours, y compris sur le chemin de conflit où aucun job n'est inséré. Écrire la ligne métier en dernier, ou l'accepter explicitement. """ if registry.get(type_) is None: # Erreur de programmation, pas une entrée utilisateur : jamais un Problem Details. raise ValueError(f"type de job inconnu : {type_}") values = {"org_id": org_id, "type": type_, "payload": payload or {}, "dedup_key": dedup_key} for _ in range(MAX_ENQUEUE_ATTEMPTS): job_id = await session.scalar( pg_insert(Job) .values(**values) .on_conflict_do_nothing(index_elements=["dedup_key"]) .returning(Job.id) ) if job_id is not None: await session.commit() return await _get(session, job_id) # ON CONFLICT DO NOTHING n'avorte pas la transaction : elle reste utilisable. await session.commit() assert dedup_key is not None # sans clé, l'insertion ne peut pas entrer en conflit existing = await _find_by_dedup_key(session, org_id, dedup_key) if existing is not None: return existing raise AppError(409, "Conflict", "Enfilage impossible : clé de déduplication instable") async def enqueue_pending( session: AsyncSession, org_id: str, type_: str, *, payload: dict[str, Any] | None = None, ) -> Job: """Insère le job **sans committer** : c'est l'appelant qui publiera, d'un seul geste. `enqueue` committe parce qu'un job doit être visible d'un autre process ; c'est vrai, mais ce n'est pas le *commit* qui appartient au socle, c'est la *publication*. Un module qui écrit sa propre ligne métier (`imports`) en même temps que son job a besoin des deux dans la même transaction : sinon un arrêt entre les deux commits laisse un job qui s'exécutera et une ligne métier qui n'existera jamais — l'import fantôme (réservoir peuplé, rien dans l'onglet Imports, aucun rapport visible). Avec cette variante, le commit unique de l'appelant publie job + ligne métier ensemble, et la promesse « visible du worker » reste tenue : c'est le commit qui rend visible, pas l'insertion. **Pas de `dedup_key` ici.** L'anti-empilement a besoin d'observer le gagnant d'une course, donc de committer pour voir la ligne d'une autre transaction (voir la boucle d'`enqueue`). Un appelant qui a besoin de déduplication passe par `enqueue` et accepte ses deux commits. """ if registry.get(type_) is None: # Erreur de programmation, pas une entrée utilisateur : jamais un Problem Details. raise ValueError(f"type de job inconnu : {type_}") job = Job(org_id=org_id, type=type_, payload=payload or {}) session.add(job) # Flush et non commit : l'id (uuidv7 côté serveur) est relu tout de suite, la ligne reste # invisible des autres transactions jusqu'au commit de l'appelant. await session.flush() return job async def get_job(session: AsyncSession, org_id: str, job_id: uuid.UUID) -> Job: job = await session.scalar(select(Job).where(Job.id == job_id, Job.org_id == org_id)) if job is None: # 404 et non 403 : un job d'une autre org n'existe pas (architecture §9). raise AppError(404, "Not Found", "Job inconnu") return job async def save_progress(session: AsyncSession, job: Job, progress: dict[str, Any]) -> bool: """Écrit le progrès du job **et son heartbeat**, committé aussitôt. Le heartbeat `updated_at` est posé ici, jamais par le handler : c'est lui que le balayage des zombies regarde, et un format libre le rendrait incomparable. À appeler par lot ou par étape, jamais à chaque ligne. **Fencé par le même jeton que `worker.finish`** (`status='running'` + l'`attempts` du claim, porté par l'instance) : un worker que le sweep a dépossédé écraserait sinon le curseur de reprise de son repreneur, et — plus grave — rafraîchirait le heartbeat. Si le repreneur mourait vraiment, le sweep croirait le job vivant tant que le dépossédé écrit : bloqué en `running` pour toujours, `dedup_key` `'active'` retenue avec lui. **Rend `False` quand le job a été repris : le handler doit s'arrêter là.** Le verdict est rendu **avant le commit** — `rowcount` est préfetché à l'exécution — et le lot en vol est alors annulé : c'est le seul point où le socle peut encore jeter le travail d'un handler dépossédé, et la promesse doit être vraie, sans quoi un auteur de handler pourrait se permettre un effet de bord non idempotent dans ce lot. La valeur écrite est reposée sur l'instance **sans relecture** : un `refresh` rouvrirait une transaction qui resterait « idle in transaction » pendant tout le lot suivant, à tenir le xmin du vacuum pendant le travail du handler. `set_committed_value` évite en prime de marquer l'instance sale — aucun UPDATE fantôme au flush suivant. Comme tout le socle, cela suppose une session `expire_on_commit=False` (celle du worker l'est). """ # Capturés avant tout `rollback()` : celui-ci expire les attributs de l'instance, et # les relire déclencherait un chargement paresseux impossible en asynchrone. job_id, attempts = job.id, job.attempts data = {**progress, "updated_at": datetime.now(UTC).isoformat()} result = await session.execute( update(Job) .where(Job.id == job_id, Job.status == "running", Job.attempts == attempts) .values(progress=data) .execution_options(synchronize_session=False) ) if result.rowcount == 0: # Avant le commit : le lot métier en vol du handler dépossédé part avec. await session.rollback() logger.warning( "job %s : progrès ignoré et lot annulé (jeton attempts=%s) — le job a été " "repris, le handler doit s'arrêter", job_id, attempts, ) return False await session.commit() set_committed_value(job, "progress", data) return True