From cc6be5e341532fa0aa5073e3e7945bd91cd4a9aa Mon Sep 17 00:00:00 2001 From: Nicolas FRYDER Date: Thu, 20 Aug 2026 16:44:07 +0200 Subject: [PATCH] =?UTF-8?q?fix(geocoder):=20fuite=20de=20cl=C3=A9=20API,?= =?UTF-8?q?=20r=C3=A9servations=20atomiques,=20arr=C3=AAt=20propre?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- apps/geocoder/src/enqueue.py | 137 +++++++++++++++++++++------- apps/geocoder/src/groq_worker.py | 87 ++++++++++++++---- apps/geocoder/src/parser.py | 8 +- apps/geocoder/src/worker.py | 149 +++++++++++++++++++++++++------ 4 files changed, 307 insertions(+), 74 deletions(-) diff --git a/apps/geocoder/src/enqueue.py b/apps/geocoder/src/enqueue.py index 179ad50..28d5152 100644 --- a/apps/geocoder/src/enqueue.py +++ b/apps/geocoder/src/enqueue.py @@ -17,10 +17,12 @@ Tourne en continu (restart: unless-stopped), poll toutes les 15s. """ import os +import signal import time import psycopg2 from parser import ( + COUNTRY_CENTROIDS, COUNTRY_NAME_TO_ISO2, country_centroid, 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")) 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: 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(): - 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()) if iso: return iso if band_country: - return band_country.strip().upper() + bc = band_country.strip().upper() + return bc if bc in COUNTRY_CENTROIDS else None return None @@ -195,46 +223,95 @@ def _execute_run(cur, trigger_label: str, job_id: int | None = None) -> None: ) -def main(): - dsn = os.environ["DATABASE_URL"] +def _connect(dsn): conn = psycopg2.connect(dsn) 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") last_auto = time.monotonic() with conn.cursor() as cur: - while True: - # Rattrapage automatique (bands rendus 'dirty' par le trigger DB) - if time.monotonic() - last_auto >= AUTO_INTERVAL_MIN * 60: - _execute_run(cur, "auto") - last_auto = time.monotonic() + _recover_stuck(cur) - # Déclenchement manuel via l'admin - cur.execute( - """ - SELECT id FROM job_triggers - WHERE job_type = 'geocoder_enqueue' AND status = 'pending' - ORDER BY created_at ASC - LIMIT 1 - FOR UPDATE SKIP LOCKED - """ - ) - row = cur.fetchone() + while not _shutdown: + try: + with conn.cursor() as cur: + # Rattrapage automatique (bands rendus 'dirty' par le trigger DB) + if time.monotonic() - last_auto >= AUTO_INTERVAL_MIN * 60: + _execute_run(cur, "auto") + last_auto = time.monotonic() - if not row: - time.sleep(POLL_INTERVAL) - continue + # 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( + """ + UPDATE job_triggers AS t + SET status='running', started_at=now() + WHERE t.id = ( + SELECT id FROM job_triggers + WHERE job_type = 'geocoder_enqueue' AND status = 'pending' + ORDER BY created_at ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + ) + RETURNING t.id + """ + ) + row = cur.fetchone() - job_id = row[0] - cur.execute( - "UPDATE job_triggers SET status='running', started_at=now() WHERE id=%s", - (job_id,), - ) - _execute_run(cur, "manuel", job_id=job_id) + if not row: + _sleep_interruptible(POLL_INTERVAL) + continue - conn.close() + _execute_run(cur, "manuel", job_id=row[0]) + except psycopg2.Error as e: + # Perte de connexion en cours de route : on se reconnecte au lieu de + # laisser mourir le process (sinon crash-loop du conteneur). + print(f"[enqueue] erreur DB ({e}), reconnexion dans 5s") + try: + 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__": diff --git a/apps/geocoder/src/groq_worker.py b/apps/geocoder/src/groq_worker.py index 781f200..0f59070 100644 --- a/apps/geocoder/src/groq_worker.py +++ b/apps/geocoder/src/groq_worker.py @@ -19,6 +19,7 @@ import hashlib import json import os import re +import signal import sys import time from datetime import UTC, datetime @@ -29,6 +30,24 @@ import requests # sys.path[0] = src/ quand lancé comme "python src/groq_worker.py" 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_BASE = "https://api.groq.com/openai/v1/chat/completions" 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": "user", "content": prompt}, ], - "max_tokens": 80, + "max_tokens": 120, "temperature": 0.0, "response_format": {"type": "json_object"}, }, @@ -156,6 +175,9 @@ def main(): print("[groq] GROQ_API_KEY non configuré — arrêt") sys.exit(0) + signal.signal(signal.SIGTERM, _handle_shutdown) + signal.signal(signal.SIGINT, _handle_shutdown) + dsn = os.environ["DATABASE_URL"] conn = psycopg2.connect(dsn) conn.autocommit = True @@ -166,7 +188,7 @@ def main(): day_start = datetime.now(UTC).date() with conn.cursor() as cur: - while True: + while not _shutdown: # Reset compteurs si nouvelle minute / nouveau jour if time.monotonic() - min_start >= 60: 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_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( """ - SELECT bl.id, bl.ma_id, bl.location_raw, bl.geocode_tries_llm, - b.country - FROM band_locations bl - JOIN bands b ON b.ma_id = bl.ma_id - WHERE bl.geocode_status = 'llm_needed' - AND bl.geocode_next_at <= now() - ORDER BY bl.geocode_tries_llm ASC, bl.id ASC - LIMIT 1 - FOR UPDATE OF bl SKIP LOCKED + UPDATE band_locations AS t + SET geocode_next_at = now() + interval '5 minutes' + WHERE t.id = ( + SELECT bl.id + FROM band_locations bl + WHERE bl.geocode_status = 'llm_needed' + AND bl.geocode_next_at <= now() + ORDER BY bl.geocode_tries_llm ASC, bl.id ASC + LIMIT 1 + 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() if not row: print("[groq] rien à traiter, attente 120s") - time.sleep(120) + _sleep_interruptible(120) continue loc_id, ma_id, location_raw, tries_llm, country = row @@ -207,11 +239,14 @@ def main(): if not chosen_model: wait = max(5, 60 - (time.monotonic() - min_start)) 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( - "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,), ) - time.sleep(wait) + _sleep_interruptible(wait) continue # Vérifier llm_cache @@ -277,7 +312,7 @@ def main(): min_count[hit_model] = 9999 print(f"[groq] rate limit {hit_model}") 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,), ) continue @@ -296,8 +331,28 @@ def main(): (new_tries, str(exc)[:300], backoff, loc_id), ) 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() diff --git a/apps/geocoder/src/parser.py b/apps/geocoder/src/parser.py index ef0967d..22020c6 100644 --- a/apps/geocoder/src/parser.py +++ b/apps/geocoder/src/parser.py @@ -77,13 +77,17 @@ COUNTRY_NAME_TO_ISO2 = { 'north macedonia': 'MK', 'macedonia': 'MK', 'montenegro': 'ME', 'kosovo': 'XK', 'san marino': 'SM', 'liechtenstein': 'LI', '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} -# 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({ '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}$') diff --git a/apps/geocoder/src/worker.py b/apps/geocoder/src/worker.py index 9e9fb54..d53b20d 100644 --- a/apps/geocoder/src/worker.py +++ b/apps/geocoder/src/worker.py @@ -14,9 +14,11 @@ Stratégie : 4. Succès → band_locations + sync bands.lat/lon (origine). """ +import contextlib import json import os import random +import signal import time import psycopg2 @@ -27,6 +29,46 @@ from parser import build_fallback_queries GEOAPIFY_API_KEY = os.environ.get("GEOAPIFY_API_KEY", "").strip() 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")) JITTER = float(os.environ.get("GEOCODER_JITTER", "0.10")) 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). """ - if is_country_only: - return True # résolu par centroïde à l'enqueue, granularité pays assumée + # La confiance est TOUJOURS exigée, y compris pour un lieu "pays seul". Une + # 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: 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: return False return True @@ -67,10 +116,18 @@ def geoapify_search(query: str) -> dict | None: raise RuntimeError("GEOAPIFY_API_KEY not set") params = {"text": query, "apiKey": GEOAPIFY_API_KEY, "limit": 1} _polite_sleep() - r = requests.get(GEOAPIFY_BASE, params=params, timeout=30) + # 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) + except requests.RequestException as e: + raise RuntimeError(f"geoapify_network:{type(e).__name__}") from None if r.status_code == 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() features = data.get("features") or [] if not features: @@ -171,7 +228,30 @@ def try_dedup(cur, loc_id, ma_id, location_raw, country) -> bool: 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(): + signal.signal(signal.SIGTERM, _handle_shutdown) + signal.signal(signal.SIGINT, _handle_shutdown) + dsn = os.environ["DATABASE_URL"] conn = psycopg2.connect(dsn) conn.autocommit = True @@ -209,36 +289,52 @@ def main(): processed = 0 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( """ - SELECT bl.id, bl.ma_id, bl.location_raw, bl.is_country_only, - bl.geocode_tries_geo, bl.geocode_query, b.country - FROM band_locations bl - JOIN bands b ON b.ma_id = bl.ma_id - WHERE bl.geocode_status = 'queued' - AND bl.geocode_next_at <= now() - ORDER BY bl.geocode_next_at ASC, bl.id ASC - LIMIT 1 - FOR UPDATE OF bl SKIP LOCKED + UPDATE band_locations AS t + SET geocode_status='processing', updated_at=now() + WHERE t.id = ( + SELECT bl.id + FROM band_locations bl + WHERE bl.geocode_status = 'queued' + AND bl.geocode_next_at <= now() + ORDER BY bl.geocode_next_at ASC, bl.id ASC + LIMIT 1 + 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() 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") health_check() - time.sleep(60) + for _ in range(60): + if _shutdown: + break + time.sleep(1) continue loc_id, ma_id, location_raw, is_country_only, tries, llm_query, country = row - cur.execute( - "UPDATE band_locations SET geocode_status='processing', updated_at=now() WHERE id=%s", - (loc_id,), - ) - - # 0. Fast-path dedup (aucun appel API) - if try_dedup(cur, loc_id, ma_id, location_raw, country): + # 0. Fast-path dedup (aucun appel API) — atomique (band_locations + bands) + with _tx(conn): + deduped = try_dedup(cur, loc_id, ma_id, location_raw, country) + if deduped: processed += 1 continue @@ -339,16 +435,17 @@ def main(): rate_limit = True time.sleep(60) break - last_err = str(exc)[:300] + last_err = _scrub(str(exc)) except Exception as exc: - last_err = str(exc)[:300] + last_err = _scrub(str(exc)) if rate_limit: continue if success: - mark_done(cur, loc_id, ma_id, lat, lon, used_query, - provider, confidence, granularity) + with _tx(conn): + mark_done(cur, loc_id, ma_id, lat, lon, used_query, + provider, confidence, granularity) else: new_tries = tries + 1 if new_tries >= MAX_GEO_TRIES: