fix(#3384): cian — в mark_failed уходит только измеренное, нули heartbeat не затираются; стаб пула как прод
All checks were successful
CI / changes (pull_request) Successful in 11s
CI Trade-In / browser-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI Trade-In / changes (pull_request) Successful in 9s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m0s
All checks were successful
CI / changes (pull_request) Successful in 11s
CI Trade-In / browser-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI Trade-In / changes (pull_request) Successful in 9s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m0s
Ревью нашло у цианa (в отличие от avito/домклика с живым counters.to_dict()) старый
словарь в общем except: реальные значения присваиваются уже ПОСЛЕ возврата из
backfill_cian_history, а отказ бывает и посреди неё — пул опустел между стадиями, упал
SELECT домов. Тогда поверх измеренного в запись прогона уезжали нули, и SQL-разбор
простоя (#3288/#3367) читал «к площадке не ходили» про прогон, который ходил.
Механизм оказался хуже описанного в ревью: mark_failed мержит counters (`counters ||
:counters`) только в kit-копии, а cian/avito/домклик зовут app.services.scrape_runs, где
UPDATE counters ЗАМЕНЯЕТ (scrape_runs.py:738). Поэтому «отдать только {no_proxy_stop: 1}»
стёрло бы измеренное начисто; вместо этого _heartbeat кладёт свой снимок в те же
counters (nonlocal), и в mark_failed уезжает последнее измеренное + флаг.
Тест по значению: heartbeat записал listings_processed=5, дальше пул пуст → в jsonb-
payload mark_failed должно остаться 5, а не 0 (проверяется сам payload UPDATE'а,
runs_mod настоящий). На HEAD ветки красный: `counters={'listings_processed': 0, ...,
'no_proxy_stop': 1}: нули поверх измеренных 5`.
Стаб пула приведён к проду: RealProxyProvider.acquire при пустом пуле ВОЗВРАЩАЕТ None
(scraper_adapters.py:230), а не поднимает, — исключение из провайдера глотал
`except Exception` в _acquire_lease и приходило к тому же отказу другим путём. Теперь
NoProxyAvailableError рождается там же, где в проде (browser_fetcher.py:712, ветка
`lease is None and use_pool and production`) — проверено прогоном против до-#3384
исходников: все три теста красные, трейс из _acquire_lease.
_prod_pool патчит app.core.config.settings явно + assert, что все три задачи держат тот
же синглтон: раньше патч через chb.settings выглядел настройкой одного циана.
This commit is contained in:
parent
caa3b8b056
commit
47dabec678
2 changed files with 109 additions and 13 deletions
|
|
@ -127,8 +127,18 @@ async def _execute_cian_backfill(
|
|||
"""Сигнал живости из середины батча. Best-effort: сбой heartbeat не должен
|
||||
ронять уже идущую работу — прогон в худшем случае вернётся к прежнему
|
||||
поведению (пометка 'zombie' на 6-м часу)."""
|
||||
nonlocal counters
|
||||
# #3384: снимок измеренного едет не только в БД, но и в `counters` — этот словарь
|
||||
# уезжает в mark_failed из общего except ниже, а тот counters ЗАМЕНЯЕТ
|
||||
# (scrape_runs.py:738 `counters = CAST(:counters AS jsonb)`; мерж `||` — только у
|
||||
# kit-копии, которую этот путь не зовёт). Пока снимок сюда не доезжал, любой отказ
|
||||
# ПОСЛЕ пройденной стадии (пул опустел между стадиями, упал SELECT домов) писал
|
||||
# поверх измеренного предынициализированные нули, и SQL-разбор простоя
|
||||
# (#3288/#3367) читал «к площадке не ходили» про прогон, который ходил.
|
||||
# Присваивание ДО записи в БД: сбой heartbeat'а не должен стирать сам факт замера.
|
||||
counters = _counters(progress)
|
||||
try:
|
||||
runs_mod.update_heartbeat(db, run_id, _counters(progress))
|
||||
runs_mod.update_heartbeat(db, run_id, counters)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"scheduler: cian_history_backfill run_id=%d heartbeat failed (ignored)",
|
||||
|
|
@ -136,6 +146,8 @@ async def _execute_cian_backfill(
|
|||
exc_info=True,
|
||||
)
|
||||
|
||||
# Стартовые нули: прогон виден в админке до первого прогресса. Держатся здесь ровно
|
||||
# до первого `_heartbeat` — дальше в `counters` лежит измеренное (см. выше).
|
||||
counters: dict[str, int] = {
|
||||
"listings_processed": 0,
|
||||
"listings_succeeded": 0,
|
||||
|
|
@ -215,8 +227,10 @@ async def _execute_cian_backfill(
|
|||
# самого `async with` (cian_history_backfill.py:218) мимо стоп-механики
|
||||
# внутри цикла, которая и ставит no_proxy_stop. Без этого ключа прогон,
|
||||
# который к площадке не ходил ВООБЩЕ, неотличим от любого другого падения:
|
||||
# counters нулевые, причина только в тексте, а разбор простоя идёт SQL'ём
|
||||
# по counters.no_proxy_stop (#3288/#3367). Ключ тот же, что у ветки выше.
|
||||
# причина только в тексте, а разбор простоя идёт SQL'ём по
|
||||
# counters.no_proxy_stop (#3288/#3367). Ключ тот же, что у ветки выше.
|
||||
# Флаг ДОБАВЛЯЕТСЯ к последнему снимку `_heartbeat`, а не подменяет его:
|
||||
# опустевший между стадиями пул — это отказ ПОСЛЕ реальной работы.
|
||||
counters["no_proxy_stop"] = 1
|
||||
runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
|
||||
raise
|
||||
|
|
|
|||
|
|
@ -8,17 +8,23 @@ break`), которая и пишет `counters.no_proxy_stop`. Общий `exce
|
|||
не ходил ВООБЩЕ, в SQL-разборе простоя по `counters.no_proxy_stop` (#3288/#3367) не
|
||||
находится и выглядит как произвольное падение задачи.
|
||||
|
||||
Подделка ровно одна — провайдер прокси, чей `acquire()` поднимает
|
||||
`NoProxyAvailableError` (так ведёт себя пул, когда живых узлов не осталось).
|
||||
`BrowserFetcher` НАСТОЯЩИЙ: отказ рождается там же, где в проде, и проходит весь
|
||||
рабочий тракт до `mark_failed`. Счётчик POST'ов в сайдкар считается на
|
||||
Подделка ровно одна — провайдер прокси, чей `acquire()` возвращает `None` (ровно так
|
||||
ведёт себя прод-`RealProxyProvider.acquire`, scraper_adapters.py:230, когда живых узлов
|
||||
не осталось: он НЕ поднимает исключение). Отказ поэтому рождается там же, где в проде, —
|
||||
в `BrowserFetcher._acquire_lease` (`lease is None and use_pool and production`), — и
|
||||
проходит весь рабочий тракт до `mark_failed`. Счётчик POST'ов в сайдкар считается на
|
||||
`httpx.AsyncClient.post` — «к площадке не ходили» проверяется, а не предполагается.
|
||||
|
||||
Третий тест — про соседний случай: пул опустел МЕЖДУ стадиями, то есть уже ПОСЛЕ
|
||||
реальной работы. Там проверяется, что в запись прогона не уезжают нули поверх
|
||||
измеренного (`mark_failed` у `app.services.scrape_runs` counters ЗАМЕНЯЕТ, а не мержит).
|
||||
|
||||
Сеть/БД замоканы; в сеть тест не ходит.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
|
@ -32,6 +38,7 @@ import httpx # noqa: E402
|
|||
import pytest # noqa: E402
|
||||
from scraper_kit.proxy_errors import NoProxyAvailableError # noqa: E402
|
||||
|
||||
from app.core.config import settings # noqa: E402
|
||||
from app.services import scheduler as sched # noqa: E402
|
||||
from app.tasks import avito_detail_backfill as adb # noqa: E402
|
||||
from app.tasks import cian_history_backfill as chb # noqa: E402
|
||||
|
|
@ -39,13 +46,21 @@ from app.tasks import domclick_detail_backfill as dcb # noqa: E402
|
|||
|
||||
|
||||
class _EmptyPoolProvider:
|
||||
"""Провайдер прокси, у которого не осталось живых узлов."""
|
||||
"""Провайдер прокси, у которого не осталось живых узлов — как прод.
|
||||
|
||||
`acquire` возвращает None, а не поднимает: `RealProxyProvider.acquire`
|
||||
(scraper_adapters.py:222-234) отдаёт None, когда `proxy_pool.acquire` не нашёл
|
||||
свободного узла. Разница не косметическая — исключение из провайдера
|
||||
`_acquire_lease` глотает своим `except Exception` (browser_fetcher.py:696) и
|
||||
приходит к тому же отказу другим путём, мимо ветки `lease is None`, которая и
|
||||
работает на проде.
|
||||
"""
|
||||
|
||||
def __init__(self, *_a: Any, **_kw: Any) -> None:
|
||||
pass
|
||||
|
||||
def acquire(self, source: str) -> None:
|
||||
raise NoProxyAvailableError(source)
|
||||
return None
|
||||
|
||||
def release(self, *_a: Any, **_kw: Any) -> None:
|
||||
pass
|
||||
|
|
@ -64,10 +79,20 @@ def _mock_db(n_rows: int, url: str) -> MagicMock:
|
|||
|
||||
|
||||
def _prod_pool(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Прод + включённый пул — единственная конфигурация, где отказ #2616 живой."""
|
||||
monkeypatch.setattr(chb.settings, "use_proxy_pool_browser", True)
|
||||
monkeypatch.setattr(chb.settings, "environment", "production")
|
||||
monkeypatch.setattr(chb.settings, "browser_http_endpoint", "http://browser:9000")
|
||||
"""Прод + включённый пул — единственная конфигурация, где отказ #2616 живой.
|
||||
|
||||
Патчится ОДИН объект — `app.core.config.settings`: все три задачи и
|
||||
`RealScraperConfig` (scraper_adapters.py:26) держат один и тот же синглтон, поэтому
|
||||
`monkeypatch.setattr(chb.settings, ...)` выглядел как настройка одного циана, а на
|
||||
деле настраивал и avito с домкликом. Assert ниже — чтобы зависимость не осталась
|
||||
молчаливой: разъедется импорт — тест скажет об этом, а не начнёт тихо мерить не то.
|
||||
"""
|
||||
assert chb.settings is settings, "cian берёт другой settings — патч мимо"
|
||||
assert adb.settings is settings, "avito берёт другой settings — патч мимо"
|
||||
assert dcb.settings is settings, "domclick берёт другой settings — патч мимо"
|
||||
monkeypatch.setattr(settings, "use_proxy_pool_browser", True)
|
||||
monkeypatch.setattr(settings, "environment", "production")
|
||||
monkeypatch.setattr(settings, "browser_http_endpoint", "http://browser:9000")
|
||||
|
||||
|
||||
def _mock_session_svc() -> MagicMock:
|
||||
|
|
@ -165,3 +190,60 @@ async def test_avito_empty_pool_at_start_marks_failed_with_no_proxy_stop(
|
|||
assert counters["no_proxy_stop"] == 1, f"counters={counters}: диагноз потерян"
|
||||
assert counters["attempted"] == 0, "к площадке не ходили — 0 попыток"
|
||||
post.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cian_empty_pool_between_stages_keeps_measured_counters() -> None:
|
||||
"""Циан: пул опустел ПОСЛЕ пройденной стадии — нули не должны лечь поверх замера.
|
||||
|
||||
У циана (в отличие от avito/домклика с их живым `counters.to_dict()`) в общий
|
||||
`except` приходит СТАРЫЙ словарь: реальные значения присваиваются уже после возврата
|
||||
из `backfill_cian_history`, а отказ бывает и посреди неё — пул опустел между
|
||||
стадиями, упал SELECT домов. `runs_mod` здесь настоящий
|
||||
(`app.services.scrape_runs`), и его `mark_failed` counters ЗАМЕНЯЕТ
|
||||
(`counters = CAST(:counters AS jsonb)`, scrape_runs.py:738) — то есть в записи
|
||||
прогона остаётся ровно то, что уехало последним аргументом.
|
||||
|
||||
Проверка по значению: смотрим jsonb-payload обоих UPDATE'ов (heartbeat и
|
||||
mark_failed), а не факт вызова.
|
||||
"""
|
||||
measured = 5
|
||||
|
||||
async def _fake_backfill(_db: Any, **kw: Any) -> None:
|
||||
"""Стадия объявлений прошла (heartbeat записал замер) — потом пул опустел."""
|
||||
progress = chb.CianBackfillResult()
|
||||
progress.listings_processed = measured
|
||||
progress.listings_succeeded = measured
|
||||
on_progress = kw["on_progress"]
|
||||
assert on_progress is not None, "прогон без сигнала живости — мерить нечего"
|
||||
on_progress(progress)
|
||||
raise NoProxyAvailableError("cian")
|
||||
|
||||
updates: list[dict[str, Any]] = []
|
||||
db = MagicMock()
|
||||
|
||||
def _capture(_stmt: Any, params: Any = None, *_a: Any, **_kw: Any) -> MagicMock:
|
||||
if isinstance(params, dict) and "counters" in params:
|
||||
updates.append({**params, "counters": json.loads(params["counters"])})
|
||||
return MagicMock()
|
||||
|
||||
db.execute.side_effect = _capture
|
||||
|
||||
with (
|
||||
patch.object(chb, "backfill_cian_history", _fake_backfill),
|
||||
pytest.raises(NoProxyAvailableError),
|
||||
):
|
||||
await sched._execute_cian_backfill(db, run_id=3384, params={"batch_size": 5})
|
||||
|
||||
beats = [u["counters"] for u in updates if "error" not in u]
|
||||
finals = [u["counters"] for u in updates if "error" in u]
|
||||
assert beats and beats[-1]["listings_processed"] == measured, (
|
||||
f"heartbeat не записал замер — проверять нечего: {beats}"
|
||||
)
|
||||
assert len(finals) == 1, f"ожидался ровно один mark_failed: {updates}"
|
||||
final = finals[0]
|
||||
assert final.get("no_proxy_stop") == 1, f"counters={final}: диагноз потерян"
|
||||
assert final["listings_processed"] == measured, (
|
||||
f"counters={final}: нули поверх измеренных {measured} — SQL-разбор простоя "
|
||||
"(#3288/#3367) прочитает «к площадке не ходили» про прогон, который ходил"
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue