perf(lps): 폴백 크롤 병렬화 + 오픈마켓 대역폭 절감

한 검색의 소요·DECODO 비용을 크게 낮춘다(실측: 110s→52s, proxy 8MB→2MB, 총비용 70%↓).

- 폴백 병렬화: _enrich_with_fallback 의 순차 for-loop → gather 동시 크롤. 각 몰 어댑터는
  별 브라우저 인스턴스라 병렬 안전. 소요=합→최댓값. AI usage 계측 동시성 불변식 주석 명시
  (judge.last_usage 세팅~읽기 사이 await 없음 → asyncio 협조 스케줄링상 안전).
- 대역폭: blocked_resource_types 를 어댑터별 설정화. 오픈마켓은 이미지/미디어/폰트만 차단
  (CSS/JS 유지 — 챌린지/렌더 보호). 격리 실험으로 차단 유무가 ESM 챌린지 성패에 무관함 확인
  (실패 원인은 IP평판/headless). st11 라이브 0.11MB 검증.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

lps/services/search/browser_base.py

@@ -23,7 +23,8 @@ from common.logger import LOG
 from services.search.contract import SearchAdapter, NormalizedProduct, AdapterError, AdapterHealth
 from services.search.rate_limiter import RateLimiter
 
-# 대역폭 절감: 이미지/미디어/폰트/CSS 는 상품 데이터·안티봇(JS) 에 불필요 → 차단(프록시 per-GB 비용↓).
+# 대역폭 절감 기본 차단 집합(쿠팡): 이미지/미디어/폰트/CSS. 쿠팡은 CSS 없이도 파싱·Akamai 통과 OK.
+# 오픈마켓(ESM/11번가)은 CSS/JS 를 막으면 렌더/챌린지가 깨져 이미지·미디어·폰트만 막는다(어댑터에서 override).
 _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"}
 
 
@@ -60,7 +61,8 @@ class BrowserSearchAdapter(SearchAdapter):
     min_result_html: int = 10000
     ready_selector: str = "body"
     ready_timeout_ms: int = 20000
-    block_resources_default: bool = True        # 쿠팡=True(대역폭↓). 오픈마켓은 리소스차단이 렌더/챌린지를 깨 False.
+    block_resources_default: bool = True        # 리소스 차단 라우팅 on/off 기본값.
+    blocked_resource_types: set = _BLOCKED_RESOURCES  # 차단할 resource_type(사이트별 override — 오픈마켓은 CSS/JS 유지)
     scroll_steps: int = 0                        # >0 이면 렌더 대기 전 스크롤(지연 로딩 트리거, 예: 11번가)
     max_proxy_retries: int = 2                   # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수
 
@@ -110,7 +112,7 @@ class BrowserSearchAdapter(SearchAdapter):
 
     # ---- 공통 브라우저 수명 -------------------------------------------
     async def _route(self, route):
-        if route.request.resource_type in _BLOCKED_RESOURCES:
+        if route.request.resource_type in self.blocked_resource_types:
             await route.abort()
         else:
             await route.continue_()

lps/services/search/esm/adapter.py

@@ -21,7 +21,8 @@ class EsmAdapter(BrowserSearchAdapter):
     block_markers = _BLOCK_MARKERS
     min_result_html = 8000              # 챌린지 페이지(~21KB)는 크므로 short 폴백은 명백한 에러만
     ready_timeout_ms = 22000            # '잠시만' 챌린지 통과(~9초) 여유
-    block_resources_default = False     # 리소스 차단은 ESM 챌린지/렌더를 깨뜨림
+    block_resources_default = True      # 라우팅 on — 단, 아래 집합만(CSS/JS 유지)
+    blocked_resource_types = {"image", "media", "font"}  # 대역폭 대부분(이미지)만 차단, 챌린지/렌더용 CSS/JS는 유지
 
     def __init__(self, site: str, **kwargs):
         if site not in SITES:

lps/services/search/st11/adapter.py

@@ -20,7 +20,8 @@ class ElevenStAdapter(BrowserSearchAdapter):
     min_result_html = 20000            # PC 정상 결과는 수백 KB — 차단/미렌더 페이지는 작다(→회전/재시도)
     ready_selector = READY_SELECTOR
     ready_timeout_ms = 20000
-    block_resources_default = False    # 리소스 차단 시 goto 가 hang / 렌더 실패
+    block_resources_default = True     # 라우팅 on — 이미지/미디어/폰트만(CSS/JS 유지해야 렌더됨)
+    blocked_resource_types = {"image", "media", "font"}
     scroll_steps = 3                   # 결과가 지연 로딩 → 스크롤로 트리거
 
     def _search_url(self, query: str, limit: int) -> str:

lps/worker/handlers.py

@@ -102,33 +102,41 @@ def build_search_handler(
         return products, per_source, tech_failed
 
     async def _match(target: dict, products: list, base_price, metrics: SearchMetrics):
-        """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. AI 토큰은 metrics 에 누적."""
+        """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. AI 토큰은 metrics 에 누적.
+        ⚠️ 동시성 불변식: judge.judge 가 last_usage 를 세팅한 뒤 여기서 읽기까지 await 이 없어야
+        한다(asyncio 협조 스케줄링상 그 사이 다른 코루틴이 last_usage 를 덮어쓸 수 없음). 병렬 폴백 안전."""
         candidates, _ = apply_filters(products, base_price=base_price)
         if judge is not None and candidates:
             verdicts = await judge.judge(target, candidates)
-            metrics.add_ai(judge.last_usage)
+            metrics.add_ai(judge.last_usage)   # ← await 직후 즉시 읽음(사이에 await 금지)
             candidates = [c for c, v in zip(candidates, verdicts) if v.is_match]
         return candidates
 
     async def _enrich_with_fallback(target: dict, query: str, matched: list, base_price, metrics: SearchMetrics):
         """네이버가 커버 못 한 오픈마켓만 직접 크롤(폴백) → 같은상품 판정 후 병합.
-        사용자 규칙: '네이버로 그 몰 값 확보 성공 → 그 값, 실패(몰 없음) → 실사이트 크롤'."""
+        사용자 규칙: '네이버로 그 몰 값 확보 성공 → 그 값, 실패(몰 없음) → 실사이트 크롤'.
+        미커버 몰들을 **동시 크롤**한다(각 어댑터=별 브라우저 인스턴스라 병렬 안전, 소요=합→최댓값)."""
         if not fallbacks:
             return matched
         covered = {canonical_mall(p) for p in matched}
-        for src, adapter in fallbacks.items():
+        todo = [(src, ad) for src, ad in fallbacks.items() if MALL_BY_SOURCE.get(src, src) not in covered]
-            mall = MALL_BY_SOURCE.get(src, src)
+        if not todo:
-            if mall in covered:  # 네이버가 이미 그 몰 최저가 확보 → 크롤 생략
+            return matched
-                continue
+
+        async def _crawl_match(src, adapter):
             try:
                 crawled = await _timed_search(adapter, query, src, metrics, crawl=True)
             except Exception as ex:
                 LOG.w(f"[fallback:{src}] 크롤 실패(무시): {type(ex).__name__}: {ex}")
-                continue
+                return []
             hits = await _match(target, crawled, base_price, metrics)
             if hits:
                 LOG.d(f"[fallback:{src}] 크롤 {len(crawled)}건 중 같은상품 {len(hits)}건 병합")
-                matched = matched + hits
+            return hits
+
+        results = await asyncio.gather(*[_crawl_match(s, a) for s, a in todo])
+        for hits in results:
+            matched = matched + hits
         return matched
 
     async def _round_queries(base_query: str, target: dict, metrics: SearchMetrics):