" + "x" * 20000, 0) == "sec-if-cpt-container"
+
+
+def test_detect_block_edge_access_denied():
+ # 실제로 잡힌 303B edge deny
+ html = ('
Access DeniedAccess Denied
'
+ "You don't have permission to access ... errors.edgesuite.net")
+ assert detect_block(html, 0) in ("errors.edgesuite.net", "You don't have permission to access")
+
+
+def test_detect_block_permission_restricted_page():
+ html = "쿠팡을 찾아주신 고객님, 입력하신 페이지주소는 사용권한이 제한된 페이지입니다."
+ assert detect_block(html, 0) in ("사용권한이 제한된", "쿠팡을 찾아주신 고객님")
+
+
+def test_detect_block_short_html_fallback():
+ # 마커 없어도 비정상적으로 짧은 0건 응답은 미지의 차단으로 폴백
+ marker = detect_block("oops", 0)
+ assert marker is not None and marker.startswith("short_html(")
+
+
+def test_detect_block_genuine_empty_large_page_not_blocked():
+ # 정상 '검색결과 없음'(전체 chrome 포함, 큰 HTML)은 차단 아님 → not_found 로 흘러야 함
+ big = "검색결과가 없습니다" + "x" * 20000 + ""
+ assert detect_block(big, 0) is None
diff --git a/lps/tests/test_health.py b/lps/tests/test_health.py
new file mode 100644
index 0000000..2446098
--- /dev/null
+++ b/lps/tests/test_health.py
@@ -0,0 +1,9 @@
+# 프레임워크 골격 스모크 테스트. 도메인 로직이 없어도 앱이 부팅되고 healthz 가 응답하는지 확인한다.
+# (DB 없이 통과 — 엔진은 lazy 라 실제 커넥션을 맺지 않는다.)
+
+
+async def test_healthz_ok(client):
+ res = await client.get("/healthz")
+ assert res.status_code == 200
+ # 기동 시각 문자열(예: "2026-07-08 04:26:58")을 그대로 반환한다.
+ assert isinstance(res.json(), str)
diff --git a/lps/tests/test_job_queue.py b/lps/tests/test_job_queue.py
new file mode 100644
index 0000000..488bc22
--- /dev/null
+++ b/lps/tests/test_job_queue.py
@@ -0,0 +1,93 @@
+"""작업 큐 엔진 테스트 — 원자적 claim(이중할당 불가)·lease 회수·재시도/dead-letter·소유권 가드.
+실제 lps_db 에 붙어 검증한다(db_engine 이 스키마 보장)."""
+
+import asyncio
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import JobStatus, JobType
+from crud.job_crud import JobQueue, compute_backoff
+
+
+@pytest_asyncio.fixture
+async def q(db_engine):
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE job"))
+ return JobQueue()
+
+
+async def test_enqueue_claim_complete(q):
+ jid = await q.enqueue(JobType.SEARCH.value, {"query": "커피"}, dedupe_key="search-커피")
+ assert jid
+ job = await q.claim("w1")
+ assert job and job["job_id"] == jid
+ assert job["payload"]["query"] == "커피" and job["attempts"] == 1
+ assert await q.complete(jid, "w1", {"count": 3}) is True
+ counts = await q.counts()
+ assert counts["DONE"] == 1 and counts["PENDING"] == 0
+
+
+async def test_dedupe_blocks_active_duplicate(q):
+ a = await q.enqueue(JobType.SEARCH.value, {"q": 1}, dedupe_key="k")
+ b = await q.enqueue(JobType.SEARCH.value, {"q": 1}, dedupe_key="k")
+ assert a and b is None # 활성 중복 차단
+ # 완료로 빠지면 같은 키 재적재 가능
+ job = await q.claim("w1")
+ await q.complete(job["job_id"], "w1")
+ c = await q.enqueue(JobType.SEARCH.value, {"q": 1}, dedupe_key="k")
+ assert c
+
+
+async def test_atomic_claim_no_double_assignment(q):
+ N = 12
+ for i in range(N):
+ await q.enqueue(JobType.SEARCH.value, {"i": i})
+ # 8개 워커가 동시에 claim → 서로 다른 잡만, 이중 할당 0
+ results = await asyncio.gather(*[q.claim(f"w{i}") for i in range(8)])
+ claimed = [r["job_id"] for r in results if r]
+ assert len(claimed) == 8
+ assert len(set(claimed)) == 8
+
+
+async def test_priority_and_order(q):
+ await q.enqueue(JobType.SEARCH.value, {"n": "low"}, priority=100)
+ await q.enqueue(JobType.SEARCH.value, {"n": "high"}, priority=1)
+ job = await q.claim("w1")
+ assert job["payload"]["n"] == "high" # priority 낮은 값 우선
+
+
+async def test_lease_reclaim_by_reaper(q):
+ jid = await q.enqueue(JobType.SEARCH.value, {"q": "x"})
+ job = await q.claim("w1", lease_sec=1)
+ assert job["job_id"] == jid and job["attempts"] == 1
+ assert await q.reap() == [] # 아직 lease 유효 → 회수 없음
+ await asyncio.sleep(1.3) # lease 만료
+ assert jid in await q.reap() # 회수됨(워커 사망 시나리오)
+ job2 = await q.claim("w2") # 다시 claim 가능, attempts 누적
+ assert job2["job_id"] == jid and job2["attempts"] == 2
+
+
+async def test_retry_then_dead_letter(q):
+ jid = await q.enqueue(JobType.SEARCH.value, {"q": "x"}, max_attempts=2)
+ await q.claim("w1")
+ assert await q.fail(jid, "w1", "boom", backoff_sec=0) == JobStatus.PENDING.value # 1/2 → 재큐
+ job2 = await q.claim("w1")
+ assert job2["attempts"] == 2
+ assert await q.fail(jid, "w1", "boom2", backoff_sec=0) == JobStatus.DEAD.value # 2/2 → dead-letter
+ assert (await q.counts())["DEAD"] == 1
+
+
+async def test_transitions_require_ownership(q):
+ jid = await q.enqueue(JobType.SEARCH.value, {"q": "x"})
+ await q.claim("w1")
+ assert await q.complete(jid, "intruder") is False # 소유 아님 → 거부(CAS 가드)
+ assert await q.fail(jid, "intruder", "no") is None
+ assert await q.complete(jid, "w1") is True
+
+
+def test_backoff_is_exponential_capped():
+ assert compute_backoff(1, base=5) == 5
+ assert compute_backoff(2, base=5) == 10
+ assert compute_backoff(3, base=5) == 20
+ assert compute_backoff(100, base=5, cap=600) == 600
diff --git a/lps/tests/test_lps_api.py b/lps/tests/test_lps_api.py
new file mode 100644
index 0000000..3e659a3
--- /dev/null
+++ b/lps/tests/test_lps_api.py
@@ -0,0 +1,94 @@
+"""LPS API 라우터 테스트 — 검색 적재/중복/상태조회/큐 통계 (ASGI 클라이언트 + 실 lps_db)."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+
+@pytest_asyncio.fixture
+async def clean_jobs(db_engine):
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE job"))
+
+
+async def test_search_enqueues_jobs(client, clean_jobs):
+ body = {"data": [
+ {"product_code": "P1", "product_name": "커피", "job_type": "new_product"},
+ {"product_code": "P2", "product_name": "무선마우스", "specification": "M170"},
+ ]}
+ r = await client.post("/v1/lps/search", json=body)
+ assert r.status_code == 200
+ j = r.json()
+ assert j["accepted"] == 2
+ assert {i["product_code"] for i in j["items"]} == {"P1", "P2"}
+ assert all(i.get("job_id") for i in j["items"])
+
+
+async def test_search_dedupes_active_product(client, clean_jobs):
+ body = {"data": [{"product_code": "P1", "product_name": "커피", "job_type": "new_product"}]}
+ await client.post("/v1/lps/search", json=body)
+ r2 = await client.post("/v1/lps/search", json=body) # 같은 product_code 재요청
+ item = r2.json()["items"][0]
+ assert item["duplicated"] is True
+ assert "job_id" not in item # None → RemoveNoneResponse 로 제거됨
+ assert r2.json()["accepted"] == 0
+
+
+async def test_job_status_flow(client, clean_jobs):
+ jid = (await client.post("/v1/lps/search", json={"data": [{"product_code": "P9", "product_name": "커피"}]})).json()["items"][0]["job_id"]
+ r = await client.get(f"/v1/lps/jobs/{jid}")
+ body = r.json()
+ assert body["status"] == "PENDING" and body["attempts"] == 0
+ assert body["result"]["success"] is True
+
+
+async def test_job_status_not_found(client, clean_jobs):
+ # 존재하지 않는(유효 UUID) 잡
+ r = await client.get("/v1/lps/jobs/00000000-0000-0000-0000-000000000000")
+ assert r.json()["result"]["success"] is False
+ assert r.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND"
+ # 잘못된 형식의 id 도 not-found 처리
+ r2 = await client.get("/v1/lps/jobs/not-a-uuid")
+ assert r2.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND"
+
+
+async def test_price_history_endpoint(client, db_engine):
+ from crud.price_history import PriceHistory
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE price_history"))
+ ph = PriceHistory()
+ await ph.record({"product_code": "GRAPH1", "outcome": "found", "final_lowest": 2500,
+ "final_source": "naver", "naver_lowest": 2500, "coupang_lowest": 2700, "matched_count": 2})
+
+ r = await client.get("/v1/lps/products/GRAPH1/history")
+ assert r.status_code == 200
+ body = r.json()
+ assert body["product_code"] == "GRAPH1"
+ assert len(body["points"]) == 1
+ pt = body["points"][0]
+ assert pt["final"] == 2500 and pt["naver"] == 2500 and pt["coupang"] == 2700 and pt["final_source"] == "naver"
+ assert "triggered_at" in pt
+
+
+async def test_queue_stats(client, clean_jobs):
+ await client.post("/v1/lps/search", json={"data": [
+ {"product_code": "A", "product_name": "x"},
+ {"product_code": "B", "product_name": "y"},
+ ]})
+ r = await client.get("/v1/lps/queue/stats")
+ counts = r.json()["counts"]
+ assert counts["PENDING"] == 2 and counts["DONE"] == 0 and counts["DEAD"] == 0
+
+
+async def test_readyz(client):
+ r = await client.get("/readyz") # DB 도달 → ready
+ assert r.status_code == 200 and r.json()["ready"] is True
+
+
+async def test_ops_snapshot(client, clean_jobs):
+ await client.post("/v1/lps/search", json={"data": [{"product_code": "A", "product_name": "x"}]})
+ r = await client.get("/v1/lps/ops")
+ assert r.status_code == 200
+ j = r.json()
+ for k in ("pending", "running", "done", "dead", "dead_1h", "stuck_running", "oldest_pending_sec", "blocks_1h"):
+ assert k in j and isinstance(j[k], int)
+ assert j["pending"] == 1
diff --git a/lps/tests/test_naver_transform.py b/lps/tests/test_naver_transform.py
new file mode 100644
index 0000000..5d49322
--- /dev/null
+++ b/lps/tests/test_naver_transform.py
@@ -0,0 +1,19 @@
+"""네이버 응답 변환 테스트 (순수, 네트워크 불필요)."""
+
+from services.search.naver.transform import transform_items
+
+
+def test_transform_strips_tags_and_keeps_catalog():
+ items = [
+ {"title": "로지텍
무선 마우스 & 키보드", "link": "https://search.shopping.naver.com/catalog/1",
+ "image": "img1", "lprice": "13500", "mallName": "네이버", "maker": "로지텍", "productId": "1"},
+ {"title": "0원상품", "link": "https://x", "lprice": "0"}, # 가격 0 → 제외
+ {"title": "가격없음", "link": "https://y", "lprice": "abc"}, # 파싱 실패 → 제외
+ ]
+ out = transform_items(items)
+ assert len(out) == 1
+ p = out[0]
+ assert p.name == "로지텍 무선 마우스 & 키보드" #
제거 + 엔티티 복원
+ assert p.price == 13500 and p.source == "naver"
+ assert p.mall_name == "네이버" and p.manufacturer == "로지텍" and p.external_id == "1"
+ assert "catalog/1" in p.detail_url # 가격비교(catalog) 최저가 유지
diff --git a/lps/tests/test_negative_cache.py b/lps/tests/test_negative_cache.py
new file mode 100644
index 0000000..7a9d6e6
--- /dev/null
+++ b/lps/tests/test_negative_cache.py
@@ -0,0 +1,36 @@
+"""네거티브 캐시 CRUD 테스트 (실 lps_db)."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from crud.negative_cache import NegativeCache
+
+
+@pytest_asyncio.fixture
+async def nc(db_engine):
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE search_negative"))
+ return NegativeCache()
+
+
+async def test_put_then_is_negative(nc):
+ assert await nc.is_negative("P1") is False
+ await nc.put("P1", ttl_sec=3600)
+ assert await nc.is_negative("P1") is True
+
+
+async def test_expired_is_not_negative(nc):
+ await nc.put("P2", ttl_sec=-1) # 이미 만료
+ assert await nc.is_negative("P2") is False
+
+
+async def test_upsert_refreshes_ttl(nc):
+ await nc.put("P3", ttl_sec=-1) # 만료 상태
+ assert await nc.is_negative("P3") is False
+ await nc.put("P3", ttl_sec=3600) # 갱신 → 유효
+ assert await nc.is_negative("P3") is True
+
+
+async def test_empty_key_is_noop(nc):
+ await nc.put("", ttl_sec=3600)
+ assert await nc.is_negative("") is False
diff --git a/lps/tests/test_openmarket_parser.py b/lps/tests/test_openmarket_parser.py
new file mode 100644
index 0000000..9c4460a
--- /dev/null
+++ b/lps/tests/test_openmarket_parser.py
@@ -0,0 +1,103 @@
+# 오픈마켓(G마켓·옥션·11번가) 크롤 파서 결정론적 단위 테스트(네트워크/브라우저 불필요).
+# fixture 는 실제 렌더된 검색결과에서 카드 3개씩 추출한 것.
+from pathlib import Path
+
+import pytest
+
+import time
+
+from services.search.card_parser import parse_cards
+from services.search.browser_base import is_proxy_error, BrowserSearchAdapter
+from services.search.esm.selectors import GMARKET, AUCTION
+from services.search.st11.selectors import CARDS as ST11
+
+FIX = Path(__file__).parent / "fixtures"
+
+CASES = [
+ ("gmarket_search.html", GMARKET.cards, "gmarket", "G마켓"),
+ ("auction_search.html", AUCTION.cards, "auction", "옥션"),
+ ("st11_search.html", ST11, "st11", "11번가"),
+]
+
+
+@pytest.mark.parametrize("fixture,cfg,source,mall", CASES)
+def test_parse_extracts_valid_products(fixture, cfg, source, mall):
+ items = parse_cards((FIX / fixture).read_text(), cfg)
+ assert len(items) >= 2, f"{source}: 카드 파싱 실패"
+ for p in items:
+ assert p.source == source
+ assert p.mall_name == mall
+ assert p.name and len(p.name) > 2
+ assert "상품명" not in p.name and "브랜드명" not in p.name # a11y 라벨 제거 확인
+ assert p.price >= 100, f"이상 저가: {p.price} ({p.name})"
+
+
+@pytest.mark.parametrize("fixture,cfg,source,mall", CASES)
+def test_shipping_type_valid(fixture, cfg, source, mall):
+ items = parse_cards((FIX / fixture).read_text(), cfg)
+ for p in items:
+ assert p.shipping_type in (None, "free", "paid")
+ if p.shipping_type == "paid":
+ assert p.shipping_fee and p.shipping_fee > 0
+ if p.shipping_type == "free":
+ assert p.shipping_fee == 0
+
+
+@pytest.mark.parametrize("msg", [
+ "Page.goto: net::ERR_TUNNEL_CONNECTION_FAILED at https://...",
+ "net::ERR_HTTP_RESPONSE_CODE_FAILURE at https://www.coupang.com/...",
+ "HTTP ERROR 407 Proxy Authentication Required",
+ "net::ERR_PROXY_CONNECTION_FAILED",
+])
+def test_is_proxy_error_true(msg):
+ # 프록시 전송 실패(포트/IP 사망·407) → IP 회전 대상
+ assert is_proxy_error(msg) is True
+
+
+@pytest.mark.parametrize("msg", [
+ "쿠팡 결과 없음/차단 (blocked=True)",
+ "net::ERR_NAME_NOT_RESOLVED", # DNS — 프록시 문제 아님
+ "Timeout 40000ms exceeded", # 단순 타임아웃(사이트 지연)
+ "",
+])
+def test_is_proxy_error_false(msg):
+ assert is_proxy_error(msg) is False
+
+
+class _IdleAdapter(BrowserSearchAdapter):
+ source = "test"
+ def _search_url(self, q, l): return ""
+ def _parse(self, h): return []
+
+
+class _FakeCtx:
+ def __init__(self): self.closed = False
+ async def close(self): self.closed = True
+
+
+async def test_close_if_idle_keeps_recent_closes_idle():
+ ad = _IdleAdapter()
+ ad._ctx = _FakeCtx()
+ ad._last_used = time.monotonic() # 방금 사용
+ await ad.close_if_idle(60)
+ assert ad._ctx is not None # 최근 사용 → 유지
+
+ ctx = ad._ctx
+ ad._last_used = time.monotonic() - 100 # 100s 전(유휴)
+ await ad.close_if_idle(60)
+ assert ad._ctx is None and ctx.closed # 유휴 초과 → 브라우저 정리
+
+
+async def test_close_if_idle_skips_when_no_ctx():
+ ad = _IdleAdapter() # 브라우저 미기동
+ await ad.close_if_idle(0) # 예외 없이 no-op
+ assert ad._ctx is None
+
+
+def test_dedup_by_link():
+ # 동일 링크 카드가 반복돼도 1건으로 축약
+ card = ('')
+ html = f""
+ items = parse_cards(html, GMARKET.cards)
+ assert len(items) == 1 and items[0].price == 5000
diff --git a/lps/tests/test_pipeline.py b/lps/tests/test_pipeline.py
new file mode 100644
index 0000000..4db98e3
--- /dev/null
+++ b/lps/tests/test_pipeline.py
@@ -0,0 +1,88 @@
+"""코어 파이프라인 테스트 — 필터·IQR 이상치·top-N 최저가 (결정론적, 네트워크 불필요)."""
+
+from pathlib import Path
+
+from services.search.contract import NormalizedProduct
+from services.search.coupang.parser import parse_search_html
+from services.pipeline.filters import filter_by_price_band, filter_out_malls
+from services.pipeline.outliers import remove_price_outliers
+from services.pipeline.core import run_price_pipeline, summarize_by_mall
+
+FIXTURE = Path(__file__).parent / "fixtures" / "coupang_search.html"
+
+
+def _p(price, name="p", mall="쿠팡"):
+ return NormalizedProduct(source="coupang", name=name, price=price, mall_name=mall)
+
+
+def test_top_n_is_lowest_price_sorted():
+ prods = [_p(x) for x in [3000, 1000, 2000, 5000, 4000]]
+ r = run_price_pipeline(prods, remove_outliers=False, top_n=3)
+ assert [p["price"] for p in r["top"]] == [1000, 2000, 3000]
+ assert r["lowest"]["price"] == 1000
+
+
+def test_outlier_removes_extreme_low_and_high():
+ normal = [_p(x) for x in [1000, 1010, 1020, 1030, 1040, 1050, 1060, 1070, 1080, 1090]]
+ kept, removed = remove_price_outliers(normal + [_p(5), _p(500000)])
+ prices_removed = {p.price for p in removed}
+ assert 5 in prices_removed and 500000 in prices_removed
+ assert all(1000 <= p.price <= 1090 for p in kept)
+
+
+def test_price_band_filter():
+ prods = [_p(x) for x in [8000, 10000, 12000, 30000, 1000]]
+ # 기준가 10000, ±70% → [3000, 17000] 만 통과
+ out = filter_by_price_band(prods, base_price=10000, tolerance=0.7)
+ assert {p.price for p in out} == {8000, 10000, 12000}
+
+
+def test_filter_out_malls():
+ prods = [_p(1000, mall="쿠팡"), _p(2000, mall="G마켓"), _p(3000, mall="옥션")]
+ out = filter_out_malls(prods, ["G마켓", "옥션"])
+ assert {p.mall_name for p in out} == {"쿠팡"}
+
+
+def test_empty_input_is_safe():
+ r = run_price_pipeline([], top_n=5)
+ assert r["lowest"] is None and r["top"] == [] and r["total_found"] == 0
+
+
+def test_stages_recorded():
+ prods = [_p(x) for x in [1000, 2000, 3000, 4000, 5000]]
+ r = run_price_pipeline(prods, base_price=3000, top_n=2)
+ names = [s["stage"] for s in r["stages"]]
+ assert "price_band" in names and "outlier" in names and names[-1] == "top_n"
+ assert r["stages"][-1]["out"] == 2 # top_n 결과 수
+
+
+def _pm(source, mall, price):
+ return NormalizedProduct(source=source, name=f"{mall}상품", price=price, mall_name=mall)
+
+
+def test_summarize_by_mall_lowest_per_mall_sorted():
+ # 같은 몰 여러 건 → 몰별 최저가만, 전체 가격 오름차순
+ prods = [
+ _pm("naver", "G마켓", 12000), _pm("naver", "G마켓", 11000),
+ _pm("naver", "11번가", 10500), _pm("naver", "네이버", 9800),
+ _pm("coupang", "쿠팡", 10200),
+ ]
+ rows = summarize_by_mall(prods)
+ # 몰별 최저가 1건씩, 전체 가격 오름차순
+ assert [(r["mall_name"], r["price"]) for r in rows] == [
+ ("네이버", 9800), ("쿠팡", 10200), ("11번가", 10500), ("G마켓", 11000),
+ ]
+
+
+def test_summarize_by_mall_in_result():
+ r = run_price_pipeline([_pm("naver", "11번가", 5000), _pm("naver", "G마켓", 6000)], remove_outliers=False)
+ malls = {row["mall_name"] for row in r["by_mall"]}
+ assert malls == {"11번가", "G마켓"}
+
+
+def test_pipeline_on_real_fixture():
+ products = parse_search_html(FIXTURE.read_text())
+ r = run_price_pipeline(products, remove_outliers=False, top_n=3)
+ prices = [p["price"] for p in r["top"]]
+ assert prices == sorted(prices) # 최저가순
+ assert r["lowest"]["price"] == min(p.price for p in products)
diff --git a/lps/tests/test_pool_autosize.py b/lps/tests/test_pool_autosize.py
new file mode 100644
index 0000000..3d780b3
--- /dev/null
+++ b/lps/tests/test_pool_autosize.py
@@ -0,0 +1,57 @@
+"""커넥션 풀 자동 산정(_autosize_pool) — process_count 기준으로 예산을 넘지 않아야 한다."""
+
+import pytest
+
+from config.config_models import MainDBConfig
+from config.server_configs import _autosize_pool
+
+
+def _conns(cfg: MainDBConfig, pc: int) -> int:
+ # 실제 동시 커넥션 = (pool + overflow) × 2엔진(R/W) × process_count
+ return (cfg.pool_size + cfg.max_overflow) * 2 * pc
+
+
+@pytest.mark.parametrize("pc", [1, 2, 4, 8, 16])
+def test_autosize_within_budget(pc):
+ cfg = MainDBConfig(connection_budget=40)
+ _autosize_pool(cfg, pc)
+ assert _conns(cfg, pc) <= 40
+ assert cfg.pool_size >= 1
+ assert cfg.max_overflow >= 0
+
+
+def test_autosize_uses_budget_efficiently():
+ # 예산을 지나치게 낭비하지 않아야(넉넉한 예산일 때 절반 이상 활용)
+ cfg = MainDBConfig(connection_budget=96)
+ _autosize_pool(cfg, 4)
+ assert _conns(cfg, 4) <= 96
+ assert _conns(cfg, 4) >= 96 // 2
+
+
+def test_autosize_split_ratio():
+ # pool(정상) 이 overflow(버스트)보다 크거나 같게 분할
+ cfg = MainDBConfig(connection_budget=96)
+ _autosize_pool(cfg, 1)
+ assert cfg.pool_size >= cfg.max_overflow
+
+
+def test_autosize_disabled_when_budget_zero():
+ # budget<=0 이면 자동 산정 끔 → toml/기본 pool 값 유지, None 반환
+ cfg = MainDBConfig(connection_budget=0, pool_size=10, max_overflow=20)
+ assert _autosize_pool(cfg, 4) is None
+ assert (cfg.pool_size, cfg.max_overflow) == (10, 20)
+
+
+def test_autosize_returns_computed_pair():
+ cfg = MainDBConfig(connection_budget=40)
+ result = _autosize_pool(cfg, 2)
+ assert result == (cfg.pool_size, cfg.max_overflow)
+
+
+def test_autosize_infeasible_process_count_floors_at_one():
+ # process_count×2 > budget 이면 예산 준수가 물리적으로 불가능(워커당 최소 1커넥션 필요).
+ # 이때는 pool_size=1/overflow=0(엔진당 1커넥션)까지 줄이는 게 한계 — 0 풀은 만들지 않는다.
+ cfg = MainDBConfig(connection_budget=40)
+ _autosize_pool(cfg, 64)
+ assert cfg.pool_size == 1
+ assert cfg.max_overflow == 0
diff --git a/lps/tests/test_price_history.py b/lps/tests/test_price_history.py
new file mode 100644
index 0000000..00def63
--- /dev/null
+++ b/lps/tests/test_price_history.py
@@ -0,0 +1,93 @@
+"""최저가 이력 — 소스별 min 스냅샷 계산 + 기록/조회 CRUD + 핸들러 기록 규칙."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import JobType
+from crud.price_history import PriceHistory
+from services.search.contract import NormalizedProduct
+from worker.handlers import _price_snapshot, build_search_handler
+
+
+def _np(source, price, name=None):
+ return NormalizedProduct(source=source, name=name or f"{source}-{price}", price=price, detail_url=f"http://{source}/{price}")
+
+
+# ── 스냅샷 계산 (순수) ─────────────────────────────────────────────
+def test_snapshot_source_lowest_and_final():
+ matched = [_np("naver", 3000), _np("naver", 2000), _np("coupang", 2500)]
+ s = _price_snapshot(matched)
+ assert s["naver_lowest"] == 2000 and s["coupang_lowest"] == 2500
+ assert s["final_lowest"] == 2000 and s["final_source"] == "naver"
+ assert s["matched_count"] == 3
+
+
+def test_snapshot_single_source_only():
+ s = _price_snapshot([_np("coupang", 1500)])
+ assert s["naver_lowest"] is None and s["coupang_lowest"] == 1500
+ assert s["final_lowest"] == 1500 and s["final_source"] == "coupang"
+
+
+def test_snapshot_empty_is_all_null():
+ s = _price_snapshot([])
+ assert s["final_lowest"] is None and s["naver_lowest"] is None and s["matched_count"] == 0
+
+
+# ── CRUD (실 DB) ───────────────────────────────────────────────────
+@pytest_asyncio.fixture
+async def ph(db_engine):
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE price_history"))
+ return PriceHistory()
+
+
+async def test_record_and_list_time_ordered(ph):
+ await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 2000, "final_source": "naver",
+ "naver_lowest": 2000, "coupang_lowest": 2500, "matched_count": 3})
+ await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 1900, "final_source": "coupang",
+ "naver_lowest": 2100, "coupang_lowest": 1900, "matched_count": 2})
+ await ph.record({"product_code": "P2", "outcome": "found", "final_lowest": 999}) # 다른 상품
+
+ points = await ph.list_by_product("P1")
+ assert len(points) == 2 # P2 제외
+ assert [p["final_lowest"] for p in points] == [2000, 1900] # 시각 오름차순
+ assert points[0]["triggered_at"] <= points[1]["triggered_at"]
+
+
+# ── 핸들러 기록 규칙 ───────────────────────────────────────────────
+class _Rec:
+ def __init__(self): self.events = []
+ async def record(self, e): self.events.append(e)
+
+
+class _FakeAdapter:
+ def __init__(self, source, products): self.source = source; self._p = products
+ async def search(self, q, limit=40): return self._p
+
+
+class _Neg:
+ def __init__(self, neg): self._neg = neg
+ async def is_negative(self, k): return self._neg
+ async def put(self, *a, **k): pass
+
+
+def _job(**p):
+ p.setdefault("product_name", "x"); p.setdefault("product_code", "PC1")
+ return {"job_type": JobType.SEARCH.value, "attempts": 1, "job_id": "j1", "payload": p}
+
+
+async def test_handler_records_found_snapshot():
+ rec = _Rec()
+ adapters = {"naver": _FakeAdapter("naver", [_np("naver", 2000)]), "coupang": _FakeAdapter("coupang", [_np("coupang", 1800)])}
+ await build_search_handler(adapters, history=rec)(_job())
+ assert len(rec.events) == 1
+ e = rec.events[0]
+ assert e["product_code"] == "PC1" and e["outcome"] == "found"
+ assert e["final_lowest"] == 1800 and e["final_source"] == "coupang"
+
+
+async def test_handler_skips_record_on_negative_cache_hit():
+ rec = _Rec()
+ adapters = {"naver": _FakeAdapter("naver", [_np("naver", 100)])}
+ await build_search_handler(adapters, neg_cache=_Neg(True), history=rec)(_job())
+ assert rec.events == [] # 캐시 히트 → 새 관측 없음 → 미기록
diff --git a/lps/tests/test_proxy.py b/lps/tests/test_proxy.py
new file mode 100644
index 0000000..49cf61e
--- /dev/null
+++ b/lps/tests/test_proxy.py
@@ -0,0 +1,52 @@
+"""DECODO 프록시 제공자 테스트 (포트 기반 sticky, 순수·네트워크 불필요)."""
+
+from config.config_models import DecodoConfig
+from services.search.proxy import DecodoProxy
+
+
+def _p(**kw):
+ base = dict(host="gate.decodo.com", username="user1", password="pw", port_start=10001, port_end=10010, session_minutes=10)
+ base.update(kw)
+ return DecodoProxy(DecodoConfig(**base))
+
+
+def test_disabled_when_credentials_missing():
+ assert DecodoProxy(DecodoConfig()).enabled is False # 전부 비어있음
+ assert _p(password="").enabled is False
+ assert _p(port_end=0).enabled is False # 포트 미설정
+ assert _p().enabled is True
+
+
+def test_playwright_proxy_shape():
+ cfg = _p().playwright_proxy()
+ assert cfg["username"] == "user1" and cfg["password"] == "pw" # 고정 자격증명
+ assert cfg["server"].startswith("http://gate.decodo.com:")
+ port = int(cfg["server"].rsplit(":", 1)[1])
+ assert 10001 <= port <= 10010 # 포트 범위 안
+
+
+def test_disabled_returns_none():
+ assert DecodoProxy(DecodoConfig()).playwright_proxy() is None
+
+
+def test_port_selected_within_range_and_stable_in_window():
+ p = _p()
+ port = p._port()
+ assert 10001 <= port <= 10010
+ assert p._port() == port # 같은 시간창에서는 동일 포트(동일 sticky IP)
+
+
+def test_single_port_range():
+ p = _p(port_start=10001, port_end=10001)
+ assert p._port() == 10001 # 포트 1개면 항상 그 포트
+
+
+def test_rotate_advances_port_immediately():
+ p = _p(port_start=10001, port_end=10003) # 포트 3개
+ before = p._port()
+ p.rotate()
+ after = p._port()
+ assert after != before # 즉시 다음 포트(새 IP)
+ assert 10001 <= after <= 10003
+ p.rotate(); p.rotate() # 3번 회전하면 한 바퀴 → 원위치
+ assert p._port() == before
diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py
new file mode 100644
index 0000000..63f782a
--- /dev/null
+++ b/lps/tests/test_search_handler.py
@@ -0,0 +1,241 @@
+"""검색 핸들러 테스트 — 병합·실패격리·AI판정·재정제 루프·not_found·네거티브 캐시 (fake 의존성)."""
+
+import asyncio
+
+import pytest
+
+from common.enums import JobType
+from services.search.contract import NormalizedProduct, AdapterError
+from worker.handlers import build_search_handler, _abandoned_fallbacks
+
+
+class FakeAdapter:
+ def __init__(self, source, by_query=None, products=None, fail=False, uses_proxy=False, last_bytes=0, delay=0.0):
+ self.source = source
+ self._by_query = by_query # {query: [products]}
+ self._products = products or []
+ self._fail = fail
+ self._delay = delay # search 지연(초) — 데드라인 테스트용
+ self.uses_proxy = uses_proxy # DECODO 경유 여부(비용 귀속)
+ self.last_bytes = last_bytes
+ self.calls = []
+
+ async def search(self, query, limit=40):
+ self.calls.append(query)
+ if self._delay:
+ await asyncio.sleep(self._delay)
+ if self._fail:
+ raise AdapterError("boom", source=self.source, blocked=True)
+ if self._by_query is not None:
+ return self._by_query.get(query, [])
+ return self._products
+
+
+class _Usage:
+ def __init__(self, prompt, completion):
+ self.prompt_tokens, self.completion_tokens = prompt, completion
+
+
+class FakeJudge:
+ def __init__(self, predicate, usage=None):
+ self._pred = predicate
+ self.last_usage = usage # 계측용(핸들러가 judge 후 읽음)
+
+ async def judge(self, target, candidates):
+ from services.ai.similarity import Judgment
+ return [Judgment(index=i + 1, is_match=self._pred(c), score=100 if self._pred(c) else 0)
+ for i, c in enumerate(candidates)]
+
+
+class FakeKeywordGen:
+ def __init__(self, precise="", broad=""):
+ self._p, self._b = precise, broad
+ self.last_usage = None
+
+ async def generate(self, target):
+ from services.ai.keyword import Keywords
+ return Keywords(precise=self._p, broad=self._b)
+
+
+class FakeNegCache:
+ def __init__(self, negative=False):
+ self._neg = negative
+ self.puts = []
+
+ async def is_negative(self, key):
+ return self._neg
+
+ async def put(self, key, ttl_sec=86400, reason="x"):
+ self.puts.append(key)
+
+
+def _np(source, price, mall=None):
+ return NormalizedProduct(source=source, name=f"{source}-{price}", price=price, mall_name=mall)
+
+
+def _job(**payload):
+ payload.setdefault("product_name", "x")
+ return {"job_type": JobType.SEARCH.value, "attempts": 1, "payload": payload}
+
+
+# ── 병합 / 실패격리 / AI 판정 (round 0) ─────────────────────────────
+async def test_merges_and_ranks_across_sources():
+ adapters = {
+ "coupang": FakeAdapter("coupang", products=[_np("coupang", 3000), _np("coupang", 1000)]),
+ "naver": FakeAdapter("naver", products=[_np("naver", 2000), _np("naver", 500)]),
+ }
+ r = await build_search_handler(adapters, top_n=3)(_job())
+ assert r["outcome"] == "found" and r["lowest"]["price"] == 500
+ assert [p["price"] for p in r["top"]] == [500, 1000, 2000]
+
+
+async def test_isolates_single_source_failure_but_still_found():
+ adapters = {"coupang": FakeAdapter("coupang", fail=True), "naver": FakeAdapter("naver", products=[_np("naver", 900)])}
+ r = await build_search_handler(adapters)(_job())
+ assert r["outcome"] == "found" and r["lowest"]["price"] == 900
+ assert "error" in r["sources"]["coupang"]
+
+
+async def test_ai_judge_filters_non_matches():
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 1000), _np("naver", 2000), _np("naver", 3000)])}
+ r = await build_search_handler(adapters, judge=FakeJudge(lambda c: c.price == 2000))(_job())
+ assert [p["price"] for p in r["top"]] == [2000]
+ assert next(s for s in r["stages"] if s["stage"] == "ai_match")["out"] == 1
+
+
+# ── 재정제 루프 ────────────────────────────────────────────────────
+async def test_refines_to_precise_query_when_original_empty():
+ adapters = {"naver": FakeAdapter("naver", by_query={"스탠리 퀜처 887ml": [_np("naver", 40000)]})} # 원본은 0건
+ kw = FakeKeywordGen(precise="스탠리 퀜처 887ml", broad="스탠리 텀블러")
+ r = await build_search_handler(adapters, keyword_gen=kw)(_job(product_name="스탠리 텀블러"))
+ assert r["outcome"] == "found" and r["round"] == "precise" and r["rounds_tried"] == 2
+ assert r["lowest"]["price"] == 40000
+
+
+async def test_not_found_after_all_rounds_and_caches():
+ adapters = {"naver": FakeAdapter("naver", by_query={})} # 어떤 쿼리든 0건
+ kw = FakeKeywordGen(precise="P", broad="B")
+ neg = FakeNegCache()
+ r = await build_search_handler(adapters, keyword_gen=kw, neg_cache=neg)(_job(product_code="PC1", product_name="없는상품"))
+ assert r["outcome"] == "not_found" and r["rounds_tried"] == 3
+ assert r["lowest"] is None and r["top"] == []
+ assert neg.puts == ["PC1"] # 네거티브 캐시에 기록
+
+
+async def test_negative_cache_short_circuits():
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 100)])}
+ neg = FakeNegCache(negative=True)
+ r = await build_search_handler(adapters, neg_cache=neg)(_job(product_code="PC9"))
+ assert r["outcome"] == "not_found" and r["cached"] is True
+ assert adapters["naver"].calls == [] # 재검색 안 함
+
+
+async def test_technical_failure_with_zero_match_raises():
+ adapters = {"coupang": FakeAdapter("coupang", fail=True), "naver": FakeAdapter("naver", by_query={})}
+ with pytest.raises(RuntimeError):
+ await build_search_handler(adapters)(_job()) # 0매칭 + 차단 → 기술 재시도
+
+
+# ── 오픈마켓 폴백 크롤 (네이버 미커버 몰만) ──────────────────────────
+async def test_fallback_crawls_only_uncovered_malls():
+ # 네이버 매칭에 G마켓은 있고(→크롤 생략), 11번가는 없음(→크롤). 옥션도 없음(→크롤).
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="G마켓")])}
+ gmarket = FakeAdapter("gmarket", products=[_np("gmarket", 4000, mall="G마켓")])
+ auction = FakeAdapter("auction", products=[_np("auction", 4500, mall="옥션")])
+ st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")])
+ handler = build_search_handler(
+ adapters, judge=FakeJudge(lambda c: True),
+ fallback_adapters={"gmarket": gmarket, "auction": auction, "st11": st11},
+ )
+ r = await handler(_job())
+ assert gmarket.calls == [] # 네이버가 G마켓 커버 → 크롤 생략
+ assert auction.calls and st11.calls # 미커버 → 크롤함
+ assert r["lowest"]["price"] == 3000 # 11번가 크롤가가 전체 최저
+ malls = {m["mall_name"] for m in r["by_mall"]}
+ assert malls == {"G마켓", "옥션", "11번가"} # 네이버 G마켓 + 크롤 옥션·11번가
+
+
+async def test_fallback_deadline_skips_slow_mall():
+ # 느린 폴백(데드라인 초과)은 스킵되고, 빠른 폴백은 병합된다 — 전체 지연에 상한.
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 9000, mall="네이버")])}
+ slow = FakeAdapter("gmarket", products=[_np("gmarket", 1000, mall="G마켓")], delay=1.0) # 데드라인 초과
+ fast = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")], delay=0.0)
+ r = await build_search_handler(
+ adapters, judge=FakeJudge(lambda c: True),
+ fallback_adapters={"gmarket": slow, "st11": fast},
+ fallback_deadline_sec=0.2,
+ )(_job())
+ malls = {m["mall_name"] for m in r["by_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():
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 9000, mall="네이버")])}
+ st11 = FakeAdapter("st11", fail=True) # 크롤 실패
+ r = await build_search_handler(
+ adapters, judge=FakeJudge(lambda c: True),
+ fallback_adapters={"st11": st11},
+ )(_job())
+ assert r["outcome"] == "found" and r["lowest"]["price"] == 9000 # 폴백 실패해도 정상 종료
+
+
+# ── 검색 원가 계측(metrics) ────────────────────────────────────────
+async def test_metrics_recorded_in_result():
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 1000), _np("naver", 2000)])}
+ adapters["naver"].last_bytes = 1234
+ judge = FakeJudge(lambda c: True, usage=_Usage(500, 40))
+ r = await build_search_handler(adapters, judge=judge, ai_model="gpt-4o-mini")(_job())
+ m = r["metrics"]
+ assert m["ai"]["calls"] == 1 and m["ai"]["prompt_tokens"] == 500 and m["ai"]["completion_tokens"] == 40
+ assert m["ai"]["est_cost_usd"] == round(500/1e6*0.15 + 40/1e6*0.60, 6) # gpt-4o-mini 단가
+ assert m["crawl"]["fetches"] == 1 and m["crawl"]["html_bytes"] == 1234
+ assert "naver" in m["source_ms"] and "duration_ms" in m
+
+
+async def test_metrics_cost_split_ai_and_proxy():
+ # 네이버(직접, 프록시X) + 프록시 경유 크롤 폴백 → proxy_usd 는 프록시 바이트만, ai_usd 는 토큰만
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="네이버")], last_bytes=2000)}
+ st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")], uses_proxy=True, last_bytes=1024**3) # 1GB
+ judge = FakeJudge(lambda c: True, usage=_Usage(1_000_000, 0)) # 1M prompt 토큰
+ handler = build_search_handler(
+ adapters, judge=judge, ai_model="gpt-4o-mini",
+ fallback_adapters={"st11": st11}, proxy_cost_per_gb=3.0,
+ )
+ m = (await handler(_job()))["metrics"]
+ # 네이버 2000B 는 프록시 경유 아님 → proxy_bytes = 1GB(st11)만
+ assert m["crawl"]["proxy_bytes"] == 1024**3
+ assert m["cost"]["proxy_usd"] == 3.0 # 1GB × $3
+ assert m["cost"]["ai_usd"] == round(m["ai"]["prompt_tokens"]/1e6*0.15, 6) # gpt-4o-mini input 단가
+ assert m["cost"]["total_usd"] == round(m["cost"]["ai_usd"] + 3.0, 6)
+
+
+async def test_metrics_counts_fallback_crawl():
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="네이버")])}
+ st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")])
+ st11.last_bytes = 9999
+ r = await build_search_handler(
+ adapters, judge=FakeJudge(lambda c: True),
+ fallback_adapters={"st11": st11},
+ )(_job())
+ m = r["metrics"]
+ assert m["crawl"]["fetches"] == 2 # naver + st11 크롤
+ assert "st11" in m["crawl"]["malls_crawled"] # 폴백 크롤 몰 기록
+ assert m["crawl"]["html_bytes"] == 9999 # st11 바이트 포함
+
+
+async def test_fallback_dedup_same_mall_keeps_lowest():
+ # 네이버 매칭에 G마켓 없음 → 크롤. 크롤 G마켓이 네이버 '네이버몰'보다 싸면 최저가 갱신.
+ adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 8000, mall="네이버")])}
+ gmarket = FakeAdapter("gmarket", products=[_np("gmarket", 6000, mall="G마켓"), _np("gmarket", 7000, mall="G마켓")])
+ r = await build_search_handler(
+ adapters, judge=FakeJudge(lambda c: True),
+ fallback_adapters={"gmarket": gmarket},
+ )(_job())
+ gm = [m for m in r["by_mall"] if m["mall_name"] == "G마켓"]
+ assert len(gm) == 1 and gm[0]["price"] == 6000 # 몰별 1건(최저)로 dedup
diff --git a/lps/tests/test_worker.py b/lps/tests/test_worker.py
new file mode 100644
index 0000000..e4b7ac8
--- /dev/null
+++ b/lps/tests/test_worker.py
@@ -0,0 +1,139 @@
+"""워커 루프 테스트 — drain→DONE, 실패→재시도→dead, reaper 회수 후 재처리, NOTIFY 깨움.
+핸들러는 fake(브라우저 없이) — 워커 로직만 결정론적으로 검증한다."""
+
+import asyncio
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import JobStatus, JobType
+from crud.job_crud import JobQueue
+from worker.notify import JobListener
+from worker.runner import Worker
+
+
+@pytest_asyncio.fixture
+async def q(db_engine):
+ async with db_engine.begin() as conn:
+ await conn.execute(text("TRUNCATE job"))
+ return JobQueue()
+
+
+async def test_worker_drains_all_to_done(q):
+ for i in range(3):
+ await q.enqueue(JobType.SEARCH.value, {"product_name": f"item{i}"})
+
+ async def handler(job):
+ return {"ok": True, "q": job["payload"]["product_name"]}
+
+ processed = await Worker("w1", q, handler).drain()
+ assert processed == 3
+ assert (await q.counts())["DONE"] == 3
+
+
+async def test_worker_failure_retries_then_dead(q):
+ await q.enqueue(JobType.SEARCH.value, {"product_name": "x"}, max_attempts=2)
+
+ async def boom(job):
+ raise RuntimeError("nope")
+
+ w = Worker("w1", q, boom, backoff_fn=lambda a: 0) # 백오프 0 → 즉시 재시도 가능
+ assert await w.process_one() is True # 1/2 실패 → PENDING
+ assert (await q.counts())["PENDING"] == 1
+ assert await w.process_one() is True # 2/2 실패 → DEAD
+ counts = await q.counts()
+ assert counts["DEAD"] == 1 and counts["PENDING"] == 0
+ assert await w.process_one() is False # DEAD 는 claim 대상 아님
+
+
+async def test_reaper_reclaims_then_worker_reprocesses(q):
+ jid = await q.enqueue(JobType.SEARCH.value, {"product_name": "x"})
+ await q.claim("dead-worker", lease_sec=1) # 점유 후 사망 흉내
+ await asyncio.sleep(1.3)
+ assert jid in await q.reap() # 회수 → PENDING
+
+ async def handler(job):
+ return {"ok": True}
+
+ assert await Worker("w2", q, handler).process_one() is True
+ assert (await q.counts())["DONE"] == 1
+
+
+async def test_job_deadline_cancels_hung_handler(q):
+ """핸들러 행 → 데드라인 초과 시 취소·fail 처리(재큐/DEAD)돼야 한다. 없으면 heartbeat 가
+ lease 를 계속 갱신해 워커 슬롯이 영구 점유된다(2026-07-10 부하테스트 실측)."""
+ jid = await q.enqueue(JobType.SEARCH.value, {"product_name": "hang"}, max_attempts=1)
+
+ async def hang(job):
+ await asyncio.sleep(3600)
+
+ w = Worker("w1", q, hang, backoff_fn=lambda a: 0, job_deadline_sec=0.2)
+ assert await w.process_one() is True # 행이어도 데드라인에 끊겨 반환된다
+ assert (await q.counts())["DEAD"] == 1 # max_attempts=1 → 즉시 DEAD
+ job = await q.get(jid)
+ assert "JobDeadlineExceeded" in job["last_error"]
+
+
+async def test_job_deadline_retries_before_dead(q):
+ """데드라인 초과도 일반 실패처럼 백오프 재큐를 탄다(시도 소진 전까지)."""
+ await q.enqueue(JobType.SEARCH.value, {"product_name": "hang"}, max_attempts=2)
+
+ async def hang(job):
+ await asyncio.sleep(3600)
+
+ w = Worker("w1", q, hang, backoff_fn=lambda a: 0, job_deadline_sec=0.2)
+ assert await w.process_one() is True
+ assert (await q.counts())["PENDING"] == 1 # 1/2 → 재큐
+ assert await w.process_one() is True
+ assert (await q.counts())["DEAD"] == 1 # 2/2 → DEAD
+
+
+async def test_ops_counts_long_running_as_stuck(q, db_engine):
+ """lease 가 계속 갱신돼도(행 상태의 heartbeat) 실행 10분 초과면 stuck_running 에 잡혀야 한다."""
+ await q.enqueue(JobType.SEARCH.value, {"product_name": "x"})
+ await q.claim("w1", lease_sec=3600) # lease 는 멀쩡(만료 안 됨)
+ assert (await q.ops())["stuck_running"] == 0
+ async with db_engine.begin() as conn:
+ await conn.execute(text("UPDATE job SET run_started_at = now() - interval '11 minutes' WHERE status = 2"))
+ assert (await q.ops())["stuck_running"] == 1
+
+
+async def test_browser_reaper_survives_hung_adapter():
+ """한 어댑터의 close 행이 정리 루프 전체를 멈추면 안 된다 — 타임아웃 후 다음 어댑터로."""
+ from worker_main import run_browser_reaper
+
+ class HungAdapter:
+ source = "hung"
+ async def close_if_idle(self, idle_sec):
+ await asyncio.sleep(3600)
+
+ class OkAdapter:
+ source = "ok"
+ closed = False
+ async def close_if_idle(self, idle_sec):
+ self.closed = True
+
+ ok = OkAdapter()
+ stop = asyncio.Event()
+ task = asyncio.create_task(run_browser_reaper(
+ [HungAdapter(), ok], stop, idle_sec=0, interval=0.01, close_timeout=0.05))
+ try:
+ await asyncio.wait_for(_until(lambda: ok.closed), timeout=3.0) # 행 어댑터를 지나 ok 까지 도달
+ finally:
+ stop.set()
+ await task
+
+
+async def _until(cond, poll: float = 0.02):
+ while not cond():
+ await asyncio.sleep(poll)
+
+
+async def test_enqueue_notifies_listener(q):
+ listener = JobListener()
+ await listener.start()
+ try:
+ await q.enqueue(JobType.SEARCH.value, {"product_name": "x"}) # pg_notify 발생
+ assert await listener.wait(3.0) is True # 즉시 깨어남
+ finally:
+ await listener.close()
diff --git a/lps/web_main.py b/lps/web_main.py
new file mode 100644
index 0000000..6f94403
--- /dev/null
+++ b/lps/web_main.py
@@ -0,0 +1,46 @@
+# 실행 방법
+# pip install -r requirements.txt
+# python web_main.py # 기본 local 환경
+# APP_ENV=dev python web_main.py # 환경 지정
+#
+# 또는 uvicorn 직접 실행:
+# uvicorn router.router:app --reload --host=0.0.0.0 --port=9600
+
+import uvicorn
+
+from common.logger import LOG
+from config.server_configs import web_server_config, main_db_config
+
+LOG.SetPrefix(web_server_config.server_name)
+
+# import 시점에 app 및 DB 세션 매니저(싱글톤)가 초기화된다.
+import router.router
+
+if __name__ == "__main__":
+ LOG.i(f"Server Name : {web_server_config.server_name}")
+ LOG.i(f"Server Port : {web_server_config.port}")
+ LOG.i(f"API Server start time : {router.router.API_SERVER_START_TIME}")
+ # 실효 커넥션 풀(자동 산정 결과) — 멀티워커 시 커넥션 예산 준수 여부 확인용.
+ _pc = web_server_config.process_count
+ _conn = (main_db_config.pool_size + main_db_config.max_overflow) * 2 * _pc
+ LOG.i(f"DB Pool : pool_size={main_db_config.pool_size} max_overflow={main_db_config.max_overflow} "
+ f"× 2engine × {_pc}workers = {_conn} conns (budget={main_db_config.connection_budget})")
+
+ if web_server_config.is_ssl:
+ uvicorn.run(
+ "router.router:app",
+ host="0.0.0.0",
+ port=web_server_config.port,
+ access_log=False,
+ workers=web_server_config.process_count,
+ ssl_keyfile="./SSL/key.pem",
+ ssl_certfile="./SSL/cert.pem",
+ )
+ else:
+ uvicorn.run(
+ "router.router:app",
+ host="0.0.0.0",
+ port=web_server_config.port,
+ access_log=False,
+ workers=web_server_config.process_count,
+ )
diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py
new file mode 100644
index 0000000..21be864
--- /dev/null
+++ b/lps/worker/handlers.py
@@ -0,0 +1,241 @@
+"""잡 핸들러 — job_type 별 처리. 현재는 SEARCH(검색)만.
+
+검색 핸들러 = 한정된 재정제 루프 + 명시적 outcome:
+ 0. 네거티브 캐시 확인(최근 not_found면 즉시 반환)
+ 각 라운드(원본 → 정밀(LLM) → 광역, 최대 max_rounds):
+ 소스 동시 검색 → 필터 → 이상치 → AI 같은상품 판정
+ ├ 매칭 있음 → DONE(outcome=found), 조기 종료
+ ├ 0매칭 + 기술적 실패(차단/예외) 있음 → raise → 큐가 잡 전체 백오프 재시도(→소진 시 DEAD)
+ └ 0매칭 + 소스 정상 → 다음 라운드
+ 라운드 소진 → DONE(outcome=not_found) + 네거티브 캐시 기록
+
+두 재시도 축을 분리한다: 기술적(큐 attempts/백오프) ≠ 검색어(refine 라운드, 유한).
+'못 찾음'은 정상 종료(DONE)지 dead-letter 가 아니다.
+"""
+
+import asyncio
+import time
+
+from common.enums import JobType
+from common.logger import LOG
+from services.metrics import SearchMetrics
+from services.search.contract import SearchAdapter, NormalizedProduct
+from services.search.card_parser import canonical_mall, MALL_BY_SOURCE
+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):
+ items = [p for p in matched if p.source == src]
+ return min(items, key=lambda p: p.price) if items else None
+
+ n, c = lowest("naver"), lowest("coupang")
+ # 최종 최저가는 소스 무관 전체 매칭 중 최저(G마켓·옥션·11번가 등 폴백 포함).
+ f = min(matched, key=lambda p: p.price) if matched else None
+ return {
+ "matched_count": len(matched),
+ "naver_lowest": n.price if n else None, "naver_name": n.name if n else None, "naver_url": n.detail_url if n else None,
+ "coupang_lowest": c.price if c else None, "coupang_name": c.name if c else None, "coupang_url": c.detail_url if c else None,
+ "final_lowest": f.price if f else None, "final_source": f.source if f else None,
+ "by_mall": summarize_by_mall(matched), # 몰별 최저가 스냅샷(열린 스키마)
+ }
+
+
+def build_search_handler(
+ adapters: dict[str, SearchAdapter],
+ sources: list[str] | None = None,
+ limit: int = 40,
+ top_n: int = 5,
+ judge=None,
+ keyword_gen=None,
+ max_rounds: int = 3,
+ neg_cache=None,
+ history=None,
+ fallback_adapters: dict[str, SearchAdapter] | None = None,
+ ai_model: str = "",
+ proxy_cost_per_gb: float = 0.0,
+ fallback_deadline_sec: float = 15.0,
+):
+ """검색 핸들러 생성.
+ judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) /
+ neg_cache: NegativeCache(TTL not_found 캐시) / history: PriceHistory(최저가 스냅샷) /
+ fallback_adapters: 오픈마켓 크롤(gmarket/auction/st11) — 네이버가 그 몰을 커버 못 했을 때만 크롤(폴백).
+ 모두 선택 — 없으면 해당 단계 생략."""
+ use = list(sources) if sources else list(adapters.keys())
+ fallbacks = fallback_adapters or {}
+
+ async def _record_history(product_code: str, job_id, outcome: str, matched: list):
+ if history is None:
+ return
+ event = {"product_code": product_code, "job_id": job_id, "outcome": outcome, **_price_snapshot(matched)}
+ try:
+ await history.record(event)
+ except Exception as ex:
+ LOG.e_no_callstack(f"[history] 스냅샷 기록 실패(무시): {ex}")
+
+ async def _timed_search(adapter, query: str, source: str, metrics: SearchMetrics, crawl: bool):
+ """어댑터 검색 1건을 타이밍+바이트 계측하며 실행. 예외는 그대로 전파(호출부가 처리)."""
+ via_proxy = bool(getattr(adapter, "uses_proxy", False))
+ t0 = time.monotonic()
+ try:
+ res = await adapter.search(query, limit=limit)
+ metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl, via_proxy=via_proxy)
+ return res
+ except BaseException: # CancelledError(데드라인 취소) 포함 — 소요/바이트는 계측하고 재전파
+ metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl, via_proxy=via_proxy)
+ raise
+
+ async def _search_round(query: str, metrics: SearchMetrics):
+ """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed). 소스별 시간/바이트 계측."""
+ results = await asyncio.gather(*[_timed_search(adapters[s], query, s, metrics, False) for s in use],
+ return_exceptions=True)
+ products, per_source, tech_failed = [], {}, False
+ for src, res in zip(use, results):
+ if isinstance(res, Exception):
+ tech_failed = True
+ per_source[src] = {"error": f"{type(res).__name__}: {res}"}
+ LOG.w(f"[{src}] 검색 실패: {type(res).__name__}: {res}")
+ else:
+ products.extend(res)
+ per_source[src] = {"count": len(res)}
+ return products, per_source, tech_failed
+
+ async def _match(target: dict, products: list, base_price, metrics: SearchMetrics):
+ """필터 → (있으면) 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) # ← 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}
+ todo = [(src, ad) for src, ad in fallbacks.items() if MALL_BY_SOURCE.get(src, src) not in covered]
+ if not todo:
+ return matched
+
+ async def _crawl_match(src, adapter):
+ # 폴백은 '있으면 좋은' 보강이라 데드라인을 건다 — 초과 시 그 몰만 스킵(전체 지연에 상한).
+ # 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 []
+ hits = await _match(target, crawled, base_price, metrics)
+ if hits:
+ LOG.d(f"[fallback:{src}] 크롤 {len(crawled)}건 중 같은상품 {len(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):
+ """라운드 쿼리 지연 생성: 원본 → (0매칭 시에만 LLM 호출로) 정밀 → 광역."""
+ yield ("original", base_query)
+ if keyword_gen is not None:
+ kw = await keyword_gen.generate(target) # 원본이 실패해 여기까지 온 경우에만 호출됨
+ metrics.add_ai(keyword_gen.last_usage)
+ seen = {base_query}
+ for label, q in (("precise", kw.precise), ("broad", kw.broad)):
+ q = (q or "").strip()
+ if q and q not in seen:
+ seen.add(q)
+ yield (label, q)
+
+ async def handler(job: dict) -> dict:
+ if job["job_type"] != JobType.SEARCH.value:
+ raise ValueError(f"unsupported job_type: {job['job_type']}")
+
+ payload = job.get("payload") or {}
+ base_query = (payload.get("product_name") or "").strip()
+ if not base_query:
+ raise ValueError("empty product_name")
+ target = {k: payload.get(k, "") for k in ("product_name", "model", "specification", "company")}
+ base_price = parse_price(payload.get("price"))
+ cache_key = payload.get("product_code") or base_query
+ metrics = SearchMetrics(ai_model, proxy_cost_per_gb) # 검색 1건의 리소스/비용/시간 계측
+
+ # 0) 네거티브 캐시 — 최근 not_found면 재검색 생략
+ if neg_cache is not None and await neg_cache.is_negative(cache_key):
+ return {"outcome": "not_found", "cached": True, "query": base_query,
+ "rounds_tried": 0, "lowest": None, "top": [], "stages": [], "sources": {},
+ "metrics": metrics.snapshot()}
+
+ rounds_done = 0
+ last_stages, last_sources = [], {}
+ async for label, query in _round_queries(base_query, target, metrics):
+ if rounds_done >= max_rounds:
+ break
+ rounds_done += 1
+
+ products, per_source, tech_failed = await _search_round(query, metrics)
+ candidates, stages = 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)
+ matched = [c for c, v in zip(candidates, verdicts) if v.is_match]
+ stages.append({"stage": "ai_match", "in": len(candidates), "out": len(matched)})
+ candidates = matched
+ last_stages, last_sources = stages, per_source
+
+ if candidates: # 찾음 → 오픈마켓 폴백 보강 후 종료
+ before = len(candidates)
+ candidates = await _enrich_with_fallback(target, query, candidates, base_price, metrics)
+ if len(candidates) > before:
+ stages.append({"stage": "fallback_crawl", "in": before, "out": len(candidates)})
+ result = rank_result(candidates, len(products), stages, top_n)
+ result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done,
+ sources=per_source, metrics=metrics.snapshot())
+ await _record_history(cache_key, job.get("job_id"), "found", candidates)
+ return result
+
+ if tech_failed: # 0매칭인데 소스가 죽어 있었음 → '없음'이라 단정 불가 → 기술 재시도
+ raise RuntimeError(f"기술적 실패로 0매칭(round={label}) — 잡 재시도: {per_source}")
+
+ # 모든 라운드 클린 0매칭 → 정상 not_found 종료
+ if neg_cache is not None:
+ await neg_cache.put(cache_key, reason=f"not_found after {rounds_done} rounds")
+ result = rank_result([], 0, last_stages, top_n)
+ result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done,
+ sources=last_sources, metrics=metrics.snapshot())
+ await _record_history(cache_key, job.get("job_id"), "not_found", [])
+ return result
+
+ return handler
diff --git a/lps/worker/notify.py b/lps/worker/notify.py
new file mode 100644
index 0000000..b90320d
--- /dev/null
+++ b/lps/worker/notify.py
@@ -0,0 +1,47 @@
+"""LISTEN/NOTIFY 리스너 — 잡 적재 시 워커를 즉시 깨운다(폴링 낭비 제거).
+
+전용 asyncpg 연결로 LISTEN 한다(SQLAlchemy 풀과 분리). 알림이 오면 이벤트를 세팅하고,
+워커는 claim 이 비었을 때 wait()로 알림 또는 짧은 타임아웃(안전망/reaper)까지 대기한다.
+"""
+
+import asyncio
+
+import asyncpg
+
+from config.server_configs import main_db_config
+from crud.job_crud import JOB_NOTIFY_CHANNEL
+
+
+def _dsn() -> str:
+ c = main_db_config
+ pw = f":{c.write_pw}" if c.write_pw else ""
+ return f"postgresql://{c.write_id}{pw}@{c.write_host}:{c.write_port}/{c.name}"
+
+
+class JobListener:
+ def __init__(self, channel: str = JOB_NOTIFY_CHANNEL):
+ self._channel = channel
+ self._conn: asyncpg.Connection | None = None
+ self._event = asyncio.Event()
+
+ async def start(self):
+ self._conn = await asyncpg.connect(_dsn())
+ await self._conn.add_listener(self._channel, self._on_notify)
+
+ def _on_notify(self, *_args):
+ self._event.set()
+
+ async def wait(self, timeout: float) -> bool:
+ """알림이 오거나 timeout 까지 대기. 알림으로 깨면 True, 타임아웃이면 False."""
+ try:
+ await asyncio.wait_for(self._event.wait(), timeout)
+ return True
+ except asyncio.TimeoutError:
+ return False
+ finally:
+ self._event.clear()
+
+ async def close(self):
+ if self._conn is not None:
+ await self._conn.close()
+ self._conn = None
diff --git a/lps/worker/runner.py b/lps/worker/runner.py
new file mode 100644
index 0000000..77f6b30
--- /dev/null
+++ b/lps/worker/runner.py
@@ -0,0 +1,99 @@
+"""워커 루프 + reaper.
+
+워커는 큐에서 잡을 원자적으로 claim → 핸들러 실행 → complete/fail 한다.
+- 처리 중 heartbeat 로 lease 를 갱신(긴 잡이 reaper 에 회수되지 않게).
+- 핸들러엔 데드라인(job_deadline_sec)을 건다 — heartbeat 가 lease 를 계속 갱신하므로
+ 핸들러가 행하면 reaper 로는 영원히 회수 불가(2026-07-10 부하테스트에서 크롤 15분 행 실측).
+ 초과 시 취소 후 fail 처리 → 백오프 재큐(소진 시 DEAD), 워커 슬롯은 즉시 다음 잡으로.
+- claim 이 비면 LISTEN 알림 또는 짧은 타임아웃까지 대기(폴링 최소화).
+- 핸들러는 주입식(async def(job)->dict) — 프로덕션은 검색 파이프라인, 테스트는 fake.
+"""
+
+import asyncio
+
+from common.enums import JobStatus
+from common.logger import LOG
+from crud.job_crud import JobQueue, compute_backoff
+
+
+class Worker:
+ def __init__(self, worker_id: str, queue: JobQueue, handler, lease_sec: int = 120, backoff_fn=compute_backoff,
+ job_deadline_sec: float = 300.0):
+ self.worker_id = worker_id
+ self.queue = queue
+ self.handler = handler
+ self.lease_sec = lease_sec
+ self.backoff_fn = backoff_fn
+ self.job_deadline_sec = job_deadline_sec # 잡 1건 처리 시간 상한(0 이면 무제한 — 테스트용)
+
+ async def process_one(self) -> bool:
+ """대기 잡 1건을 claim·처리. 처리했으면 True, 없으면 False."""
+ job = await self.queue.claim(self.worker_id, self.lease_sec)
+ if not job:
+ return False
+ await self._process(job)
+ return True
+
+ async def drain(self) -> int:
+ """큐가 빌 때까지 처리(테스트/일회성 배치용). 처리한 잡 수 반환."""
+ n = 0
+ while await self.process_one():
+ n += 1
+ return n
+
+ async def run(self, listener=None, stop: asyncio.Event | None = None, idle_timeout: float = 5.0):
+ """상시 루프. stop 이 설정될 때까지 처리하고, 유휴 시 알림/타임아웃까지 대기."""
+ stop = stop or asyncio.Event()
+ while not stop.is_set():
+ worked = await self.process_one()
+ if not worked:
+ if listener is not None:
+ await listener.wait(idle_timeout)
+ else:
+ await asyncio.sleep(idle_timeout)
+
+ async def _process(self, job: dict):
+ jid = job["job_id"]
+ hb = asyncio.create_task(self._heartbeat(jid))
+ try:
+ if self.job_deadline_sec > 0:
+ result = await asyncio.wait_for(self.handler(job), timeout=self.job_deadline_sec)
+ else:
+ result = await self.handler(job)
+ await self.queue.complete(jid, self.worker_id, result)
+ LOG.d(f"[{self.worker_id}] done {jid}")
+ except TimeoutError:
+ # 데드라인 초과 — wait_for 가 핸들러 태스크를 취소한 뒤 여기로 온다. in-flight 크롤이
+ # 취소되며 브라우저가 어중간한 상태로 남을 수 있지만, 어댑터가 다음 검색에서 재기동으로
+ # 회복한다. 행이 워커 슬롯을 영구 점유하는 것보다 낫다.
+ backoff = self.backoff_fn(job["attempts"])
+ st = await self.queue.fail(jid, self.worker_id, f"JobDeadlineExceeded: {self.job_deadline_sec:.0f}s", backoff)
+ LOG.w(f"[{self.worker_id}] deadline {jid} → {JobStatus(st).name if st else '?'} ({self.job_deadline_sec:.0f}s 초과, 핸들러 취소)")
+ except Exception as ex:
+ backoff = self.backoff_fn(job["attempts"])
+ st = await self.queue.fail(jid, self.worker_id, f"{type(ex).__name__}: {ex}", backoff)
+ LOG.w(f"[{self.worker_id}] fail {jid} → {JobStatus(st).name if st else '?'} ({type(ex).__name__}: {ex})")
+ finally:
+ hb.cancel()
+ try:
+ await hb
+ except asyncio.CancelledError:
+ pass
+
+ async def _heartbeat(self, jid: str):
+ interval = max(1, self.lease_sec // 3)
+ while True:
+ await asyncio.sleep(interval)
+ await self.queue.renew_lease(jid, self.worker_id, self.lease_sec)
+
+
+async def run_reaper(queue: JobQueue, stop: asyncio.Event, interval: float = 30.0):
+ """만료 lease(워커 사망) 잡을 주기적으로 회수. 재시도 남으면 재큐, 소진되면 DEAD."""
+ while not stop.is_set():
+ reclaimed = await queue.reap()
+ if reclaimed:
+ LOG.w(f"[reaper] reclaimed {len(reclaimed)} stale job(s)")
+ try:
+ await asyncio.wait_for(stop.wait(), interval)
+ except asyncio.TimeoutError:
+ pass
diff --git a/lps/worker_main.py b/lps/worker_main.py
new file mode 100644
index 0000000..c0195fc
--- /dev/null
+++ b/lps/worker_main.py
@@ -0,0 +1,276 @@
+# LPS 워커 프로세스 진입점 (API 와 분리 실행 — 코드베이스 공유, 독립 스케일).
+# python worker_main.py
+# WORKER_CONCURRENCY=3 python worker_main.py # 상품 3개 동시 검색(권장 2~3, 로컬)
+#
+# 브라우저 어댑터는 컨텍스트당 직렬(lock)이라, 진짜 병렬을 위해 **워커마다 자기 브라우저 세트**를 준다:
+# 프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP). 동시성 N → 최대 4×N Chrome.
+
+import asyncio
+import os
+import signal
+import time
+
+import httpx
+
+from common.logger import LOG
+from config.server_configs import web_server_config, openai_config, decodo_config
+from crud.job_crud import JobQueue
+from crud.negative_cache import NegativeCache
+from crud.bot_detection import BotDetectionLog
+from crud.price_history import PriceHistory
+from services.search.proxy import DecodoProxy
+from services.search.coupang.adapter import CoupangAdapter
+from services.search.naver.adapter import NaverAdapter
+from services.search.esm.adapter import EsmAdapter
+from services.search.st11.adapter import ElevenStAdapter
+from services.ai.similarity import SimilarityJudge
+from services.ai.keyword import KeywordGenerator
+from worker.handlers import build_search_handler
+from worker.notify import JobListener
+from worker.runner import Worker, run_reaper
+
+LOG.SetPrefix(f"{web_server_config.server_name}-worker")
+
+# 오픈마켓 폴백(G마켓·옥션·11번가)은 **기본 비활성** — 2026-07-10 협의 결정.
+# 실측상 크롤 몰이 최종 최저가를 바꾼 적이 없고(0회), 검색당 최대 15s + 프록시 대역폭의
+# 대부분을 차지해 로직에서 제외했다(코드·테스트는 유지, 핸들러는 빈 폴백을 정상 처리).
+# 재가동: LPS_FALLBACKS=gmarket,auction,st11 (일부만도 가능) — 켜기 전 라이브 스모크로
+# 셀렉터 드리프트 점검. 배경은 docs/decision-openmarket-crawler.md.
+_FALLBACK_SOURCES = ("gmarket", "auction", "st11")
+
+
+def _enabled_fallbacks() -> list[str]:
+ names = [s.strip() for s in os.environ.get("LPS_FALLBACKS", "").split(",") if s.strip()]
+ unknown = [n for n in names if n not in _FALLBACK_SOURCES]
+ if unknown:
+ LOG.w(f"LPS_FALLBACKS 무시된 값: {unknown} (가능: {list(_FALLBACK_SOURCES)})")
+ return [n for n in names if n in _FALLBACK_SOURCES]
+
+
+def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history):
+ """워커 1개의 자립 세트(브라우저 어댑터·AI·핸들러)를 만든다.
+ 프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP)로 진짜 병렬을 보장한다."""
+ # 워커별 프록시(다른 포트=다른 IP). 100포트를 워커 수로 균등 분할해 시작점을 벌린다.
+ proxy = DecodoProxy()
+ if proxy.enabled and concurrency > 1:
+ n = proxy.port_end - proxy.port_start + 1
+ proxy.seed_offset(i * max(1, n // concurrency))
+ bot_log = BotDetectionLog()
+ suffix = f"_w{i}" if concurrency > 1 else ""
+
+ def _pf(source): # 워커별 Chrome 프로필 경로(중복 실행 시 ProcessSingleton 충돌 방지)
+ # LPS_PROFILE_DIR 를 영속 볼륨으로 마운트하면 재시작해도 cf_clearance 등 쿠키 유지(재웜업 회피).
+ base = os.environ.get("LPS_PROFILE_DIR", "/tmp")
+ return f"{base}/lps_{source}{suffix}"
+
+ adapters = {
+ "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, on_detect=bot_log.record),
+ "naver": NaverAdapter(), # httpx 직접(프록시 미경유) — 워커별 인스턴스(last_bytes 경합 회피)
+ }
+ # 폴백은 기본 비활성(LPS_FALLBACKS 로 켬 — 상단 주석 참고). 켤 땐 데드라인이 상한이라
+ # 봇감지 재시도(챌린지 대기 2배)를 끈다(max_block_retries=0) — 빠르게 포기·스킵.
+ fallback_adapters = {}
+ for name in _enabled_fallbacks():
+ if name == "st11":
+ fallback_adapters[name] = ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy, on_detect=bot_log.record, max_block_retries=0)
+ else:
+ fallback_adapters[name] = EsmAdapter(name, headless=False, user_data_dir=_pf(name), proxy=proxy, on_detect=bot_log.record, max_block_retries=0)
+ # AI 도 워커별 인스턴스 — 공유 상태(last_usage) 경합 원천 제거
+ judge = SimilarityJudge() if has_openai else None
+ keyword_gen = KeywordGenerator() if has_openai else None
+ handler = build_search_handler(
+ adapters, judge=judge, keyword_gen=keyword_gen,
+ neg_cache=neg_cache, history=history,
+ fallback_adapters=fallback_adapters,
+ ai_model=openai_config.model,
+ proxy_cost_per_gb=decodo_config.cost_per_gb,
+ )
+ return handler, list(adapters.values()) + list(fallback_adapters.values())
+
+
+async def _warmup_worker(worker_adapters, tries: int = 3, attempt_timeout: float = 60.0):
+ """워커의 챌린지 소스(Turnstile/Akamai)를 미리 풀어 쿠키(cf_clearance 등)를 확보한다.
+ 콜드 비용을 시작 시 몰아, 이후 실 작업은 웜(빠름). 백그라운드로 돌려 잡 처리를 막지 않는다.
+ 나쁜 IP 는 인터랙티브 Turnstile 로 에스컬레이션되므로, 실패 시 **다른 IP 로 회전 재시도**한다.
+ 시도당 타임아웃 필수 — 웜업은 search 중 어댑터 락을 쥐므로, 여기서 행하면 그 워커의
+ 모든 실 검색이 락 대기로 함께 멈춘다(2026-07-10 부하테스트에서 15분 행 실측)."""
+ for ad in worker_adapters:
+ if ad.source not in ("gmarket", "auction", "coupang"):
+ continue
+ for attempt in range(tries):
+ try:
+ await asyncio.wait_for(ad.search("생수", limit=1), timeout=attempt_timeout)
+ LOG.i(f"[warmup:{ad.source}] 챌린지 통과·쿠키 확보 (시도 {attempt + 1})")
+ break
+ except Exception as ex:
+ if attempt < tries - 1:
+ ad._rotate_ip(f"웜업 재시도({type(ex).__name__}) — 새 IP")
+ else:
+ LOG.w(f"[warmup:{ad.source}] {tries}회 실패(첫 잡에서 재시도): {type(ex).__name__}")
+
+
+async def _post_webhook(url: str, text: str, snap: dict):
+ """Slack 호환 웹훅으로 알림 전송(있을 때만). 실패는 무시."""
+ try:
+ async with httpx.AsyncClient(timeout=5) as c:
+ await c.post(url, json={"text": f":rotating_light: LPS {text}\n```{snap}```"})
+ except Exception:
+ pass
+
+
+async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0):
+ """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가
+ 행/좀비 워커 감지) (2) 큐/차단 지표 점검 → 임계 초과 시 WARN 로그 + (env 있으면) 웹훅 알림."""
+ hb_path = os.environ.get("LPS_HEARTBEAT_FILE", "/tmp/lps_worker_heartbeat")
+ webhook = os.environ.get("LPS_ALERT_WEBHOOK")
+ th_dead = int(os.environ.get("LPS_ALERT_DEAD_1H", "20"))
+ th_blocks = int(os.environ.get("LPS_ALERT_BLOCKS_1H", "80"))
+ th_lag = int(os.environ.get("LPS_ALERT_QUEUE_LAG_SEC", "300"))
+ while not stop.is_set():
+ try:
+ with open(hb_path, "w") as f:
+ f.write(str(int(time.time()))) # 하트비트(mtime) — HEALTHCHECK 가 신선도 확인
+ except Exception:
+ pass
+ try:
+ snap = await queue.ops()
+ snap["blocks_1h"] = await bot_log.recent_count(60)
+ alerts = []
+ if snap["dead_1h"] >= th_dead: alerts.append(f"DEAD 1h={snap['dead_1h']}")
+ if snap["blocks_1h"] >= th_blocks: alerts.append(f"차단 1h={snap['blocks_1h']}")
+ if snap["oldest_pending_sec"] >= th_lag: alerts.append(f"큐지연={snap['oldest_pending_sec']}s")
+ if snap["stuck_running"] > 0: alerts.append(f"stuck={snap['stuck_running']}")
+ if alerts:
+ msg = "[ops-alert] " + " · ".join(alerts)
+ LOG.w(msg)
+ if webhook:
+ await _post_webhook(webhook, msg, snap)
+ except Exception as ex:
+ LOG.e_no_callstack(f"[ops-monitor] {type(ex).__name__}: {ex}")
+ try:
+ await asyncio.wait_for(stop.wait(), timeout=interval)
+ except asyncio.TimeoutError:
+ pass
+
+
+async def run_browser_reaper(adapters, stop, idle_sec: float = 120.0, interval: float = 30.0,
+ close_timeout: float = 60.0):
+ """유휴 브라우저 정리 루프 — 일정 시간 검색 없는 어댑터의 Chrome 을 닫아 메모리를 회수한다.
+ 쿠키는 user_data_dir 에 남아, 다음 검색 때 재기동해도 (같은 IP면) 웜 유지.
+ 순차 순회라 close 1건에도 타임아웃을 건다 — 한 어댑터의 close 행이 루프 전체를 멈춰
+ 다른 워커의 브라우저까지 못 닫게 되는 것을 실측(2026-07-10 부하테스트)했다."""
+ while not stop.is_set():
+ try:
+ await asyncio.wait_for(stop.wait(), timeout=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 asyncio.wait_for(close_if_idle(idle_sec), timeout=close_timeout)
+ except asyncio.TimeoutError:
+ LOG.w(f"[browser-reaper] {getattr(ad, 'source', '?')} 정리 {close_timeout:.0f}s 초과 — 취소·스킵(다음 틱 재시도)")
+ except Exception as ex:
+ LOG.e_no_callstack(f"[browser-reaper] 정리 실패(무시): {ex}")
+
+
+async def main(concurrency: int = 1):
+ queue = JobQueue()
+ neg_cache, history = NegativeCache(), PriceHistory() # DB 기반 — 워커 공유 안전
+ has_openai = bool(openai_config.api_key)
+
+ # 시작 프리플라이트: DECODO 게이트가 살아있는지(인증) 대표 프록시로 1회 확인. 포트는 워커별로 각자 잡음.
+ probe = DecodoProxy()
+ LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(probe.session_minutes) + '분 회전)' if probe.enabled else 'OFF(미설정)'}")
+ if probe.enabled:
+ egress_ip, egress_port = await probe.healthcheck()
+ LOG.i(f"DECODO 프리플라이트 OK — egress IP {egress_ip} (port {egress_port})") if egress_ip \
+ else LOG.w("DECODO 프리플라이트 실패 — 살아있는 포트를 못 찾음(런타임 회전으로 재시도)")
+ fb = _enabled_fallbacks()
+ LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'} · "
+ f"오픈마켓 폴백: {', '.join(fb) if fb else 'OFF(기본 — LPS_FALLBACKS 로 활성화)'}")
+
+ stop = asyncio.Event()
+ listeners: list[JobListener] = []
+ tasks: list[asyncio.Task] = []
+ bg_tasks: list[asyncio.Task] = [] # 웜업 등 백그라운드(짧게 끝남, gather 대상 아님)
+ all_adapters = []
+
+ # ── graceful shutdown: SIGINT(Ctrl+C)/SIGTERM(docker stop) → stop 이벤트 ──
+ # asyncio.run 기본 동작(SIGINT=메인 태스크 즉시 cancel)은 하던 잡을 도중에 끊어
+ # RUNNING 인 채 lease 만료(120s)까지 묶어둔다. 대신 stop 을 set 해 "새 잡은 안 받고,
+ # 하던 잡은 마무리"로 종료한다. 같은 신호를 한 번 더 받으면 강제 종료(태스크 취소).
+ def _request_stop(sig_name: str):
+ if not stop.is_set():
+ LOG.i(f"{sig_name} 수신 — graceful 종료: 새 잡 중단, 하던 잡 마무리 (한 번 더 = 강제 종료)")
+ stop.set()
+ for t in bg_tasks: # 웜업은 선택 작업 — 즉시 취소해 어댑터 락을 비운다
+ t.cancel()
+ else:
+ LOG.w(f"{sig_name} 재수신 — 강제 종료(실행 중 잡은 lease 만료 후 reaper 가 재큐)")
+ for t in tasks:
+ t.cancel()
+
+ loop = asyncio.get_running_loop()
+ for sig in (signal.SIGINT, signal.SIGTERM):
+ loop.add_signal_handler(sig, _request_stop, sig.name)
+
+ # 잡 1건 데드라인 — 정상 검색은 폴백 포함 수분 내 끝난다(실측 15~22s). 크롤 행 실측(15분) 대비 상한.
+ job_deadline = float(os.environ.get("LPS_JOB_DEADLINE_SEC", "300"))
+ for i in range(concurrency):
+ handler, worker_adapters = _build_worker(i, concurrency, has_openai, neg_cache, history)
+ all_adapters += worker_adapters
+ bg_tasks.append(asyncio.create_task(_warmup_worker(worker_adapters))) # 챌린지 쿠키 선점(백그라운드)
+ listener = JobListener()
+ await listener.start()
+ listeners.append(listener)
+ worker = Worker(f"worker-{i}", queue, handler, job_deadline_sec=job_deadline)
+ tasks.append(asyncio.create_task(worker.run(listener, stop)))
+
+ tasks.append(asyncio.create_task(run_reaper(queue, stop)))
+ tasks.append(asyncio.create_task(run_browser_reaper(all_adapters, stop))) # 유휴 브라우저 정리
+ tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop))) # 하트비트 + 임계 알림
+ LOG.i(f"LPS 워커 {concurrency}개 + reaper + 브라우저정리 + ops모니터(하트비트/알림) 기동 (워커별 세트 · 상품 {concurrency}개 동시)")
+
+ # 종료 유예: stop 후 하던 잡이 이 시간 안에 끝나면 자연 종료, 초과하면 강제 취소.
+ # docker stop 을 쓰면 compose 의 stop_grace_period 를 이보다 길게 잡아야 SIGKILL 전에 마무리된다.
+ grace = float(os.environ.get("LPS_SHUTDOWN_GRACE_SEC", "60"))
+ gathered = asyncio.gather(*tasks)
+ stop_waiter = asyncio.create_task(stop.wait())
+ try:
+ await asyncio.wait({gathered, stop_waiter}, return_when=asyncio.FIRST_COMPLETED)
+ if gathered.done():
+ gathered.result() # 워커/리퍼가 예외로 죽은 경우 → 전파(finally 가 정리 후 종료)
+ else:
+ # 종료 신호 경로 — 워커 루프들이 stop 을 보고 하던 잡을 마친 뒤 스스로 끝나길 기다린다
+ try:
+ await asyncio.wait_for(gathered, timeout=grace)
+ LOG.i("graceful 종료 — 모든 워커가 하던 잡을 마무리함")
+ except asyncio.TimeoutError:
+ LOG.w(f"종료 유예 {grace:.0f}s 초과 — 남은 태스크 강제 취소(잡은 lease 만료 후 재큐)")
+ except asyncio.CancelledError: # 신호 재수신(강제 종료)로 태스크가 취소된 경우
+ LOG.w("강제 종료 — 남은 리소스 정리 후 종료")
+ finally:
+ stop.set()
+ stop_waiter.cancel()
+ for t in (*tasks, *bg_tasks):
+ t.cancel()
+ # 취소 완주를 기다린 뒤 정리 — 실행 중 태스크가 브라우저/커넥션을 쓰는 채로 닫지 않게
+ await asyncio.gather(gathered, stop_waiter, *bg_tasks, return_exceptions=True)
+ for listener in listeners:
+ try:
+ await listener.close()
+ except Exception as ex:
+ LOG.e_no_callstack(f"[shutdown] 리스너 정리 실패(무시): {ex}")
+ for adapter in all_adapters: # 항목별 격리 — 하나가 실패해도 나머지 Chrome 은 닫는다
+ try:
+ await adapter.close()
+ except Exception as ex:
+ LOG.e_no_callstack(f"[shutdown] {getattr(adapter, 'source', '?')} 정리 실패(무시): {ex}")
+ LOG.i("LPS 워커 종료 완료")
+
+
+if __name__ == "__main__":
+ asyncio.run(main(int(os.environ.get("WORKER_CONCURRENCY", "1"))))
