qananasikq/dubizzle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stabilize worker sync flow

Browse Source
This commit is contained in:
qananasikq
2026-04-17 17:30:28 +03:00
parent 3c3aea8eb3
commit 05d1ffb5ac
9 changed files with 803 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
iaai_scraper/worker/celery_app.py
View File

@@ -4,11 +4,13 @@ import logging
from celery import Celery
from celery.signals import worker_process_init, worker_ready, setup_logging as celery_setup_logging
from redis import Redis
from ..core.config import settings
from ..core.logs import setup_logging
logger = logging.getLogger("iaai_scraper.worker.celery_app")
STARTUP_SYNC_DISPATCH_KEY = "iaai:state:startup_sync_dispatched"
@celery_setup_logging.connect
@@ -96,6 +98,24 @@ celery_app.autodiscover_tasks(["iaai_scraper.worker"])
def _on_worker_ready(**kwargs):
"""Сразу при старте worker отправляем первую задачу sync_listing,
чтобы не ждать час до первого beat-цикла."""
try:
redis_client = Redis.from_url(
settings.redis.url,
decode_responses=True,
socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
socket_timeout=settings.redis.socket_timeout_seconds,
health_check_interval=settings.redis.health_check_interval_seconds,
retry_on_timeout=True,
)
should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600))
except Exception:
logger.warning("Worker ready startup sync dedupe check failed; skipping immediate dispatch", exc_info=True)
return
if not should_dispatch:
logger.info("Worker ready immediate sync already dispatched recently; skipping duplicate enqueue")
return
logger.info("Worker ready — dispatching initial sync_listing task")
celery_app.send_task(
"iaai_scraper.worker.tasks.sync_listing_task",

222
iaai_scraper/worker/tasks.py
View File

@@ -20,6 +20,12 @@ logger = logging.getLogger("iaai_scraper.worker.tasks")
SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
SYNC_LISTING_CHECKPOINT_TTL_SECONDS = 7 * 24 * 60 * 60
SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT = 2
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY = "iaai:state:sync_listing_bootstrap_failure_streak"
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT = 3
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS = 24 * 60 * 60
SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
SYNC_LISTING_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task"
@@ -101,7 +107,14 @@ def _get_persistence() -> PersistenceService:
def _get_redis() -> Redis:
settings = Settings()
redis_client = Redis.from_url(settings.redis.url, decode_responses=True)
redis_client = Redis.from_url(
settings.redis.url,
decode_responses=True,
socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
socket_timeout=settings.redis.socket_timeout_seconds,
health_check_interval=settings.redis.health_check_interval_seconds,
retry_on_timeout=True,
)
def _ping_redis() -> None:
redis_client.ping()
@@ -249,8 +262,12 @@ def _load_sync_checkpoint(redis_client: Redis) -> dict[str, object] | None:
data = json.loads(str(raw))
except Exception:
logger.warning("Failed to decode sync listing checkpoint", exc_info=True)
_clear_sync_checkpoint(redis_client)
return None
return data if isinstance(data, dict) else None
if not isinstance(data, dict):
_clear_sync_checkpoint(redis_client)
return None
return data
def _save_sync_checkpoint(
@@ -266,7 +283,10 @@ def _save_sync_checkpoint(
payload = {
"status": "in_progress",
"task_id": task_id,
"last_successful_page": int(page_number),
# page_number=0 — sentinel: прошлый checkpoint признан stale,
# следующий запуск должен начать текущий scope заново с page 1.
"last_successful_page": max(0, int(page_number)),
"resume_failures": 0,
"make": make,
"model": model,
"lane": lane,
@@ -274,7 +294,11 @@ def _save_sync_checkpoint(
"updated_at": int(time.time()),
}
try:
redis_client.set(SYNC_LISTING_CHECKPOINT_KEY, json.dumps(payload))
redis_client.set(
SYNC_LISTING_CHECKPOINT_KEY,
json.dumps(payload),
ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
)
except Exception:
logger.warning("Failed to save sync listing checkpoint", exc_info=True)
@@ -286,6 +310,99 @@ def _clear_sync_checkpoint(redis_client: Redis) -> None:
logger.warning("Failed to clear sync listing checkpoint", exc_info=True)
def _bump_checkpoint_resume_failure(
redis_client: Redis,
checkpoint: dict[str, object] | None,
*,
reason: str,
) -> tuple[int, bool]:
if not checkpoint:
return 0, False
failures = int(checkpoint.get("resume_failures") or 0) + 1
if failures >= SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT:
logger.warning(
"Checkpoint resume failed %d times; deleting checkpoint and restarting from page 1 next run (reason=%s)",
failures,
reason,
)
_clear_sync_checkpoint(redis_client)
return failures, True
payload = dict(checkpoint)
payload["resume_failures"] = failures
payload["updated_at"] = int(time.time())
try:
redis_client.set(
SYNC_LISTING_CHECKPOINT_KEY,
json.dumps(payload),
ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
)
except Exception:
logger.warning("Failed to persist checkpoint resume failure counter", exc_info=True)
logger.warning(
"Checkpoint resume failure %d/%d recorded (reason=%s)",
failures,
SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT,
reason,
)
return failures, False
def _try_set_followup_pending(redis_client: Redis, *, ttl_seconds: int) -> bool:
try:
return bool(redis_client.set(SYNC_LISTING_FOLLOWUP_PENDING_KEY, "1", nx=True, ex=max(60, int(ttl_seconds))))
except Exception:
logger.warning("Failed to set follow-up pending flag", exc_info=True)
return True
def _clear_followup_pending(redis_client: Redis) -> None:
try:
redis_client.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
except Exception:
logger.warning("Failed to clear follow-up pending flag", exc_info=True)
def _bump_bootstrap_failure_streak(
redis_client: Redis,
*,
reason: str,
) -> tuple[int, bool]:
try:
streak = int(redis_client.incr(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY))
redis_client.expire(
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY,
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS,
)
except Exception:
logger.warning("Failed to update bootstrap failure streak", exc_info=True)
return 0, True
should_enqueue = streak < SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT
if should_enqueue:
logger.warning(
"Bootstrap failure streak %d/%d recorded (reason=%s)",
streak,
SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT,
reason,
)
else:
logger.error(
"Bootstrap follow-up circuit breaker opened after %d consecutive failures (reason=%s)",
streak,
reason,
)
return streak, should_enqueue
def _clear_bootstrap_failure_streak(redis_client: Redis) -> None:
try:
redis_client.delete(SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY)
except Exception:
logger.warning("Failed to clear bootstrap failure streak", exc_info=True)
def _start_lock_heartbeat(
redis_client: Redis,
key: str,
@@ -366,7 +483,26 @@ def sync_listing_task(
heartbeat_thread: Thread | None = None
force_bootstrap_full_scan = False
def _enqueue_bootstrap_followup(reason: str, delay_seconds: int = 5) -> None:
def _enqueue_bootstrap_followup(
reason: str,
delay_seconds: int = 5,
*,
count_as_failure: bool = False,
) -> None:
flag_ttl = max(lock_ttl, delay_seconds + 300)
if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
logger.info(
"Bootstrap follow-up already pending; skip enqueue (reason=%s)",
reason,
)
return
if count_as_failure:
_, should_enqueue = _bump_bootstrap_failure_streak(redis_client, reason=reason)
if not should_enqueue:
_clear_followup_pending(redis_client)
return
else:
_clear_bootstrap_failure_streak(redis_client)
try:
self.app.send_task(
"iaai_scraper.worker.tasks.sync_listing_task",
@@ -386,6 +522,7 @@ def sync_listing_task(
reason,
)
except Exception:
_clear_followup_pending(redis_client)
logger.warning("Failed to enqueue bootstrap follow-up sync", exc_info=True)
lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)
@@ -410,8 +547,9 @@ def sync_listing_task(
effective_only_new = False if force_bootstrap_full_scan else only_new
checkpoint = _load_sync_checkpoint(redis_client)
resume_from_page = 1
used_checkpoint_resume = False
if checkpoint and str(checkpoint.get("status") or "") == "in_progress":
if force_bootstrap_full_scan and checkpoint and str(checkpoint.get("status") or "") == "in_progress":
checkpoint_page = int(checkpoint.get("last_successful_page") or 0)
checkpoint_make = checkpoint.get("make")
checkpoint_model = checkpoint.get("model")
@@ -426,6 +564,7 @@ def sync_listing_task(
is_segmented_checkpoint = checkpoint_segment is not None and checkpoint_lane == lane
if checkpoint_page > 0 and (same_scope or is_segmented_checkpoint):
resume_from_page = checkpoint_page + 1
used_checkpoint_resume = True
logger.warning(
"Resuming sync_listing from page %d (segment=%s) using checkpoint",
resume_from_page,
@@ -434,6 +573,9 @@ def sync_listing_task(
elif checkpoint_page > 0:
logger.info("Ignoring stale checkpoint due to different sync parameters")
_clear_sync_checkpoint(redis_client)
elif checkpoint:
logger.info("Ignoring leftover checkpoint because full scan is already complete; next run starts from page 1")
_clear_sync_checkpoint(redis_client)
if force_bootstrap_full_scan:
logger.info(
@@ -446,6 +588,8 @@ def sync_listing_task(
effective_limit,
)
_clear_followup_pending(redis_client)
heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
redis_client,
SYNC_LISTING_LOCK_KEY,
@@ -460,7 +604,7 @@ def sync_listing_task(
use_segmented = bool(segments) and make is None and model is None
resume_from_segment = 0
if use_segmented and checkpoint and str(checkpoint.get("status") or "") == "in_progress":
if force_bootstrap_full_scan and use_segmented and checkpoint and str(checkpoint.get("status") or "") == "in_progress":
cp_segment = checkpoint.get("segment_index")
if cp_segment is not None and int(cp_segment) >= 0:
resume_from_segment = int(cp_segment)
@@ -475,14 +619,17 @@ def sync_listing_task(
only_new=effective_only_new,
start_segment=resume_from_segment,
start_page=resume_from_page,
progress_callback=lambda seg_idx, page_number: _save_sync_checkpoint(
redis_client,
task_id=task_id,
page_number=page_number,
make=None,
model=None,
lane=lane,
segment_index=seg_idx,
progress_callback=(
(lambda seg_idx, page_number: _save_sync_checkpoint(
redis_client,
task_id=task_id,
page_number=page_number,
make=None,
model=None,
lane=lane,
segment_index=seg_idx,
))
if force_bootstrap_full_scan else None
),
)
return scraper.sync_listing(
@@ -492,13 +639,16 @@ def sync_listing_task(
limit=effective_limit,
only_new=effective_only_new,
start_page=resume_from_page,
progress_callback=lambda page_number: _save_sync_checkpoint(
redis_client,
task_id=task_id,
page_number=page_number,
make=make,
model=model,
lane=lane,
progress_callback=(
(lambda page_number: _save_sync_checkpoint(
redis_client,
task_id=task_id,
page_number=page_number,
make=make,
model=model,
lane=lane,
))
if force_bootstrap_full_scan else None
),
)
@@ -509,12 +659,24 @@ def sync_listing_task(
if bootstrap_completed:
_set_full_scan_done(redis_client, True)
_clear_sync_checkpoint(redis_client)
_clear_bootstrap_failure_streak(redis_client)
logger.info("Bootstrap full scan completed; hourly schedule continues")
else:
_set_full_scan_done(redis_client, False)
listing_payload = result.get("listing") if isinstance(result.get("listing"), dict) else {}
had_progress = any(
int(result.get(key) or 0) > 0
for key in ("cars_upserted", "images_upserted", "skipped_existing", "total_discovered")
) or int(listing_payload.get("vehicles_collected") or 0) > 0
count_as_failure = str(result.get("status") or "") == "failed" and not had_progress
logger.info("Bootstrap full scan not complete yet; queuing immediate continuation")
_enqueue_bootstrap_followup("bootstrap_not_completed")
elif result.get("status") == "success":
_enqueue_bootstrap_followup(
"bootstrap_not_completed",
count_as_failure=count_as_failure,
)
elif result.get("status") in ("success", "partial_success") and bool(result.get("full_scan_completed", False)):
_clear_sync_checkpoint(redis_client)
elif not force_bootstrap_full_scan:
_clear_sync_checkpoint(redis_client)
summary = {
@@ -543,7 +705,7 @@ def sync_listing_task(
)
if force_bootstrap_full_scan:
_set_full_scan_done(redis_client, False)
_enqueue_bootstrap_followup("soft_time_limit_exceeded")
_enqueue_bootstrap_followup("soft_time_limit_exceeded", count_as_failure=False)
# Partial progress уже записан в БД через finish_sync_run.
# Не retry — следующий запуск продолжит обработку по расписанию.
return {
@@ -555,6 +717,14 @@ def sync_listing_task(
except Exception as exc:
logger.error("sync_listing_task failed: %s", exc, exc_info=True)
if force_bootstrap_full_scan and used_checkpoint_resume:
_, deleted = _bump_checkpoint_resume_failure(
redis_client,
checkpoint,
reason=str(exc),
)
if deleted:
checkpoint = None
# Retry только на не-таймаутные ошибки (сеть, БД, браузер).
try:
if force_bootstrap_full_scan:
@@ -565,7 +735,7 @@ def sync_listing_task(
logger.error("sync_listing_task max retries exceeded, giving up")
if force_bootstrap_full_scan:
_set_full_scan_done(redis_client, False)
_enqueue_bootstrap_followup("max_retries_exceeded")
_enqueue_bootstrap_followup("max_retries_exceeded", count_as_failure=True)
return {
"status": "failed",
"task_id": task_id,