fix(geocoder): fuite de clé API, réservations atomiques, arrêt propre

La clé Geoapify part en query string. Toute exception requests — timeout,
raise_for_status, ConnectionError — embarque l'URL complète, donc la clé. Ce
message était écrit tel quel dans band_locations.geocode_error, que la vue
Localisations de l'admin affiche. Les exceptions sont remplacées par une erreur
sans URL, avec un masquage en filet.

Le groq-worker ne rattrapait que RuntimeError. Or requests lève des exceptions
héritant d'OSError : un 500 Groq ou un timeout s'échappait de la boucle et
tuait le process, que restart: unless-stopped relançait pour remourir aussitôt.

Les trois daemons faisaient `SELECT … FOR UPDATE SKIP LOCKED` avec
autocommit = True. Chaque instruction étant sa propre transaction, le verrou
tombait immédiatement : la clause ne protégeait rien. Sans conséquence avec une
réplique unique, mais toute mise à l'échelle aurait produit des appels API
payants en double. La réservation tient désormais en un seul UPDATE … RETURNING.

Autres correctifs :

- Une ligne passée en 'processing' puis abandonnée (redeploy, OOM) y restait
  pour toujours : la file ne sélectionne que 'queued' et aucune route admin ne
  visait ce statut. Récupération au démarrage et pendant les temps morts.
- is_reliable() acceptait n'importe quel résultat pour un lieu « pays seul »,
  y compris sans confiance. Or une telle ligne n'arrive au worker que si le
  centroïde a échoué — précisément quand il ne faut pas faire confiance.
- resolve_iso2() renvoyait n'importe quel code à deux lettres, "XX" compris.
- Alias non standards fréquents chez MA résolus (UK, Great Britain, Holland,
  Czechia). COUNTRY_NAMES étant dérivé par inversion, où le dernier libellé
  gagne, 'NL' est explicitement réimposé à « Netherlands ».
- max_tokens Groq relevé de 80 à 120 : la réponse JSON était parfois tronquée
  à la source, ce que le repli de parsing ne peut pas réparer.
- Arrêt propre sur SIGTERM/SIGINT et reconnexion DB de l'enqueue.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Nicolas FRYDER 2026-08-20 16:44:07 +02:00
parent aa4ef68dd8
commit cc6be5e341
4 changed files with 307 additions and 74 deletions

View file

@ -17,10 +17,12 @@ Tourne en continu (restart: unless-stopped), poll toutes les 15s.
""" """
import os import os
import signal
import time import time
import psycopg2 import psycopg2
from parser import ( from parser import (
COUNTRY_CENTROIDS,
COUNTRY_NAME_TO_ISO2, COUNTRY_NAME_TO_ISO2,
country_centroid, country_centroid,
parse_location_text, parse_location_text,
@ -30,16 +32,42 @@ POLL_INTERVAL = int(os.environ.get("GEOCODE_ENQUEUE_POLL", "15"))
BATCH = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000")) BATCH = int(os.environ.get("GEOCODE_ENQUEUE_BATCH", "100000"))
AUTO_INTERVAL_MIN = int(os.environ.get("ENQUEUE_AUTO_INTERVAL_MIN", "60")) AUTO_INTERVAL_MIN = int(os.environ.get("ENQUEUE_AUTO_INTERVAL_MIN", "60"))
# Arrêt propre sur SIGTERM/SIGINT.
_shutdown = False
def _handle_shutdown(signum, frame):
global _shutdown
_shutdown = True
print(f"[enqueue] signal {signum} reçu — arrêt propre")
def _sleep_interruptible(seconds: float, step: float = 1.0):
remaining = seconds
while remaining > 0 and not _shutdown:
time.sleep(min(step, remaining))
remaining -= step
def resolve_iso2(location_raw: str, band_country: str | None) -> str | None: def resolve_iso2(location_raw: str, band_country: str | None) -> str | None:
t = location_raw.strip() t = location_raw.strip()
# Un code à 2 lettres est validé contre la liste connue : sans ça, un faux
# code ("XX") était renvoyé tel quel, et country_centroid retournait None
# plus loin sans qu'on sache pourquoi. Les alias non standards ("UK") sont
# rattrapés par le mapping.
if len(t) == 2 and t.upper().isalpha(): if len(t) == 2 and t.upper().isalpha():
return t.upper() code = t.upper()
if code in COUNTRY_CENTROIDS:
return code
alias = COUNTRY_NAME_TO_ISO2.get(t.lower())
if alias:
return alias
iso = COUNTRY_NAME_TO_ISO2.get(t.lower()) iso = COUNTRY_NAME_TO_ISO2.get(t.lower())
if iso: if iso:
return iso return iso
if band_country: if band_country:
return band_country.strip().upper() bc = band_country.strip().upper()
return bc if bc in COUNTRY_CENTROIDS else None
return None return None
@ -195,46 +223,95 @@ def _execute_run(cur, trigger_label: str, job_id: int | None = None) -> None:
) )
def main(): def _connect(dsn):
dsn = os.environ["DATABASE_URL"]
conn = psycopg2.connect(dsn) conn = psycopg2.connect(dsn)
conn.autocommit = True conn.autocommit = True
return conn
def _recover_stuck(cur):
"""Referme les runs/jobs geocoder_enqueue laissés 'running' par une instance
précédente tuée brutalement. Strictement bornés à ce type : les autres
appartiennent au crawler, qui fait sa propre récupération."""
cur.execute(
"""UPDATE crawl_run SET status='error', finished_at=now(),
error='interrompu (redémarrage enqueue)'
WHERE run_type='geocoder_enqueue' AND status='running'"""
)
cur.execute(
"""UPDATE job_triggers SET status='error', finished_at=now(),
error='interrompu (redémarrage enqueue)'
WHERE job_type='geocoder_enqueue' AND status='running'"""
)
def main():
signal.signal(signal.SIGTERM, _handle_shutdown)
signal.signal(signal.SIGINT, _handle_shutdown)
dsn = os.environ["DATABASE_URL"]
conn = _connect(dsn)
print(f"[enqueue] daemon démarré, poll {POLL_INTERVAL}s, auto toutes les {AUTO_INTERVAL_MIN}min") print(f"[enqueue] daemon démarré, poll {POLL_INTERVAL}s, auto toutes les {AUTO_INTERVAL_MIN}min")
last_auto = time.monotonic() last_auto = time.monotonic()
with conn.cursor() as cur: with conn.cursor() as cur:
while True: _recover_stuck(cur)
while not _shutdown:
try:
with conn.cursor() as cur:
# Rattrapage automatique (bands rendus 'dirty' par le trigger DB) # Rattrapage automatique (bands rendus 'dirty' par le trigger DB)
if time.monotonic() - last_auto >= AUTO_INTERVAL_MIN * 60: if time.monotonic() - last_auto >= AUTO_INTERVAL_MIN * 60:
_execute_run(cur, "auto") _execute_run(cur, "auto")
last_auto = time.monotonic() last_auto = time.monotonic()
# Déclenchement manuel via l'admin # Déclenchement manuel via l'admin — claim ATOMIQUE : en
# autocommit, le `SELECT … FOR UPDATE` relâchait son verrou avant
# l'UPDATE suivant, donc il ne protégeait rien.
cur.execute( cur.execute(
""" """
UPDATE job_triggers AS t
SET status='running', started_at=now()
WHERE t.id = (
SELECT id FROM job_triggers SELECT id FROM job_triggers
WHERE job_type = 'geocoder_enqueue' AND status = 'pending' WHERE job_type = 'geocoder_enqueue' AND status = 'pending'
ORDER BY created_at ASC ORDER BY created_at ASC
LIMIT 1 LIMIT 1
FOR UPDATE SKIP LOCKED FOR UPDATE SKIP LOCKED
)
RETURNING t.id
""" """
) )
row = cur.fetchone() row = cur.fetchone()
if not row: if not row:
time.sleep(POLL_INTERVAL) _sleep_interruptible(POLL_INTERVAL)
continue continue
job_id = row[0] _execute_run(cur, "manuel", job_id=row[0])
cur.execute( except psycopg2.Error as e:
"UPDATE job_triggers SET status='running', started_at=now() WHERE id=%s", # Perte de connexion en cours de route : on se reconnecte au lieu de
(job_id,), # laisser mourir le process (sinon crash-loop du conteneur).
) print(f"[enqueue] erreur DB ({e}), reconnexion dans 5s")
_execute_run(cur, "manuel", job_id=job_id) try:
conn.close() conn.close()
except Exception: # noqa: S110 - la connexion est déjà morte, c'est le cas nominal ici
pass
_sleep_interruptible(5)
if _shutdown:
break
try:
conn = _connect(dsn)
except Exception as e2:
print(f"[enqueue] reconnexion échouée ({e2})")
_sleep_interruptible(10)
try:
conn.close()
except Exception: # noqa: S110 - fermeture au mieux à l'arrêt, rien à journaliser
pass
if __name__ == "__main__": if __name__ == "__main__":

View file

@ -19,6 +19,7 @@ import hashlib
import json import json
import os import os
import re import re
import signal
import sys import sys
import time import time
from datetime import UTC, datetime from datetime import UTC, datetime
@ -29,6 +30,24 @@ import requests
# sys.path[0] = src/ quand lancé comme "python src/groq_worker.py" # sys.path[0] = src/ quand lancé comme "python src/groq_worker.py"
from parser import COUNTRY_NAMES from parser import COUNTRY_NAMES
# Arrêt propre sur SIGTERM/SIGINT (les redeploys Coolify sont fréquents).
_shutdown = False
def _handle_shutdown(signum, frame):
global _shutdown
_shutdown = True
print(f"[groq] signal {signum} reçu — arrêt propre après l'itération courante")
def _sleep_interruptible(seconds: float, step: float = 1.0):
"""Attend `seconds`, en se réveillant tôt si un arrêt a été demandé."""
remaining = seconds
while remaining > 0 and not _shutdown:
time.sleep(min(step, remaining))
remaining -= step
GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "").strip() GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "").strip()
GROQ_BASE = "https://api.groq.com/openai/v1/chat/completions" GROQ_BASE = "https://api.groq.com/openai/v1/chat/completions"
MAX_LLM_TRIES = int(os.environ.get("GROQ_MAX_TRIES", "5")) MAX_LLM_TRIES = int(os.environ.get("GROQ_MAX_TRIES", "5"))
@ -97,7 +116,7 @@ def _call_groq(model: str, location_raw: str, country: str) -> tuple[str, int, i
{"role": "system", "content": _SYSTEM}, {"role": "system", "content": _SYSTEM},
{"role": "user", "content": prompt}, {"role": "user", "content": prompt},
], ],
"max_tokens": 80, "max_tokens": 120,
"temperature": 0.0, "temperature": 0.0,
"response_format": {"type": "json_object"}, "response_format": {"type": "json_object"},
}, },
@ -156,6 +175,9 @@ def main():
print("[groq] GROQ_API_KEY non configuré — arrêt") print("[groq] GROQ_API_KEY non configuré — arrêt")
sys.exit(0) sys.exit(0)
signal.signal(signal.SIGTERM, _handle_shutdown)
signal.signal(signal.SIGINT, _handle_shutdown)
dsn = os.environ["DATABASE_URL"] dsn = os.environ["DATABASE_URL"]
conn = psycopg2.connect(dsn) conn = psycopg2.connect(dsn)
conn.autocommit = True conn.autocommit = True
@ -166,7 +188,7 @@ def main():
day_start = datetime.now(UTC).date() day_start = datetime.now(UTC).date()
with conn.cursor() as cur: with conn.cursor() as cur:
while True: while not _shutdown:
# Reset compteurs si nouvelle minute / nouveau jour # Reset compteurs si nouvelle minute / nouveau jour
if time.monotonic() - min_start >= 60: if time.monotonic() - min_start >= 60:
min_count = {m[0]: 0 for m in MODELS} min_count = {m[0]: 0 for m in MODELS}
@ -176,23 +198,33 @@ def main():
day_count = {m[0]: 0 for m in MODELS} day_count = {m[0]: 0 for m in MODELS}
day_start = today day_start = today
# Claim ATOMIQUE : on réserve la ligne en poussant geocode_next_at
# 5 min dans le futur, en un seul statement. En autocommit, un
# `SELECT … FOR UPDATE` séparé relâche son verrou immédiatement,
# laissant une 2e réplique sélectionner la même ligne et payer un
# appel LLM en double. Si le worker meurt ensuite, la ligne reste
# 'llm_needed' et sera reprise après 5 min.
cur.execute( cur.execute(
""" """
SELECT bl.id, bl.ma_id, bl.location_raw, bl.geocode_tries_llm, UPDATE band_locations AS t
b.country SET geocode_next_at = now() + interval '5 minutes'
WHERE t.id = (
SELECT bl.id
FROM band_locations bl FROM band_locations bl
JOIN bands b ON b.ma_id = bl.ma_id
WHERE bl.geocode_status = 'llm_needed' WHERE bl.geocode_status = 'llm_needed'
AND bl.geocode_next_at <= now() AND bl.geocode_next_at <= now()
ORDER BY bl.geocode_tries_llm ASC, bl.id ASC ORDER BY bl.geocode_tries_llm ASC, bl.id ASC
LIMIT 1 LIMIT 1
FOR UPDATE OF bl SKIP LOCKED FOR UPDATE SKIP LOCKED
)
RETURNING t.id, t.ma_id, t.location_raw, t.geocode_tries_llm,
(SELECT country FROM bands WHERE ma_id = t.ma_id) AS country
""" """
) )
row = cur.fetchone() row = cur.fetchone()
if not row: if not row:
print("[groq] rien à traiter, attente 120s") print("[groq] rien à traiter, attente 120s")
time.sleep(120) _sleep_interruptible(120)
continue continue
loc_id, ma_id, location_raw, tries_llm, country = row loc_id, ma_id, location_raw, tries_llm, country = row
@ -207,11 +239,14 @@ def main():
if not chosen_model: if not chosen_model:
wait = max(5, 60 - (time.monotonic() - min_start)) wait = max(5, 60 - (time.monotonic() - min_start))
print(f"[groq] quota atteint, attente {wait:.0f}s") print(f"[groq] quota atteint, attente {wait:.0f}s")
# Relâcher la réservation posée par le claim : sinon la ligne
# resterait inutilement décalée de 5 min alors qu'on ne l'a pas
# traitée.
cur.execute( cur.execute(
"UPDATE band_locations SET geocode_status='llm_needed', updated_at=now() WHERE id=%s", "UPDATE band_locations SET geocode_next_at=now(), updated_at=now() WHERE id=%s",
(loc_id,), (loc_id,),
) )
time.sleep(wait) _sleep_interruptible(wait)
continue continue
# Vérifier llm_cache # Vérifier llm_cache
@ -277,7 +312,7 @@ def main():
min_count[hit_model] = 9999 min_count[hit_model] = 9999
print(f"[groq] rate limit {hit_model}") print(f"[groq] rate limit {hit_model}")
cur.execute( cur.execute(
"UPDATE band_locations SET geocode_status='llm_needed', updated_at=now() WHERE id=%s", "UPDATE band_locations SET geocode_next_at=now(), updated_at=now() WHERE id=%s",
(loc_id,), (loc_id,),
) )
continue continue
@ -296,8 +331,28 @@ def main():
(new_tries, str(exc)[:300], backoff, loc_id), (new_tries, str(exc)[:300], backoff, loc_id),
) )
print(f"[groq] id={loc_id} error: {exc}") print(f"[groq] id={loc_id} error: {exc}")
except Exception as exc:
# requests lève HTTPError / Timeout / ConnectionError, qui
# héritent d'OSError et NON de RuntimeError : sans ce filet,
# elles s'échappaient de la boucle et tuaient le process — donc
# crash-loop du conteneur sur toute panne durable de l'API Groq.
# Idem pour un JSON invalide renvoyé par r.json().
new_tries = tries_llm + 1
backoff = 30 * new_tries
cur.execute(
"""
UPDATE band_locations
SET geocode_tries_llm=%s,
geocode_error=%s,
geocode_next_at=now() + (%s || ' minutes')::interval,
updated_at=now()
WHERE id=%s
""",
(new_tries, f"{type(exc).__name__}: {exc}"[:300], backoff, loc_id),
)
print(f"[groq] id={loc_id} erreur inattendue: {type(exc).__name__}: {exc}")
time.sleep(CALL_DELAY) _sleep_interruptible(CALL_DELAY)
conn.close() conn.close()

View file

@ -77,13 +77,17 @@ COUNTRY_NAME_TO_ISO2 = {
'north macedonia': 'MK', 'macedonia': 'MK', 'montenegro': 'ME', 'north macedonia': 'MK', 'macedonia': 'MK', 'montenegro': 'ME',
'kosovo': 'XK', 'san marino': 'SM', 'liechtenstein': 'LI', 'kosovo': 'XK', 'san marino': 'SM', 'liechtenstein': 'LI',
'monaco': 'MC', 'andorra': 'AD', 'vatican': 'VA', 'monaco': 'MC', 'andorra': 'AD', 'vatican': 'VA',
# Alias non standards fréquents sur Metal Archives
'uk': 'GB', 'great britain': 'GB', 'holland': 'NL', 'czechia': 'CZ',
} }
COUNTRY_NAMES = {v: k.title() for k, v in COUNTRY_NAME_TO_ISO2.items() if len(v) == 2} COUNTRY_NAMES = {v: k.title() for k, v in COUNTRY_NAME_TO_ISO2.items() if len(v) == 2}
# Fix doublons (en → GB wins "United Kingdom") # Plusieurs libellés pointent vers le même ISO2 ; l'inversion ci-dessus garde le
# DERNIER rencontré, qui n'est pas forcément le nom canonique. On réimpose donc
# explicitement le nom à envoyer au géocodeur pour chaque code concerné.
COUNTRY_NAMES.update({ COUNTRY_NAMES.update({
'GB': 'United Kingdom', 'CZ': 'Czech Republic', 'BA': 'Bosnia and Herzegovina', 'GB': 'United Kingdom', 'CZ': 'Czech Republic', 'BA': 'Bosnia and Herzegovina',
'MK': 'North Macedonia', 'XK': 'Kosovo', 'MK': 'North Macedonia', 'XK': 'Kosovo', 'NL': 'Netherlands',
}) })
_ISO2_RE = re.compile(r'^[A-Z]{2}$') _ISO2_RE = re.compile(r'^[A-Z]{2}$')

View file

@ -14,9 +14,11 @@ Stratégie :
4. Succès band_locations + sync bands.lat/lon (origine). 4. Succès band_locations + sync bands.lat/lon (origine).
""" """
import contextlib
import json import json
import os import os
import random import random
import signal
import time import time
import psycopg2 import psycopg2
@ -27,6 +29,46 @@ from parser import build_fallback_queries
GEOAPIFY_API_KEY = os.environ.get("GEOAPIFY_API_KEY", "").strip() GEOAPIFY_API_KEY = os.environ.get("GEOAPIFY_API_KEY", "").strip()
GEOAPIFY_BASE = "https://api.geoapify.com/v1/geocode/search" GEOAPIFY_BASE = "https://api.geoapify.com/v1/geocode/search"
# Délai au-delà duquel une ligne restée 'processing' est considérée orpheline
# (worker tué en plein traitement) et remise en file.
STUCK_PROCESSING_MIN = int(os.environ.get("GEOCODE_STUCK_PROCESSING_MIN", "10"))
# Arrêt propre sur SIGTERM/SIGINT (les redeploys Coolify sont fréquents) : on
# termine l'itération en cours au lieu de laisser une ligne en 'processing'.
_shutdown = False
def _handle_shutdown(signum, frame):
global _shutdown
_shutdown = True
print(f"[worker] signal {signum} reçu — arrêt propre après l'itération courante")
def _scrub(msg: str) -> str:
"""Ne jamais laisser la clé API dans un message d'erreur.
Ces messages atterrissent dans band_locations.geocode_error, que le
dashboard admin affiche. Tronque aussi à 300 caractères.
"""
if GEOAPIFY_API_KEY and GEOAPIFY_API_KEY in msg:
msg = msg.replace(GEOAPIFY_API_KEY, "***")
return msg[:300]
@contextlib.contextmanager
def _tx(conn):
"""Transaction explicite ponctuelle malgré autocommit=True : rend atomique
un groupe d'UPDATE liés (band_locations + bands)."""
conn.autocommit = False
try:
yield
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.autocommit = True
MIN_DELAY = float(os.environ.get("GEOCODER_MIN_DELAY", "0.22")) MIN_DELAY = float(os.environ.get("GEOCODER_MIN_DELAY", "0.22"))
JITTER = float(os.environ.get("GEOCODER_JITTER", "0.10")) JITTER = float(os.environ.get("GEOCODER_JITTER", "0.10"))
MAX_PER_RUN = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000")) MAX_PER_RUN = int(os.environ.get("GEOCODE_MAX_PER_RUN", "100000"))
@ -53,10 +95,17 @@ def is_reliable(confidence, granularity, is_country_only) -> bool:
On exige STRICTEMENT plus que le seuil : 0.7 et en-dessous rejeté (LLM). On exige STRICTEMENT plus que le seuil : 0.7 et en-dessous rejeté (LLM).
""" """
if is_country_only: # La confiance est TOUJOURS exigée, y compris pour un lieu "pays seul". Une
return True # résolu par centroïde à l'enqueue, granularité pays assumée # ligne is_country_only est censée avoir été résolue par centroïde dès
# l'enqueue ; si elle atterrit ici, c'est que le centroïde a échoué (code non
# standard type "UK"). On ne peut donc PAS lui faire confiance aveuglément,
# sous peine de poser un point aberrant sur la carte.
if confidence is None or confidence <= MIN_CONFIDENCE: if confidence is None or confidence <= MIN_CONFIDENCE:
return False return False
if is_country_only:
# Un résultat "pays" est légitimement grossier : on saute le rejet de
# granularité, mais la confiance vient bien d'être vérifiée.
return True
if granularity and granularity.lower() in COARSE_TYPES: if granularity and granularity.lower() in COARSE_TYPES:
return False return False
return True return True
@ -67,10 +116,18 @@ def geoapify_search(query: str) -> dict | None:
raise RuntimeError("GEOAPIFY_API_KEY not set") raise RuntimeError("GEOAPIFY_API_KEY not set")
params = {"text": query, "apiKey": GEOAPIFY_API_KEY, "limit": 1} params = {"text": query, "apiKey": GEOAPIFY_API_KEY, "limit": 1}
_polite_sleep() _polite_sleep()
# La clé part en query string : toute exception requests (raise_for_status,
# ConnectionError, Timeout…) embarque l'URL COMPLÈTE, donc la clé. Ces
# messages finissent dans band_locations.geocode_error, affiché par l'admin.
# On les remplace donc par une erreur propre, sans URL.
try:
r = requests.get(GEOAPIFY_BASE, params=params, timeout=30) r = requests.get(GEOAPIFY_BASE, params=params, timeout=30)
except requests.RequestException as e:
raise RuntimeError(f"geoapify_network:{type(e).__name__}") from None
if r.status_code == 429: if r.status_code == 429:
raise RuntimeError("geoapify_throttle:429") raise RuntimeError("geoapify_throttle:429")
r.raise_for_status() if r.status_code != 200:
raise RuntimeError(f"geoapify_http:{r.status_code}")
data = r.json() data = r.json()
features = data.get("features") or [] features = data.get("features") or []
if not features: if not features:
@ -171,7 +228,30 @@ def try_dedup(cur, loc_id, ma_id, location_raw, country) -> bool:
return True return True
def recover_stuck_processing(cur) -> int:
"""Remet en file toute ligne 'processing' orpheline.
Sans ça, un worker tué en plein traitement (redeploy, OOM) laisse sa ligne
bloquée pour toujours : rien d'autre ne relit ce statut.
"""
cur.execute(
"""
UPDATE band_locations
SET geocode_status='queued', updated_at=now()
WHERE geocode_status='processing'
AND updated_at < now() - make_interval(mins => %s)
""",
(STUCK_PROCESSING_MIN,),
)
if cur.rowcount:
print(f"[worker] {cur.rowcount} ligne(s) 'processing' bloquée(s) remise(s) en file")
return cur.rowcount
def main(): def main():
signal.signal(signal.SIGTERM, _handle_shutdown)
signal.signal(signal.SIGINT, _handle_shutdown)
dsn = os.environ["DATABASE_URL"] dsn = os.environ["DATABASE_URL"]
conn = psycopg2.connect(dsn) conn = psycopg2.connect(dsn)
conn.autocommit = True conn.autocommit = True
@ -209,36 +289,52 @@ def main():
processed = 0 processed = 0
with conn.cursor() as cur: with conn.cursor() as cur:
while processed < MAX_PER_RUN: # Au démarrage : récupérer les lignes laissées 'processing' par une
# instance précédente tuée brutalement.
recover_stuck_processing(cur)
while processed < MAX_PER_RUN and not _shutdown:
# Claim ATOMIQUE : la sélection et le passage en 'processing' sont un
# SEUL statement. En autocommit, un `SELECT … FOR UPDATE` séparé
# relâche son verrou avant l'UPDATE suivant — deux répliques
# pouvaient alors réclamer la même ligne et payer deux appels API.
cur.execute( cur.execute(
""" """
SELECT bl.id, bl.ma_id, bl.location_raw, bl.is_country_only, UPDATE band_locations AS t
bl.geocode_tries_geo, bl.geocode_query, b.country SET geocode_status='processing', updated_at=now()
WHERE t.id = (
SELECT bl.id
FROM band_locations bl FROM band_locations bl
JOIN bands b ON b.ma_id = bl.ma_id
WHERE bl.geocode_status = 'queued' WHERE bl.geocode_status = 'queued'
AND bl.geocode_next_at <= now() AND bl.geocode_next_at <= now()
ORDER BY bl.geocode_next_at ASC, bl.id ASC ORDER BY bl.geocode_next_at ASC, bl.id ASC
LIMIT 1 LIMIT 1
FOR UPDATE OF bl SKIP LOCKED FOR UPDATE SKIP LOCKED
)
RETURNING t.id, t.ma_id, t.location_raw, t.is_country_only,
t.geocode_tries_geo, t.geocode_query,
(SELECT country FROM bands WHERE ma_id = t.ma_id) AS country
""" """
) )
row = cur.fetchone() row = cur.fetchone()
if not row: if not row:
# File vide : profiter de l'inactivité pour récupérer les lignes
# 'processing' bloquées, puis attendre.
recover_stuck_processing(cur)
print("[worker] rien à traiter, attente 60s") print("[worker] rien à traiter, attente 60s")
health_check() health_check()
time.sleep(60) for _ in range(60):
if _shutdown:
break
time.sleep(1)
continue continue
loc_id, ma_id, location_raw, is_country_only, tries, llm_query, country = row loc_id, ma_id, location_raw, is_country_only, tries, llm_query, country = row
cur.execute( # 0. Fast-path dedup (aucun appel API) — atomique (band_locations + bands)
"UPDATE band_locations SET geocode_status='processing', updated_at=now() WHERE id=%s", with _tx(conn):
(loc_id,), deduped = try_dedup(cur, loc_id, ma_id, location_raw, country)
) if deduped:
# 0. Fast-path dedup (aucun appel API)
if try_dedup(cur, loc_id, ma_id, location_raw, country):
processed += 1 processed += 1
continue continue
@ -339,14 +435,15 @@ def main():
rate_limit = True rate_limit = True
time.sleep(60) time.sleep(60)
break break
last_err = str(exc)[:300] last_err = _scrub(str(exc))
except Exception as exc: except Exception as exc:
last_err = str(exc)[:300] last_err = _scrub(str(exc))
if rate_limit: if rate_limit:
continue continue
if success: if success:
with _tx(conn):
mark_done(cur, loc_id, ma_id, lat, lon, used_query, mark_done(cur, loc_id, ma_id, lat, lon, used_query,
provider, confidence, granularity) provider, confidence, granularity)
else: else: