Guard sold reconciliation by scope

mobilede_scraper/worker/refresh_cycle.py

@@ -15,6 +15,7 @@ from .constants import (
     MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY,
     MOBILEDE_REFRESH_CYCLE_TOTAL_KEY,
     MOBILEDE_REFRESH_CYCLE_TTL_SECONDS,
+    MOBILEDE_REFRESH_CYCLE_UNSAFE_REASONS_KEY_FMT,
 )
 
 
@@ -26,6 +27,41 @@ def _mobilede_refresh_cycle_finalized_key(cycle_id: str) -> str:
     return MOBILEDE_REFRESH_CYCLE_FINALIZED_KEY_FMT.format(cycle_id=cycle_id)
 
 
+def _mobilede_refresh_cycle_unsafe_reasons_key(cycle_id: str) -> str:
+    return MOBILEDE_REFRESH_CYCLE_UNSAFE_REASONS_KEY_FMT.format(cycle_id=cycle_id)
+
+
+def _mobilede_mark_refresh_cycle_reconciliation_unsafe(
+    redis_client: Redis,
+    *,
+    cycle_id: str | None,
+    reason: str,
+    logger: logging.Logger,
+) -> None:
+    if not cycle_id or not reason:
+        return
+    active_cycle_id = _mobilede_redis_text(redis_client.get(MOBILEDE_REFRESH_CYCLE_ID_KEY)).strip()
+    if active_cycle_id != cycle_id:
+        return
+    key = _mobilede_refresh_cycle_unsafe_reasons_key(cycle_id)
+    ttl = max(3600, int(MOBILEDE_REFRESH_CYCLE_TTL_SECONDS))
+    redis_client.sadd(key, reason)
+    redis_client.expire(key, ttl)
+    logger.warning("mobile.de refresh sold reconciliation disabled: cycle=%s reason=%s", cycle_id, reason)
+
+
+def _mobilede_refresh_cycle_unsafe_reasons(redis_client: Redis, *, cycle_id: str) -> set[str]:
+    key = _mobilede_refresh_cycle_unsafe_reasons_key(cycle_id)
+    try:
+        return {
+            _mobilede_redis_text(value).strip()
+            for value in redis_client.smembers(key)
+            if _mobilede_redis_text(value).strip()
+        }
+    except Exception:
+        return set()
+
+
 def _mobilede_redis_text(value: object) -> str:
     if isinstance(value, bytes):
         return value.decode("utf-8", errors="ignore")
@@ -55,6 +91,7 @@ def _mobilede_clear_refresh_cycle(redis_client: Redis) -> None:
     if cycle_id:
         keys.append(_mobilede_refresh_cycle_done_set_key(cycle_id))
         keys.append(_mobilede_refresh_cycle_finalized_key(cycle_id))
+        keys.append(_mobilede_refresh_cycle_unsafe_reasons_key(cycle_id))
     redis_client.delete(*keys)
 
 
@@ -76,6 +113,7 @@ def _mobilede_start_refresh_cycle(
     pipe.set(MOBILEDE_REFRESH_CYCLE_DONE_KEY, "0", ex=ttl)
     pipe.delete(done_set_key)
     pipe.delete(_mobilede_refresh_cycle_finalized_key(cycle_id))
+    pipe.delete(_mobilede_refresh_cycle_unsafe_reasons_key(cycle_id))
     pipe.execute()
     logger.info("mobile.de refresh cycle started: id=%s total=%s started_at=%s", cycle_id, total, started_at)
     return cycle_id
@@ -180,6 +218,20 @@ def _mobilede_finalize_refresh_cycle_sold_marking(
     post_refresh_probe_batch_size: int,
     schedule_post_refresh_probe: Callable[[], None],
 ) -> int:
+    active_cycle_id = _mobilede_redis_text(redis_client.get(MOBILEDE_REFRESH_CYCLE_ID_KEY)).strip()
+    done_now = int(redis_client.get(MOBILEDE_REFRESH_CYCLE_DONE_KEY) or 0)
+    total_now = int(redis_client.get(MOBILEDE_REFRESH_CYCLE_TOTAL_KEY) or 0)
+    unsafe_reasons = _mobilede_refresh_cycle_unsafe_reasons(redis_client, cycle_id=cycle_id)
+    if active_cycle_id != cycle_id or total_now <= 0 or done_now < total_now or unsafe_reasons:
+        logger.warning(
+            "mobile.de refresh sold marking skipped: cycle=%s active=%s progress=%s/%s unsafe_reasons=%s",
+            cycle_id,
+            active_cycle_id,
+            done_now,
+            total_now,
+            sorted(unsafe_reasons),
+        )
+        return 0
     started_at_raw = redis_client.get(MOBILEDE_REFRESH_CYCLE_STARTED_AT_KEY)
     if not started_at_raw:
         logger.warning("mobile.de refresh sold marking skipped: missing started_at for cycle=%s", cycle_id)
@@ -200,8 +252,6 @@ def _mobilede_finalize_refresh_cycle_sold_marking(
         prefix=origin_prefixes,
         safety_ratio=0.8,
     )
-    done_now = int(redis_client.get(MOBILEDE_REFRESH_CYCLE_DONE_KEY) or 0)
-    total_now = int(redis_client.get(MOBILEDE_REFRESH_CYCLE_TOTAL_KEY) or 0)
     logger.info(
         "mobile.de refresh sold marking completed: cycle=%s progress=%s/%s cutoff=%s sold_marked=%s",
         cycle_id,
@@ -243,17 +293,13 @@ def _mobilede_try_finalize_refresh_cycle_after_bootstrap_completion(
     if total_now <= 0:
         return False
     if done_now < total_now:
-        redis_client.set(
-            MOBILEDE_REFRESH_CYCLE_DONE_KEY,
-            str(total_now),
-            ex=max(3600, int(MOBILEDE_REFRESH_CYCLE_TTL_SECONDS)),
-        )
         logger.warning(
-            "mobile.de refresh finalize fallback: bootstrap completed while refresh progress lagged cycle=%s done=%s/%s",
+            "mobile.de refresh sold marking skipped: bootstrap completed while refresh progress is incomplete cycle=%s done=%s/%s",
             cycle_id,
             done_now,
             total_now,
         )
+        return False
     if not _mobilede_claim_refresh_cycle_finalization(redis_client, cycle_id=cycle_id):
         return False
     finalize_refresh_cycle_sold_marking(redis_client, cycle_id=cycle_id)