From ae04dcb4b91ea01e34545bdcf93695d699d9a6d3 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 22:12:00 +0300 Subject: [PATCH] perf(cadastre): freshness skip + grid-walk heartbeat + reduce WMS log noise MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three operational improvements after ekb_full v3 completed at 99.1%: Fix H — per-quarter freshness skip (scrape_cadastre.py): - enqueue_cadastre_harvest queries cad_quarter_stats.fetched_at per quarter - Skip quarters where fetched_at >= NOW() - skip_fresh_hours (default 1h) - extra_config.skip_fresh_hours=0 disables skip (full re-fetch) - Saves ~30k NSPD calls on repeated runs of same scope Fix I — heartbeat in grid-walk loop (bulk_harvest.py:_grid_walk_category): - 225 cells x 70-300ms = 15-75s heartbeat silence - Pass update_progress callback to _grid_walk_category - Call every heartbeat_every (50) cells with phase progress info - Prevents cleanup_zombies false-positive cancellation Fix J — NSPD WMS HTTP 500 noise to debug (bulk_harvest.py:_grid_walk_category): - NSPD WMS endpoint frequently returns HTTP 500 (server-side issue) - Was: logger.warning each, 200+ warnings per quarter - Now: logger.debug (visible via --log-level=DEBUG) 40 tests passing. --- backend/app/services/cadastre/bulk_harvest.py | 27 +++++++++- backend/app/workers/tasks/scrape_cadastre.py | 49 +++++++++++++++++-- 2 files changed, 71 insertions(+), 5 deletions(-) diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py index 173a328d..50d12ffc 100644 --- a/backend/app/services/cadastre/bulk_harvest.py +++ b/backend/app/services/cadastre/bulk_harvest.py @@ -218,6 +218,7 @@ async def harvest_quarter( client=client, quarter=quarter, layer_id=cat_id, + update_progress=update_progress, ) result.grid_walk_requests += n_requests @@ -276,12 +277,19 @@ async def _grid_walk_category( layer_id: int, grid_size: int = 15, tile_size: int = 512, + update_progress: Callable[[dict[str, Any]], None] | None = None, + heartbeat_every: int = 50, ) -> tuple[int, int]: """Grid-walk layer_id в bbox квартала. Генерирует grid_size×grid_size ячеек, для каждой делает wms_feature_info. Дедуплицирует по cad_num, upsert'ит новые объекты. + Args: + update_progress: Fix I — callback для heartbeat внутри grid-walk loop. + 225 cells × ~70ms = ~15-75s без heartbeat, zombie-cleanup ложно + помечал worker как мёртвого. Вызываем каждые heartbeat_every cells. + Returns: (upserted_count, requests_count) """ @@ -300,7 +308,19 @@ async def _grid_walk_category( upserted = 0 requests = 0 - for cell_bbox, click_xy in grid_points: + for idx, (cell_bbox, click_xy) in enumerate(grid_points): + # Fix I: heartbeat каждые N cells — обновляет cadastre_jobs.heartbeat_at + # чтобы cleanup_zombies (5min stale check) не cancel'нул живой worker. + if update_progress is not None and idx > 0 and idx % heartbeat_every == 0: + update_progress( + { + "phase": f"grid_walk_{layer_id}_cell_{idx}", + "quarter": quarter, + "grid_cells_done": idx, + "grid_cells_total": len(grid_points), + } + ) + try: features = await client.wms_feature_info( layer_id=layer_id, @@ -311,7 +331,10 @@ async def _grid_walk_category( ) requests += 1 except Exception as e: - logger.warning( + # Fix J: NSPD WMS возвращает HTTP 500 на десятки cells per quarter — + # это server-side noise, не наш bug. logger.debug чтобы не засорять + # prod логи (раньше было warning → 200+ warnings per quarter). + logger.debug( "_grid_walk_category: wms_feature_info error layer=%d quarter=%s: %s", layer_id, quarter, diff --git a/backend/app/workers/tasks/scrape_cadastre.py b/backend/app/workers/tasks/scrape_cadastre.py index 27dd9ec9..0cf57026 100644 --- a/backend/app/workers/tasks/scrape_cadastre.py +++ b/backend/app/workers/tasks/scrape_cadastre.py @@ -357,22 +357,65 @@ def enqueue_cadastre_harvest(self: Any, job_id: int) -> dict[str, Any]: logger.warning("enqueue_cadastre_harvest: job=%d имеет пустой список quarters", job_id) return {"job_id": job_id, "enqueued": 0} + # Fix H: per-quarter freshness skip — не переобрабатывать quarters с свежим + # cad_quarter_stats.fetched_at (последний успешный обход 0: + threshold = _dt.datetime.now(_dt.UTC) - _dt.timedelta(hours=skip_fresh_hours) + rows = ( + db.execute( + text( + "SELECT cad_number FROM cad_quarter_stats " + "WHERE cad_number = ANY(CAST(:qs AS text[])) " + "AND fetched_at >= :t" + ), + {"qs": quarters, "t": threshold}, + ) + .scalars() + .all() + ) + skipped_fresh = set(rows) + if skipped_fresh: + # Корректируем targets_total — skipped quarters не входят в прогресс + db.execute( + text( + "UPDATE cadastre_jobs SET targets_total = :new_total, " + "targets_skipped = COALESCE(targets_skipped, 0) + :sk " + "WHERE job_id = :id" + ), + { + "new_total": len(quarters) - len(skipped_fresh), + "sk": len(skipped_fresh), + "id": job_id, + }, + ) + # Обновить статус на running _update_job(db, job_id, status="running", started_at=_dt.datetime.now(_dt.UTC)) db.commit() + enqueued = 0 for quarter in quarters: + if quarter in skipped_fresh: + continue bulk_harvest_quarter_task.apply_async( args=[quarter, job_id], queue=CADASTRE_QUEUE, ) + enqueued += 1 logger.info( - "enqueue_cadastre_harvest: job=%d enqueued %d quarters", + "enqueue_cadastre_harvest: job=%d enqueued %d (skipped_fresh=%d, threshold=%dh)", job_id, - len(quarters), + enqueued, + len(skipped_fresh), + skip_fresh_hours, ) - return {"job_id": job_id, "enqueued": len(quarters)} + return {"job_id": job_id, "enqueued": enqueued, "skipped_fresh": len(skipped_fresh)} finally: db.close()