diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py index d4f52f7..63f782a 100644 --- a/lps/tests/test_search_handler.py +++ b/lps/tests/test_search_handler.py @@ -6,7 +6,7 @@ import pytest from common.enums import JobType from services.search.contract import NormalizedProduct, AdapterError -from worker.handlers import build_search_handler +from worker.handlers import build_search_handler, _abandoned_fallbacks class FakeAdapter: @@ -169,6 +169,10 @@ async def test_fallback_deadline_skips_slow_mall(): assert "11번가" in malls # 빠른 폴백 병합됨 assert "G마켓" not in malls # 느린 폴백은 데드라인 초과로 스킵 assert r["lowest"]["price"] == 3000 # G마켓 1000은 스킵됐으므로 최저가 아님 + # 버려진 크롤은 cancel 없이 백그라운드 종료된다 — 루프 닫기 전에 배수(pending 태스크 파괴 경고 방지) + assert _abandoned_fallbacks # 느린 폴백이 버려짐 + await asyncio.gather(*_abandoned_fallbacks, return_exceptions=True) + assert not _abandoned_fallbacks # 종료 콜백이 집합에서 제거함 async def test_fallback_failure_is_isolated(): diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index 9b0ba8b..21be864 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -25,6 +25,21 @@ from services.search.util import parse_price from services.pipeline.core import apply_filters, rank_result, summarize_by_mall +# 데드라인 초과로 버린 폴백 태스크의 강한 참조(asyncio 는 태스크를 약참조만 유지 — 없으면 GC 로 중도 파괴될 수 있음). +# 완료 시 콜백에서 스스로 제거된다. 테스트는 이 집합을 gather 해 잔여 태스크를 배수(drain)할 수 있다. +_abandoned_fallbacks: set[asyncio.Task] = set() + + +def _reap_abandoned(task: asyncio.Task): + """버려진 폴백 태스크 종료 시 예외를 회수 — 'Future exception was never retrieved' 노이즈 방지.""" + _abandoned_fallbacks.discard(task) + if task.cancelled(): + return + ex = task.exception() + if ex is not None: + LOG.d(f"[fallback] 데드라인 초과 태스크 종료(무시): {type(ex).__name__}") + + def _price_snapshot(matched: list[NormalizedProduct]) -> dict: """매칭 목록에서 소스별 최저가 + 전체 최저가 스냅샷을 만든다(price_history 기록용).""" def lowest(src): @@ -126,14 +141,18 @@ def build_search_handler( async def _crawl_match(src, adapter): # 폴백은 '있으면 좋은' 보강이라 데드라인을 건다 — 초과 시 그 몰만 스킵(전체 지연에 상한). - try: - crawled = await asyncio.wait_for( - _timed_search(adapter, query, src, metrics, crawl=True), - timeout=fallback_deadline_sec, - ) - except asyncio.TimeoutError: - LOG.w(f"[fallback:{src}] 데드라인 {fallback_deadline_sec:.0f}s 초과 → 스킵") + # cancel 하지 않고 버린다(asyncio.wait): in-flight page.goto 를 취소하면 patchright 내부 + # future 가 미회수 예외 노이즈를 남기고 페이지가 어중간한 상태로 남는다. 버려진 크롤은 + # 백그라운드에서 자체 타임아웃(goto 40s 등)으로 끝나고 _reap_abandoned 가 예외를 회수한다. + task = asyncio.ensure_future(_timed_search(adapter, query, src, metrics, crawl=True)) + done, _ = await asyncio.wait({task}, timeout=fallback_deadline_sec) + if not done: + _abandoned_fallbacks.add(task) + task.add_done_callback(_reap_abandoned) + LOG.w(f"[fallback:{src}] 데드라인 {fallback_deadline_sec:.0f}s 초과 → 스킵(크롤은 백그라운드 종료)") return [] + try: + crawled = task.result() except Exception as ex: LOG.w(f"[fallback:{src}] 크롤 실패(무시): {type(ex).__name__}: {ex}") return [] diff --git a/lps/worker_main.py b/lps/worker_main.py index eb09852..81f5dfd 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -142,8 +142,11 @@ async def run_browser_reaper(adapters, stop, idle_sec: float = 120.0, interval: except asyncio.TimeoutError: pass for ad in adapters: + close_if_idle = getattr(ad, "close_if_idle", None) + if close_if_idle is None: # 네이버(httpx) 등 브라우저 없는 어댑터는 정리 대상 아님 + continue try: - await ad.close_if_idle(idle_sec) + await close_if_idle(idle_sec) except Exception as ex: LOG.e_no_callstack(f"[browser-reaper] 정리 실패(무시): {ex}")