fix(tradein/avito-detail): bound in-loop warm/research await'ы в wait_for (#1950 регрессия) + abort-check до cooldown
This commit is contained in:
parent
5bb998a625
commit
6c64e778bf
3 changed files with 63 additions and 21 deletions
|
|
@ -303,8 +303,8 @@ async def warm_up_session(session: AsyncSession) -> bool:
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
await session.get(_AVITO_WARM_YANDEX_URL) # органический referer-источник, best-effort
|
await session.get(_AVITO_WARM_YANDEX_URL) # органический referer-источник, best-effort
|
||||||
except Exception:
|
except Exception as exc:
|
||||||
pass
|
logger.debug("avito warm-up: yandex GET failed (non-fatal): %s", exc)
|
||||||
await asyncio.sleep(random.uniform(2.0, 3.5))
|
await asyncio.sleep(random.uniform(2.0, 3.5))
|
||||||
try:
|
try:
|
||||||
r = await session.get(
|
r = await session.get(
|
||||||
|
|
|
||||||
|
|
@ -99,6 +99,7 @@ async def run_avito_detail_backfill(
|
||||||
warm_batch = int(params.get("warm_batch", 500))
|
warm_batch = int(params.get("warm_batch", 500))
|
||||||
research_every = int(params.get("research_every", 50))
|
research_every = int(params.get("research_every", 50))
|
||||||
block_cooldown_sec = float(params.get("block_cooldown_sec", 30.0))
|
block_cooldown_sec = float(params.get("block_cooldown_sec", 30.0))
|
||||||
|
warm_timeout_s = float(params.get("warm_timeout_s", 90.0))
|
||||||
|
|
||||||
# #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один
|
# #1950: hard-timeout'ы на блокирующие await'ы внутри loop'а. Без них один
|
||||||
# зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч).
|
# зависший fetch_detail/rotate ронял весь run в zombie (run 423 завис 7.7ч).
|
||||||
|
|
@ -264,20 +265,43 @@ async def run_avito_detail_backfill(
|
||||||
# (re-GET search той же сессией, освежить куки). On-block — cooldown ниже.
|
# (re-GET search той же сессией, освежить куки). On-block — cooldown ниже.
|
||||||
if use_curl and session is not None:
|
if use_curl and session is not None:
|
||||||
if warm_batch and items_since_warm >= warm_batch:
|
if warm_batch and items_since_warm >= warm_batch:
|
||||||
# периодический полный re-warm на том же sticky exit-IP (свежий TLS+куки)
|
# периодический полный re-warm на том же sticky exit-IP (свежий TLS+куки).
|
||||||
|
# #1950: bound сетевой re-warm; строим НОВУЮ сессию до close старой — на
|
||||||
|
# timeout/fail сохраняем текущую рабочую сессию (graceful degrade).
|
||||||
try:
|
try:
|
||||||
await session.close()
|
new_session = await asyncio.wait_for(
|
||||||
|
build_warmed_session(), timeout=warm_timeout_s
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
await session.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
session = new_session
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
logger.warning(
|
||||||
session = await build_warmed_session()
|
"avito_detail_backfill: run_id=%d re-warm timed out/failed (>%.0fs) "
|
||||||
|
"-- keeping current session",
|
||||||
|
run_id,
|
||||||
|
warm_timeout_s,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
items_since_warm = 0
|
items_since_warm = 0
|
||||||
elif (
|
elif (
|
||||||
research_every
|
research_every
|
||||||
and items_since_warm > 0
|
and items_since_warm > 0
|
||||||
and items_since_warm % research_every == 0
|
and items_since_warm % research_every == 0
|
||||||
):
|
):
|
||||||
# лёгкий in-session перепоиск: тот же exit-IP, освежить куки
|
# лёгкий in-session перепоиск: тот же exit-IP, освежить куки.
|
||||||
await research_in_session(session)
|
# #1950: bound сетевой re-search.
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(research_in_session(session), timeout=warm_timeout_s)
|
||||||
|
except Exception:
|
||||||
|
logger.warning(
|
||||||
|
"avito_detail_backfill: run_id=%d in-session re-search "
|
||||||
|
"timed out/failed",
|
||||||
|
run_id,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не
|
# #1950: hard-timeout — зависший fetch (browser hang / curl-stall) не
|
||||||
|
|
@ -344,6 +368,18 @@ async def run_avito_detail_backfill(
|
||||||
consecutive_blocks,
|
consecutive_blocks,
|
||||||
e,
|
e,
|
||||||
)
|
)
|
||||||
|
# abort FIRST — на последнем блоке не тратим cooldown+research/rotate
|
||||||
|
# (consecutive_blocks уже инкрементнут выше).
|
||||||
|
if consecutive_blocks >= max_consecutive_blocks:
|
||||||
|
logger.error(
|
||||||
|
"avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
|
||||||
|
"IP rate-limited. enriched=%d attempted=%d",
|
||||||
|
run_id,
|
||||||
|
consecutive_blocks,
|
||||||
|
counters.enriched,
|
||||||
|
counters.attempted,
|
||||||
|
)
|
||||||
|
break
|
||||||
# МГТС sticky-IP: один фикс. exit-IP, per-connection ротации нет (проверено:
|
# МГТС sticky-IP: один фикс. exit-IP, per-connection ротации нет (проверено:
|
||||||
# 6/6 свежих сессий = тот же IP 109.252.125.80; ротация только вручную
|
# 6/6 свежих сессий = тот же IP 109.252.125.80; ротация только вручную
|
||||||
# кнопкой). Уйти на свежий IP софтом нельзя → блок = rate-limit текущего IP:
|
# кнопкой). Уйти на свежий IP софтом нельзя → блок = rate-limit текущего IP:
|
||||||
|
|
@ -353,7 +389,18 @@ async def run_avito_detail_backfill(
|
||||||
if use_curl:
|
if use_curl:
|
||||||
await asyncio.sleep(block_cooldown_sec)
|
await asyncio.sleep(block_cooldown_sec)
|
||||||
if session is not None:
|
if session is not None:
|
||||||
await research_in_session(session)
|
# #1950: bound сетевой re-search.
|
||||||
|
try:
|
||||||
|
await asyncio.wait_for(
|
||||||
|
research_in_session(session), timeout=warm_timeout_s
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
logger.warning(
|
||||||
|
"avito_detail_backfill: run_id=%d on-block re-search "
|
||||||
|
"timed out/failed",
|
||||||
|
run_id,
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
# #1950: bound rotate — _rotate_ip ждёт settle + до 3 changeip-попыток;
|
# #1950: bound rotate — _rotate_ip ждёт settle + до 3 changeip-попыток;
|
||||||
# на блокирующем changeip (зависшее соединение) без timeout loop виснет.
|
# на блокирующем changeip (зависшее соединение) без timeout loop виснет.
|
||||||
|
|
@ -367,16 +414,6 @@ async def run_avito_detail_backfill(
|
||||||
rotate_timeout_s,
|
rotate_timeout_s,
|
||||||
exc_info=True,
|
exc_info=True,
|
||||||
)
|
)
|
||||||
if consecutive_blocks >= max_consecutive_blocks:
|
|
||||||
logger.error(
|
|
||||||
"avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
|
|
||||||
"IP rate-limited. enriched=%d attempted=%d",
|
|
||||||
run_id,
|
|
||||||
consecutive_blocks,
|
|
||||||
counters.enriched,
|
|
||||||
counters.attempted,
|
|
||||||
)
|
|
||||||
break
|
|
||||||
|
|
||||||
except TimeoutError:
|
except TimeoutError:
|
||||||
# asyncio.wait_for → TimeoutError (py3.12: asyncio.TimeoutError — alias).
|
# asyncio.wait_for → TimeoutError (py3.12: asyncio.TimeoutError — alias).
|
||||||
|
|
|
||||||
|
|
@ -140,7 +140,11 @@ async def test_backfill_processes_snapshot_to_completion() -> None:
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_backfill_blocked_abort_after_max_consecutive() -> None:
|
async def test_backfill_blocked_abort_after_max_consecutive() -> None:
|
||||||
"""5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed), rotate_ip x5."""
|
"""5 consecutive AvitoBlockedError -> abort, mark_done (NOT mark_failed).
|
||||||
|
|
||||||
|
#1950 abort-reorder: abort-check ПЕРЕД recovery → на 5-м (аборт-)блоке rotate_ip
|
||||||
|
НЕ дёргается (не тратим recovery на финальном блоке). rotate_ip x4 (блоки 1-4).
|
||||||
|
"""
|
||||||
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
from app.services.scrapers.avito_exceptions import AvitoBlockedError
|
||||||
|
|
||||||
snapshot = _make_snapshot(10)
|
snapshot = _make_snapshot(10)
|
||||||
|
|
@ -170,7 +174,8 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None:
|
||||||
assert result.enriched == 0
|
assert result.enriched == 0
|
||||||
runs.mark_done.assert_called_once()
|
runs.mark_done.assert_called_once()
|
||||||
runs.mark_failed.assert_not_called()
|
runs.mark_failed.assert_not_called()
|
||||||
assert mock_scraper.return_value._rotate_ip.call_count == 5
|
# abort-check до recovery → 5-й блок абортит без rotate; rotate только на блоках 1-4.
|
||||||
|
assert mock_scraper.return_value._rotate_ip.call_count == 4
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue