From 6a5827561bb06de30c3619061eeed21092a41ab1 Mon Sep 17 00:00:00 2001 From: FuelBoard Contributor Date: Thu, 13 Aug 2026 14:01:13 +0100 Subject: [PATCH] =?UTF-8?q?Sync=20retry=20with=20exponential=20backoff=20(?= =?UTF-8?q?15s=E2=86=9215min=20cap);=20health=20exposes=20retry=20state?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/main.py | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/app/main.py b/app/main.py index 945c58b..a471b5d 100644 --- a/app/main.py +++ b/app/main.py @@ -94,6 +94,10 @@ _state = { "last_error": None, "sync_attempts": 0, "sync_failures": 0, + "retry_attempt": 0, # consecutive failures (backoff exponent) + "retry_active": False, # a backoff sleep is in flight + "retry_in": None, # seconds until the next retry (None when idle) + "next_retry_at": 0.0, # epoch of the next scheduled retry "source_used": None, "data_updated": None, # freshest govUK price_last_updated across dataset } @@ -416,6 +420,9 @@ import threading _refresh_lock = threading.Lock() +_RETRY_BASE_S = 15 # first retry 15s after a failure +_RETRY_MAX_S = 900 # exponential backoff capped at 15 minutes + def _refresh_cache() -> None: """Refresh the station cache (called synchronously or from a thread).""" @@ -447,9 +454,35 @@ def _refresh_cache_locked() -> None: _state["source_used"] = "csv" _state["stations_fetched_at"] = time.time() _state["last_error"] = None + _state["retry_attempt"] = 0 + _state["retry_in"] = None + _state["next_retry_at"] = 0.0 except Exception as exc: # keep serving stale cache on failure _state["sync_failures"] += 1 _state["last_error"] = f"{type(exc).__name__}: {exc}" + _state["retry_attempt"] += 1 + delay = min(_RETRY_BASE_S * (2 ** (_state["retry_attempt"] - 1)), _RETRY_MAX_S) + _state["retry_in"] = delay + _state["next_retry_at"] = time.time() + delay + _schedule_retry(delay) + + +def _schedule_retry(delay: float) -> None: + """Retry a failed sync on a daemon thread after `delay` seconds. + + Only one retry sleeps at a time; the worker clears the flag before + refreshing, so a repeated failure can chain the next (longer) retry. + """ + if _state["retry_active"]: + return + _state["retry_active"] = True + + def worker() -> None: + time.sleep(delay) + _state["retry_active"] = False + _refresh_cache() + + threading.Thread(target=worker, daemon=True).start() def _ensure_fresh_cache() -> None: @@ -530,6 +563,10 @@ def health(): "last_sync": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_state["stations_fetched_at"])) if _state["stations_fetched_at"] else None, "sync_attempts": _state["sync_attempts"], "sync_failures": _state["sync_failures"], + "retry_active": _state["retry_active"], + "retry_in": _state["retry_in"], + "next_retry_at": (time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(_state["next_retry_at"])) + if _state["next_retry_at"] else None), "last_error": _state["last_error"], }