fix(tradein/imv): «временная» ошибка снова временная — 1390 домов возвращаются в очередь #2843
8 changed files with 516 additions and 23 deletions
|
|
@ -2230,7 +2230,10 @@ class HouseIMVBackfillRequest(BaseModel):
|
|||
)
|
||||
only_status: str = Field(
|
||||
default="pending",
|
||||
description="Обрабатывать дома с этим imv_status. 'transient_error' — retry.",
|
||||
description=(
|
||||
"Обрабатывать дома с этим imv_status. По умолчанию 'pending' + автоповтор "
|
||||
"'transient_error' на половине пакета; явное значение = только этот статус."
|
||||
),
|
||||
)
|
||||
house_id: int | None = Field(
|
||||
default=None,
|
||||
|
|
@ -2267,7 +2270,12 @@ async def scrape_house_imv_backfill(
|
|||
batch_size: сколько домов обработать за запуск (default 50).
|
||||
request_delay_sec: пауза между IMV-вызовами (default 5s). ВАЖНО: Avito IMV
|
||||
реагирует на частые запросы с datacenter-IP. Не снижать < 3s.
|
||||
only_status: по умолчанию 'pending'. Для retry failed — 'transient_error'.
|
||||
only_status: по умолчанию 'pending' — и тогда половина пакета сама уходит на
|
||||
повтор домов в 'transient_error' с непотраченным лимитом попыток (#2674:
|
||||
раньше повтор существовал только как этот параметр, и за 41 прогон его
|
||||
не передали ни разу — 1390 домов застряли навсегда). Явное значение
|
||||
отключает автоповтор и обрабатывает РОВНО указанный статус, включая
|
||||
дома, исчерпавшие лимит (imv_transient_attempts >= 3).
|
||||
house_id: обработать один дом (debug).
|
||||
|
||||
Примечание по прокси: Avito IMV использует собственную curl_cffi-сессию.
|
||||
|
|
|
|||
|
|
@ -411,7 +411,8 @@ def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluat
|
|||
UPDATE houses
|
||||
SET imv_status = 'ok',
|
||||
last_imv_attempt_at = NOW(),
|
||||
imv_error_reason = NULL
|
||||
imv_error_reason = NULL,
|
||||
imv_transient_attempts = 0
|
||||
WHERE id = :hid
|
||||
"""),
|
||||
{"hid": house_id},
|
||||
|
|
@ -424,12 +425,19 @@ def _mark_status(
|
|||
status: str,
|
||||
reason: str | None = None,
|
||||
) -> None:
|
||||
# #2674: счётчик растёт ТОЛЬКО на transient_error — это «сколько раз подряд дом
|
||||
# падал по временной причине», а не «сколько раз его трогали». no_params /
|
||||
# no_address / not_found счётчик не двигают: они не занимают retry-слот.
|
||||
db.execute(
|
||||
text("""
|
||||
UPDATE houses
|
||||
SET imv_status = :s,
|
||||
last_imv_attempt_at = NOW(),
|
||||
imv_error_reason = :r
|
||||
imv_error_reason = :r,
|
||||
imv_transient_attempts = CASE
|
||||
WHEN :s = 'transient_error' THEN imv_transient_attempts + 1
|
||||
ELSE imv_transient_attempts
|
||||
END
|
||||
WHERE id = :hid
|
||||
"""),
|
||||
{"hid": house_id, "s": status, "r": reason},
|
||||
|
|
@ -443,6 +451,104 @@ _IMVStatus = Literal[
|
|||
"ok", "no_params", "no_address", "not_found", "auth_error", "transient", "error"
|
||||
]
|
||||
|
||||
# #2674: сколько раз подряд дом может упасть в transient_error, прежде чем
|
||||
# перестанет занимать retry-слот. Число из замера: после починки сайдкара (04.08)
|
||||
# доля отказов на попытку — 2/27 и 3/25 (прогоны 3708/3467), т.е. ~10%. На 1039
|
||||
# застрявших это ~104 повторных отказа на первом проходе, ~10 на втором, ~1 на
|
||||
# третьем. Порог 3 стоит максимум ~115 слотов ВСЕГО (≈2 прогона) и гарантирует,
|
||||
# что дом со СВОЕЙ (не инфраструктурной) причиной не крутится в пакете вечно.
|
||||
# Исчерпавшие лимит не исчезают из наблюдаемости: они остаются imv_status=
|
||||
# 'transient_error' и считаются как
|
||||
# WHERE imv_status='transient_error' AND imv_transient_attempts >= 3.
|
||||
_MAX_TRANSIENT_ATTEMPTS = 3
|
||||
|
||||
# Доля пакета под повтор transient_error. Половина — потому что остальные слоты
|
||||
# после #2674 достаются ТОЛЬКО домам, по которым реально будет запрос к площадке
|
||||
# (см. _premark_unusable): раньше из 50 слотов до площадки доходили 17 (замер
|
||||
# головы очереди на 12.08), так что pending на половине пакета всё равно идёт
|
||||
# быстрее, чем на целом до правки.
|
||||
_RETRY_SLOTS_SHARE = 0.5
|
||||
|
||||
# Дом без пригодных параметров backfill всё равно пометит no_params — но только
|
||||
# заплатив слотом пакета и паузой request_delay_sec. Тот же вердикт берётся одним
|
||||
# запросом: нет ни одного объявления с rooms+area (pick_lot_params вернёт {}) ИЛИ
|
||||
# не из чего взять house_type (_map_house_type вернёт None → «unknown house_type»).
|
||||
# Причины пишем ТЕМИ ЖЕ строками, что и поштучный путь, — старые разрезы по
|
||||
# imv_error_reason продолжают работать.
|
||||
# Условие сознательно УЖЕ питоновского: normalize_house_type схлопывает в None ещё
|
||||
# и нераспознанный вокабуляр ('other', 'wireframe'), который тут остаётся текстом.
|
||||
# Промахнуться можно только в безопасную сторону — пометить меньше, чем пометил бы
|
||||
# поштучный путь.
|
||||
_PREMARK_UNUSABLE_SQL = text("""
|
||||
WITH unusable AS (
|
||||
SELECT h.id,
|
||||
CASE WHEN NOT EXISTS (
|
||||
SELECT 1 FROM listings l
|
||||
WHERE l.house_id_fk = h.id
|
||||
AND l.rooms IS NOT NULL
|
||||
AND l.area_m2 IS NOT NULL)
|
||||
THEN 'no listings with rooms+area'
|
||||
ELSE 'unknown house_type'
|
||||
END AS reason
|
||||
FROM houses h
|
||||
WHERE h.imv_status = ANY(CAST(:statuses AS text[]))
|
||||
AND h.lat IS NOT NULL
|
||||
AND h.lon IS NOT NULL
|
||||
AND h.address IS NOT NULL
|
||||
AND (
|
||||
NOT EXISTS (
|
||||
SELECT 1 FROM listings l
|
||||
WHERE l.house_id_fk = h.id
|
||||
AND l.rooms IS NOT NULL
|
||||
AND l.area_m2 IS NOT NULL)
|
||||
OR COALESCE(
|
||||
NULLIF(TRIM((
|
||||
SELECT mode() WITHIN GROUP (ORDER BY l.house_type)
|
||||
FROM listings l
|
||||
WHERE l.house_id_fk = h.id
|
||||
AND l.rooms IS NOT NULL
|
||||
AND l.area_m2 IS NOT NULL)), ''),
|
||||
NULLIF(TRIM(h.house_type), '')
|
||||
) IS NULL
|
||||
)
|
||||
)
|
||||
UPDATE houses
|
||||
SET imv_status = 'no_params',
|
||||
last_imv_attempt_at = NOW(),
|
||||
imv_error_reason = unusable.reason
|
||||
FROM unusable
|
||||
WHERE houses.id = unusable.id
|
||||
""")
|
||||
|
||||
# Основная очередь: один статус, как и было (only_status — публичный параметр
|
||||
# admin-API, семантику не трогаем).
|
||||
_QUEUE_SQL = text("""
|
||||
SELECT id, address, full_address, lat, lon
|
||||
FROM houses
|
||||
WHERE imv_status = :status
|
||||
AND lat IS NOT NULL
|
||||
AND lon IS NOT NULL
|
||||
AND address IS NOT NULL
|
||||
ORDER BY last_imv_attempt_at NULLS FIRST, id
|
||||
LIMIT :batch
|
||||
""")
|
||||
|
||||
# Retry-очередь (#2674). Отдельный запрос, а не OR к основной: у pending
|
||||
# last_imv_attempt_at всегда NULL, поэтому при общем ORDER BY ... NULLS FIRST
|
||||
# transient_error не попал бы в пакет, пока не кончится pending (по замеру
|
||||
# 12.08 — 5747 домов ≈ год). Отдельная квота = отдельный проход.
|
||||
_RETRY_QUEUE_SQL = text("""
|
||||
SELECT id, address, full_address, lat, lon
|
||||
FROM houses
|
||||
WHERE imv_status = 'transient_error'
|
||||
AND imv_transient_attempts < :max_attempts
|
||||
AND lat IS NOT NULL
|
||||
AND lon IS NOT NULL
|
||||
AND address IS NOT NULL
|
||||
ORDER BY last_imv_attempt_at NULLS FIRST, id
|
||||
LIMIT :batch
|
||||
""")
|
||||
|
||||
|
||||
@dataclass
|
||||
class HouseIMVBackfillResult:
|
||||
|
|
@ -452,6 +558,23 @@ class HouseIMVBackfillResult:
|
|||
errors: int = 0
|
||||
duration_sec: float = field(default=0.0)
|
||||
status_counts: dict[str, int] = field(default_factory=dict)
|
||||
# #2674: сколько домов пакета пришло из retry-очереди transient_error и
|
||||
# сколько помечено no_params до пакета (без запроса к площадке).
|
||||
retried: int = 0
|
||||
premarked: int = 0
|
||||
|
||||
|
||||
def _premark_unusable(db: Session, statuses: list[str]) -> int:
|
||||
"""Пометить no_params дома, по которым запрос к площадке невозможен. → сколько.
|
||||
|
||||
Не новое поведение, а тот же вердикт _process_one_house одним запросом: на
|
||||
12.08 в очереди 1925 таких домов из 5143 (113 без объявлений с rooms+area,
|
||||
1812 без house_type) — каждый занимал слот пакета и паузу, чтобы получить
|
||||
ответ, который виден в SQL.
|
||||
"""
|
||||
res = db.execute(_PREMARK_UNUSABLE_SQL, {"statuses": statuses})
|
||||
db.commit()
|
||||
return int(res.rowcount or 0)
|
||||
|
||||
|
||||
def _beat(heartbeat: Callable[[], None] | None) -> None:
|
||||
|
|
@ -480,7 +603,11 @@ async def backfill_house_imv(
|
|||
batch_size: max houses to process (ignored when house_id given).
|
||||
request_delay_sec: sleep between Avito API calls (default 5s — anti-bot).
|
||||
only_status: process houses with this imv_status (default 'pending').
|
||||
Use 'transient_error' to retry failures.
|
||||
Use 'transient_error' to retry failures. При значении по умолчанию
|
||||
часть пакета (_RETRY_SLOTS_SHARE) автоматически уходит на повтор
|
||||
transient_error с непотраченным лимитом попыток (#2674) — явно
|
||||
переданный only_status этот проход отключает, оператор получает
|
||||
ровно то, что попросил, включая исчерпавшие лимит дома.
|
||||
house_id: process a single specific house (debug).
|
||||
heartbeat: optional callback дёргается каждые _HEARTBEAT_EVERY_N_HOUSES
|
||||
домов — caller обновляет scrape_runs.heartbeat_at, чтобы reap_zombies
|
||||
|
|
@ -509,23 +636,45 @@ async def backfill_house_imv(
|
|||
.all()
|
||||
)
|
||||
else:
|
||||
rows = (
|
||||
# Повторный проход только на расписании (only_status по умолчанию): явный
|
||||
# only_status от оператора — это ручной запрос ровно одного статуса.
|
||||
retry_lane = only_status == "pending"
|
||||
|
||||
statuses = [only_status] + (["transient_error"] if retry_lane else [])
|
||||
result.premarked = _premark_unusable(db, statuses)
|
||||
if result.premarked:
|
||||
logger.info(
|
||||
"house_imv_backfill: %d домов помечены no_params до пакета (нет rooms+area "
|
||||
"или house_type) — слоты пакета не потрачены",
|
||||
result.premarked,
|
||||
)
|
||||
|
||||
retry_rows: list = []
|
||||
if retry_lane:
|
||||
retry_rows = (
|
||||
db.execute(
|
||||
_RETRY_QUEUE_SQL,
|
||||
{
|
||||
"max_attempts": _MAX_TRANSIENT_ATTEMPTS,
|
||||
"batch": int(batch_size * _RETRY_SLOTS_SHARE),
|
||||
},
|
||||
)
|
||||
.mappings()
|
||||
.all()
|
||||
)
|
||||
result.retried = len(retry_rows)
|
||||
|
||||
# Недобор retry-очереди (она кончится раньше pending: 1039 против 3218 на
|
||||
# 12.08) возвращается pending — пакет не простаивает.
|
||||
fresh_rows = (
|
||||
db.execute(
|
||||
text("""
|
||||
SELECT id, address, full_address, lat, lon
|
||||
FROM houses
|
||||
WHERE imv_status = :status
|
||||
AND lat IS NOT NULL
|
||||
AND lon IS NOT NULL
|
||||
AND address IS NOT NULL
|
||||
ORDER BY last_imv_attempt_at NULLS FIRST, id
|
||||
LIMIT :batch
|
||||
"""),
|
||||
{"status": only_status, "batch": batch_size},
|
||||
_QUEUE_SQL,
|
||||
{"status": only_status, "batch": max(batch_size - result.retried, 0)},
|
||||
)
|
||||
.mappings()
|
||||
.all()
|
||||
)
|
||||
rows = list(fresh_rows) + list(retry_rows)
|
||||
|
||||
result.checked = len(rows)
|
||||
if not rows:
|
||||
|
|
@ -534,9 +683,11 @@ async def backfill_house_imv(
|
|||
return result
|
||||
|
||||
logger.info(
|
||||
"house_imv_backfill: %d houses (status=%r delay=%.1fs)",
|
||||
"house_imv_backfill: %d houses (status=%r retry=%d premarked=%d delay=%.1fs)",
|
||||
result.checked,
|
||||
only_status,
|
||||
result.retried,
|
||||
result.premarked,
|
||||
request_delay_sec,
|
||||
)
|
||||
|
||||
|
|
@ -610,11 +761,14 @@ async def backfill_house_imv(
|
|||
|
||||
result.duration_sec = time.time() - t0
|
||||
logger.info(
|
||||
"house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
|
||||
"house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d "
|
||||
"retried=%d premarked=%d %.1fs %s",
|
||||
result.checked,
|
||||
result.saved,
|
||||
result.skipped,
|
||||
result.errors,
|
||||
result.retried,
|
||||
result.premarked,
|
||||
result.duration_sec,
|
||||
result.status_counts,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -447,6 +447,12 @@ async def _job_house_imv_backfill(
|
|||
# алерт — но пустая очередь при ежедневном расписании это и правда сигнал.
|
||||
"total_seen": result.checked,
|
||||
"new_count": result.saved,
|
||||
# #2674: из скольких слотов пакета взяты дома на ПОВТОР (transient_error)
|
||||
# и сколько домов ушло в no_params до пакета одним запросом. Без этих
|
||||
# двух счётчиков в scrape_runs.counters проверить, что застрявшие
|
||||
# действительно возвращаются в очередь, можно только по houses.
|
||||
"retried": result.retried,
|
||||
"premarked": result.premarked,
|
||||
}
|
||||
# Честный статус (#2674, тот же класс, что #2670/#2657): успех — это
|
||||
# «сделали то, что собирались», а не «не поймали известное исключение».
|
||||
|
|
|
|||
|
|
@ -0,0 +1,41 @@
|
|||
-- 258_houses_imv_transient_attempts.sql
|
||||
-- Счётчик подряд идущих временных отказов домовой оценки Авито (эпик #2674).
|
||||
--
|
||||
-- ЗАЧЕM. imv_status='transient_error' был состоянием БЕЗ ВЫХОДА: очередь
|
||||
-- backfill'а выбирает ровно один статус за прогон (only_status, по умолчанию
|
||||
-- 'pending'), и за всю историю (41 прогон, 26.06–11.08) ни один не был запущен
|
||||
-- с другим значением. На 12.08.2026 в этом статусе лежали 1390 домов, 1337 из
|
||||
-- них — с причиной «503/500 от tradein-browser:3000/fetch-json» или «All
|
||||
-- connection attempts failed», то есть с ИНФРАСТРУКТУРНОЙ причиной, которой
|
||||
-- больше нет (сайдкар починен #2698; за 7 суток до 12.08 в его access-логе
|
||||
-- 108 из 108 POST /fetch-json = 200).
|
||||
--
|
||||
-- Сервис теперь отдаёт часть пакета на повтор transient_error автоматически
|
||||
-- (house_imv_backfill._RETRY_QUEUE_SQL). Этот счётчик — условие ВЫХОДА из
|
||||
-- повтора: дом, падающий по своей причине, а не по инфраструктурной, перестаёт
|
||||
-- занимать слот пакета после _MAX_TRANSIENT_ATTEMPTS (3) подряд.
|
||||
--
|
||||
-- Наблюдаемость НЕ переименовывается: статус остаётся 'transient_error',
|
||||
-- прежние разрезы по imv_status/imv_error_reason работают как работали, а
|
||||
-- «застряли окончательно» — это
|
||||
-- SELECT count(*) FROM houses
|
||||
-- WHERE imv_status='transient_error' AND imv_transient_attempts >= 3;
|
||||
--
|
||||
-- Индекс не добавляем: houses_imv_status_idx (064) уже частичный по
|
||||
-- imv_status IN ('pending','transient_error') с сортировкой по
|
||||
-- last_imv_attempt_at — фильтр по счётчику остаётся остаточным условием на
|
||||
-- выборке в тысячи строк.
|
||||
|
||||
BEGIN;
|
||||
|
||||
SET LOCAL lock_timeout = '5s';
|
||||
|
||||
ALTER TABLE houses
|
||||
ADD COLUMN IF NOT EXISTS imv_transient_attempts smallint NOT NULL DEFAULT 0;
|
||||
|
||||
COMMENT ON COLUMN houses.imv_transient_attempts IS
|
||||
'Сколько раз подряд домовая IMV-оценка падала в transient_error. '
|
||||
'Растёт только на transient_error, обнуляется успехом. '
|
||||
'>= 3 — дом больше не берётся в автоматический повтор (эпик #2674).';
|
||||
|
||||
COMMIT;
|
||||
|
|
@ -246,3 +246,4 @@
|
|||
251_listings_drop_ceiling_height.sql
|
||||
254_listings_backfill_avito_rating_glued_address.sql
|
||||
257_listings_backfill_yandex_source_url.sql
|
||||
258_houses_imv_transient_attempts.sql
|
||||
|
|
|
|||
|
|
@ -500,7 +500,11 @@ async def test_backfill_house_imv_ok_path():
|
|||
patch("app.services.house_imv_backfill.save_imv_result") as mock_save,
|
||||
):
|
||||
mock_mappings = MagicMock()
|
||||
mock_mappings.all.return_value = houses
|
||||
# #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
|
||||
# затем основную по only_status. Фейк отвечает одним и тем же списком на любой
|
||||
# SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
|
||||
# придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
|
||||
mock_mappings.all.side_effect = [[], houses]
|
||||
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
||||
|
||||
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
||||
|
|
@ -539,7 +543,11 @@ async def test_backfill_house_imv_no_params():
|
|||
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
|
||||
):
|
||||
mock_mappings = MagicMock()
|
||||
mock_mappings.all.return_value = houses
|
||||
# #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
|
||||
# затем основную по only_status. Фейк отвечает одним и тем же списком на любой
|
||||
# SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
|
||||
# придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
|
||||
mock_mappings.all.side_effect = [[], houses]
|
||||
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
||||
|
||||
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
||||
|
|
@ -593,7 +601,11 @@ async def test_backfill_house_imv_not_found():
|
|||
patch("app.services.house_imv_backfill._mark_status") as mock_mark,
|
||||
):
|
||||
mock_mappings = MagicMock()
|
||||
mock_mappings.all.return_value = houses
|
||||
# #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
|
||||
# затем основную по only_status. Фейк отвечает одним и тем же списком на любой
|
||||
# SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
|
||||
# придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
|
||||
mock_mappings.all.side_effect = [[], houses]
|
||||
mock_db.execute.return_value.mappings.return_value = mock_mappings
|
||||
|
||||
result = await backfill_house_imv(mock_db, batch_size=10, request_delay_sec=0.0)
|
||||
|
|
|
|||
|
|
@ -138,7 +138,11 @@ async def test_unknown_house_type_skips_request_and_marks_house() -> None:
|
|||
patch.object(hib, "evaluate_via_imv", new_callable=AsyncMock) as mock_eval,
|
||||
patch.object(hib, "_mark_status") as mock_mark,
|
||||
):
|
||||
db.execute.return_value.mappings.return_value.all.return_value = houses
|
||||
# #2674: backfill делает ДВЕ выборки — сначала retry-очередь transient_error,
|
||||
# затем основную по only_status. Фейк отвечает одним и тем же списком на любой
|
||||
# SELECT, поэтому очередь повтора отдаём пустой явно — иначе один и тот же дом
|
||||
# придёт в пакет дважды (артефакт фейка, в БД статусы не пересекаются).
|
||||
db.execute.return_value.mappings.return_value.all.side_effect = [[], houses]
|
||||
result = await hib.backfill_house_imv(db, batch_size=10, request_delay_sec=0.0)
|
||||
|
||||
mock_eval.assert_not_called()
|
||||
|
|
|
|||
267
tradein-mvp/backend/tests/test_house_imv_retry_stuck.py
Normal file
267
tradein-mvp/backend/tests/test_house_imv_retry_stuck.py
Normal file
|
|
@ -0,0 +1,267 @@
|
|||
"""#2674: «временная» ошибка домовой IMV-оценки снова означает временную.
|
||||
|
||||
Что было. Очередь backfill'а выбирает РОВНО ОДИН imv_status за прогон
|
||||
(`only_status`, дефолт 'pending'). Возможность повтора существовала и была
|
||||
задокументирована в admin-API, но за всю историю (41 прогон, 26.06–11.08.2026)
|
||||
её не передал никто: все прогоны ушли с 'pending'. Дома, упавшие по причине,
|
||||
которой больше нет (503/500 от браузерного сайдкара, #2698), лежали в
|
||||
'transient_error' без единого шанса вернуться — 1390 штук на 12.08.2026.
|
||||
|
||||
Что проверяем — по существу, а не по настройке:
|
||||
* дом в 'transient_error' ПОПАДАЕТ в пакет прогона с дефолтными параметрами
|
||||
(тот же вызов, что делает расписание) — на старом коде он не попадает;
|
||||
* pending при этом из пакета НЕ вытесняется;
|
||||
* дом, исчерпавший лимит попыток, в автоматический повтор не берётся, а
|
||||
ЯВНЫЙ only_status='transient_error' (ручной запрос оператора) берёт и его;
|
||||
* дом без пригодных параметров помечается no_params БЕЗ запроса к площадке.
|
||||
|
||||
Живой Postgres обязателен: проверяется поведение SQL-выборки, а не то, какие
|
||||
строки вернул мок. Без БД тест самоскипается (в CI БД есть — ci-tradein.yml
|
||||
поднимает Postgres и накатывает схему).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from app.services import house_imv_backfill as hib
|
||||
|
||||
# Фикстуры живут в собственном диапазоне id (ср. tests/test_house_dedup_merge.py).
|
||||
_H_STUCK = 900601 # transient_error, попыток 0 → обязан вернуться в очередь
|
||||
_H_EXHAUSTED = 900602 # transient_error, попытки исчерпаны → в автоповтор не берём
|
||||
_H_PENDING = 900603 # pending → не должен быть вытеснен retry-очередью
|
||||
_H_NO_PARAMS = 900604 # pending без объявлений с rooms+area → no_params без запроса
|
||||
_IDS = (_H_STUCK, _H_EXHAUSTED, _H_PENDING, _H_NO_PARAMS)
|
||||
|
||||
|
||||
def _live_session() -> Any | None:
|
||||
"""Session к достижимому не-заглушечному Postgres, иначе None."""
|
||||
try:
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
|
||||
if not dsn or "localhost:5432/test" in dsn:
|
||||
return None
|
||||
engine = create_engine(dsn, future=True)
|
||||
conn = engine.connect()
|
||||
conn.execute(text("SELECT 1"))
|
||||
conn.close()
|
||||
return sessionmaker(bind=engine, future=True)()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _seed(db: Any) -> None:
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
# url/source/ext_house_id — NOT NULL в houses (002_core_tables).
|
||||
db.execute(
|
||||
_t(
|
||||
"INSERT INTO houses (id, source, ext_house_id, url, address, lat, lon, "
|
||||
" imv_status, imv_transient_attempts, last_imv_attempt_at) VALUES "
|
||||
"(:s, 'avito', 'EXT-2674-S', 'u', 'тестдом 2674, 1', 56.84, 60.60, "
|
||||
" 'transient_error', 0, NOW() - INTERVAL '10 days'),"
|
||||
"(:x, 'avito', 'EXT-2674-X', 'u', 'тестдом 2674, 2', 56.84, 60.60, "
|
||||
" 'transient_error', :cap, NOW() - INTERVAL '10 days'),"
|
||||
"(:p, 'avito', 'EXT-2674-P', 'u', 'тестдом 2674, 3', 56.84, 60.60, "
|
||||
" 'pending', 0, NULL),"
|
||||
"(:n, 'avito', 'EXT-2674-N', 'u', 'тестдом 2674, 4', 56.84, 60.60, "
|
||||
" 'pending', 0, NULL)"
|
||||
),
|
||||
{
|
||||
"s": _H_STUCK,
|
||||
"x": _H_EXHAUSTED,
|
||||
"p": _H_PENDING,
|
||||
"n": _H_NO_PARAMS,
|
||||
# Литерал, а не hib._MAX_TRANSIENT_ATTEMPTS: подготовка теста обязана
|
||||
# работать и на СТАРОМ коде, иначе красный прогон получится про
|
||||
# отсутствующий атрибут, а не про застрявший дом. Синхронность
|
||||
# литерала и константы стережёт test_transient_attempts_counter_*.
|
||||
"cap": 3,
|
||||
},
|
||||
)
|
||||
# Объявления с rooms+area+house_type — у всех, КРОМЕ _H_NO_PARAMS.
|
||||
# source_url / dedup_hash / price_rub — NOT NULL, dedup_hash UNIQUE.
|
||||
db.execute(
|
||||
_t(
|
||||
"INSERT INTO listings (id, source, source_url, source_id, dedup_hash, price_rub, "
|
||||
" house_id_fk, rooms, area_m2, floor, total_floors, house_type) "
|
||||
"VALUES "
|
||||
"(:l1, 'avito', 'http://t/2674/1', 'L-2674-1', 'dh-2674-1', 5000000, :s, "
|
||||
" 2, 52.0, 3, 9, 'brick'),"
|
||||
"(:l2, 'avito', 'http://t/2674/2', 'L-2674-2', 'dh-2674-2', 5000000, :x, "
|
||||
" 2, 52.0, 3, 9, 'brick'),"
|
||||
"(:l3, 'avito', 'http://t/2674/3', 'L-2674-3', 'dh-2674-3', 5000000, :p, "
|
||||
" 2, 52.0, 3, 9, 'brick')"
|
||||
),
|
||||
{
|
||||
"l1": 910601,
|
||||
"l2": 910602,
|
||||
"l3": 910603,
|
||||
"s": _H_STUCK,
|
||||
"x": _H_EXHAUSTED,
|
||||
"p": _H_PENDING,
|
||||
},
|
||||
)
|
||||
db.commit()
|
||||
|
||||
|
||||
def _cleanup(db: Any) -> None:
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
db.rollback()
|
||||
db.execute(_t("DELETE FROM listings WHERE id IN (910601, 910602, 910603)"))
|
||||
db.execute(_t("DELETE FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"), {"ids": list(_IDS)})
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
|
||||
async def _run(db: Any, **kwargs: Any) -> Any:
|
||||
"""Прогнать backfill с замоканным вызовом площадки.
|
||||
|
||||
Мокаем ровно сетевую границу: evaluate_via_imv поднимает «адрес не найден»,
|
||||
поэтому дом уходит в not_found и путь сохранения (три таблицы) не нужен —
|
||||
проверяем именно ВЫБОРКУ. Флаг браузерного фетчера гасим, чтобы не поднимать
|
||||
сайдкар.
|
||||
"""
|
||||
err = hib.IMVAddressNotFoundError("address not found (test)")
|
||||
with (
|
||||
patch.object(hib.settings, "avito_imv_use_browser_fetcher", False),
|
||||
patch.object(hib, "evaluate_via_imv", AsyncMock(side_effect=err)),
|
||||
):
|
||||
return await hib.backfill_house_imv(db, request_delay_sec=0.0, **kwargs)
|
||||
|
||||
|
||||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||||
async def test_stuck_transient_house_returns_to_the_queue_by_itself() -> None:
|
||||
"""Прогон с дефолтными параметрами (как у расписания) забирает застрявший дом.
|
||||
|
||||
На старом коде выборка — `WHERE imv_status = :status` с единственным
|
||||
значением 'pending', поэтому дом остаётся в 'transient_error' навсегда.
|
||||
"""
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
db = _live_session()
|
||||
assert db is not None
|
||||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||||
try:
|
||||
_seed(db)
|
||||
|
||||
# Ровно то, что вызывает расписание: batch_size из default_params, статус по умолчанию.
|
||||
result = await _run(db, batch_size=50)
|
||||
|
||||
picked = {
|
||||
r.id: r.imv_status
|
||||
for r in db.execute(
|
||||
_t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
|
||||
{"ids": list(_IDS)},
|
||||
)
|
||||
}
|
||||
assert picked[_H_STUCK] != "transient_error", (
|
||||
"дом с временной ошибкой и непотраченным лимитом обязан попасть в пакет "
|
||||
f"обычного прогона; фактически статус {picked[_H_STUCK]!r} "
|
||||
f"(в пакет прогона попало домов: {result.checked})"
|
||||
)
|
||||
# pending не вытеснен повтором.
|
||||
assert picked[_H_PENDING] != "pending", "pending-дом обязан остаться в том же пакете"
|
||||
# Исчерпавший лимит в автоповтор не берётся — иначе вечный цикл.
|
||||
assert picked[_H_EXHAUSTED] == "transient_error"
|
||||
assert result.retried == 1
|
||||
# Дом без объявлений с rooms+area помечен ДО пакета, слот не потрачен.
|
||||
assert picked[_H_NO_PARAMS] == "no_params"
|
||||
assert result.premarked == 1
|
||||
finally:
|
||||
_cleanup(db)
|
||||
|
||||
|
||||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||||
async def test_explicit_only_status_still_takes_exhausted_houses() -> None:
|
||||
"""Явный only_status — ручной запрос оператора: лимит попыток его не режет."""
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
db = _live_session()
|
||||
assert db is not None
|
||||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||||
try:
|
||||
_seed(db)
|
||||
|
||||
result = await _run(db, batch_size=50, only_status="transient_error")
|
||||
|
||||
statuses = {
|
||||
r.id: r.imv_status
|
||||
for r in db.execute(
|
||||
_t("SELECT id, imv_status FROM houses WHERE id = ANY(CAST(:ids AS bigint[]))"),
|
||||
{"ids": list(_IDS)},
|
||||
)
|
||||
}
|
||||
assert (
|
||||
statuses[_H_EXHAUSTED] != "transient_error"
|
||||
), "явно запрошенный статус обрабатывается целиком, включая исчерпавшие лимит"
|
||||
# Автоповтора поверх явного запроса нет: pending не тронут.
|
||||
assert statuses[_H_PENDING] == "pending"
|
||||
assert result.retried == 0
|
||||
finally:
|
||||
_cleanup(db)
|
||||
|
||||
|
||||
@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
|
||||
async def test_transient_attempts_counter_only_counts_transient() -> None:
|
||||
"""Счётчик — это «сколько раз подряд упал по временной причине», и он обнуляем.
|
||||
|
||||
Без верхней границы у повтора не было бы выхода; без обнуления успехом дом,
|
||||
однажды поймавший 3 сетевых сбоя, навсегда терял бы право на повтор.
|
||||
"""
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
db = _live_session()
|
||||
assert db is not None
|
||||
if db.execute(_t("SELECT count(*) FROM houses")).scalar():
|
||||
pytest.skip("тестовая БД не пуста — фикстуры не попадут в голову очереди")
|
||||
try:
|
||||
_seed(db)
|
||||
assert hib._MAX_TRANSIENT_ATTEMPTS == 3, "литерал 'cap' в _seed рассинхронизирован"
|
||||
|
||||
def _attempts() -> int:
|
||||
return int(
|
||||
db.execute(
|
||||
_t("SELECT imv_transient_attempts FROM houses WHERE id = :i"),
|
||||
{"i": _H_STUCK},
|
||||
).scalar()
|
||||
)
|
||||
|
||||
hib._mark_status(db, _H_STUCK, "transient_error", "boom")
|
||||
assert _attempts() == 1
|
||||
hib._mark_status(db, _H_STUCK, "not_found", "адрес не найден")
|
||||
assert _attempts() == 1, "нетранзиентные исходы счётчик не двигают"
|
||||
db.execute(
|
||||
_t("UPDATE houses SET imv_status='ok', imv_transient_attempts=0 WHERE id = :i"),
|
||||
{"i": _H_STUCK},
|
||||
)
|
||||
db.commit()
|
||||
assert _attempts() == 0
|
||||
finally:
|
||||
_cleanup(db)
|
||||
|
||||
|
||||
def test_retry_queue_is_not_bound_to_only_status() -> None:
|
||||
"""Статика (бежит и без БД): retry-выборка не зависит от параметра only_status.
|
||||
|
||||
Ровно этот дефект и застрял на проде: единственный путь к 'transient_error'
|
||||
шёл через значение, которое никто не передавал.
|
||||
"""
|
||||
retry_sql = str(hib._RETRY_QUEUE_SQL.text)
|
||||
assert "imv_status = 'transient_error'" in retry_sql
|
||||
assert ":status" not in retry_sql, "retry-очередь обязана быть независимой от only_status"
|
||||
assert "imv_transient_attempts < :max_attempts" in retry_sql, "у повтора должен быть выход"
|
||||
# psycopg v3: только CAST(:x AS type), никогда :x::type.
|
||||
for sql in (retry_sql, str(hib._QUEUE_SQL.text), str(hib._PREMARK_UNUSABLE_SQL.text)):
|
||||
assert "::" not in sql
|
||||
Loading…
Add table
Reference in a new issue