Sync retry with exponential backoff (15s→15min cap); health exposes retry state
This commit is contained in:
+37
@@ -94,6 +94,10 @@ _state = {
|
|||||||
"last_error": None,
|
"last_error": None,
|
||||||
"sync_attempts": 0,
|
"sync_attempts": 0,
|
||||||
"sync_failures": 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,
|
"source_used": None,
|
||||||
"data_updated": None, # freshest govUK price_last_updated across dataset
|
"data_updated": None, # freshest govUK price_last_updated across dataset
|
||||||
}
|
}
|
||||||
@@ -416,6 +420,9 @@ import threading
|
|||||||
|
|
||||||
_refresh_lock = threading.Lock()
|
_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:
|
def _refresh_cache() -> None:
|
||||||
"""Refresh the station cache (called synchronously or from a thread)."""
|
"""Refresh the station cache (called synchronously or from a thread)."""
|
||||||
@@ -447,9 +454,35 @@ def _refresh_cache_locked() -> None:
|
|||||||
_state["source_used"] = "csv"
|
_state["source_used"] = "csv"
|
||||||
_state["stations_fetched_at"] = time.time()
|
_state["stations_fetched_at"] = time.time()
|
||||||
_state["last_error"] = None
|
_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
|
except Exception as exc: # keep serving stale cache on failure
|
||||||
_state["sync_failures"] += 1
|
_state["sync_failures"] += 1
|
||||||
_state["last_error"] = f"{type(exc).__name__}: {exc}"
|
_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:
|
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,
|
"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_attempts": _state["sync_attempts"],
|
||||||
"sync_failures": _state["sync_failures"],
|
"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"],
|
"last_error": _state["last_error"],
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user