feat(lps): 알림 룰 4종 추가 — 데드라인·검색원가·포트 고갈·예산 누수

협의로 선정한 조기 신호 4종을 AlertManager 에 추가한다.

- deadline: 최근 1h JobDeadlineExceeded 수 ≥ LPS_ALERT_DEADLINE_1H(5).
  재시도로 살아나면 dead 룰엔 안 잡히는 크롤 행 반복 신호를 별도 집계.
- cost: 최근 1h 완료 잡 검색원가 합 ≥ LPS_ALERT_COST_1H_USD(1.0).
  비용의 87%가 프록시 대역폭 — 리소스차단 풀림·재시도 루프의 조용한
  비용 폭주를 감시. job.result 의 metrics.cost.total_usd JSONB 합산.
- proxy_ports_low: 가용 포트 비율 ≤ LPS_ALERT_PORTS_LOW_PCT(30%).
  쿨다운 격리 누적 — blocks_1h(80건)보다 먼저 우는 대규모 차단 조기
  신호. 워커별 프록시 중 가장 소진된 것 기준(min).
- budget_leak: 최근 6h end_reason=block 세션 ≥ LPS_ALERT_BLOCK_SESSIONS_6H(1).
  요청 예산(3회)을 지켰는데도 차단됨 = 예산 하향 검토 신호.
- deadline_1h·cost_1h_usd 는 queue.ops() 에 편입 → /v1/lps/ops 로도 노출.
  포트·세션 지표는 워커 웹훅 스냅샷에 포함(프록시 상태는 워커에만 있음).
- 테스트 4건 추가(ops 집계 2·포트 스냅샷 2), 전체 145 passed.

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

lps/crud/job_crud.py

@@ -217,13 +217,22 @@ class JobQueue:
                   (lease_until IS NOT NULL AND lease_until < now())
                   OR run_started_at < now() - interval '10 minutes'
               )) AS stuck_running,
-              COALESCE(EXTRACT(EPOCH FROM (now() - min(created_at) FILTER (WHERE status = 1)))::int, 0) AS oldest_pending_sec
+              COALESCE(EXTRACT(EPOCH FROM (now() - min(created_at) FILTER (WHERE status = 1)))::int, 0) AS oldest_pending_sec,
+              -- 데드라인 강제종료(크롤 행 신호). 재시도로 살아나면 dead 엔 안 잡혀 별도 집계.
+              count(*) FILTER (WHERE last_error LIKE 'JobDeadlineExceeded%'
+                  AND updated_at > now() - interval '1 hour') AS deadline_1h,
+              -- 최근 1h 완료 잡의 검색원가 합($) — 비용 폭주(리소스차단 풀림·재시도 루프) 감시.
+              COALESCE(sum((result #>> '{metrics,cost,total_usd}')::float)
+                  FILTER (WHERE status = 3 AND updated_at > now() - interval '1 hour'), 0) AS cost_1h_usd
             FROM job
         """)
         s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
         try:
-            row = (await s.execute(sql)).mappings().first()
+            snap = dict((await s.execute(sql)).mappings().first())
-            return {k: int(v) for k, v in dict(row).items()}
+            cost = float(snap.pop("cost_1h_usd") or 0)
+            snap = {k: int(v) for k, v in snap.items()}
+            snap["cost_1h_usd"] = round(cost, 4)
+            return snap
         finally:
             await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
 

lps/docs/operations.md

@@ -152,12 +152,17 @@ SELECT key, until, reason FROM search_negative ORDER BY created_at DESC;
 | `stuck` | lease 만료 RUNNING 잔존 | (0 초과 시) |
 | `db_pool` | DB 커넥션 풀 포화율(%) — 워커·API 각자 자기 풀 감시 | `LPS_ALERT_POOL_PCT`(90) |
 | `source_fail:<src>` | 소스별 최근 30분 시도 N회 이상 & 성공 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) | `LPS_ALERT_SOURCE_FAIL_30M`(5) |
+| `deadline` | 최근 1h 잡 데드라인 강제종료 수(크롤 행 반복 신호 — 재시도로 살아나면 dead 엔 안 잡힘) | `LPS_ALERT_DEADLINE_1H`(5) |
+| `cost` | 최근 1h 완료 잡 검색원가 합($) — 비용 폭주(리소스차단 풀림·재시도 루프) 감시 | `LPS_ALERT_COST_1H_USD`(1.0) |
+| `proxy_ports_low` | 가용 프록시 포트 비율(%) — 쿨다운 격리 누적, blocks 보다 먼저 우는 대규모 차단 조기 신호 | `LPS_ALERT_PORTS_LOW_PCT`(30) |
+| `budget_leak` | 최근 6h '예산 회전에도 차단된' IP 세션 수 — 현재 요청 예산이 안전하지 않다는 신호(예산 하향 검토) | `LPS_ALERT_BLOCK_SESSIONS_6H`(1) |
 
 ```
 LPS_ALERT_WEBHOOK=https://hooks.slack.com/...   # 있으면 웹훅 알림 전송(워커·API 공통)
 LPS_ALERT_COOLDOWN_MIN=30                        # 같은 룰 재발송 억제 시간
 ```
-풀 사용률은 `GET /v1/lps/ops` 의 `pool_pct`(API 프로세스 기준)로도 노출된다 — 외부 모니터 스크랩용.
+지표는 알림 없이도 `GET /v1/lps/ops` 로 노출된다(`pool_pct`·`deadline_1h`·`cost_1h_usd` 포함) — 외부 모니터 스크랩용.
+(`proxy_ports_avail`·`block_sessions_6h` 는 워커 웹훅 스냅샷에만 포함 — 프록시 상태는 워커 프로세스에만 있음)
 
 ## 5. 테스트
 

lps/tests/test_alerts.py

@@ -97,3 +97,31 @@ def test_pool_status_shape():
     assert set(st) == {"checked_out", "capacity", "pct"}
     assert st["capacity"] > 0                      # R/W 2엔진 × (pool+overflow)
     assert 0 <= st["pct"] <= 100
+
+
+# ---- 가용 프록시 포트 현황(포트 고갈 룰의 데이터) --------------------------
+
+def test_proxy_ports_snapshot_min_across_workers():
+    from config.config_models import DecodoConfig
+    from services.search.proxy import DecodoProxy
+    from worker_main import _proxy_ports_snapshot
+
+    def _proxy():
+        return DecodoProxy(DecodoConfig(host="h", username="u", password="p",
+                                        port_start=10001, port_end=10010, session_minutes=10))
+
+    class _Ad:
+        def __init__(self, proxy):
+            self._proxy = proxy
+
+    p1, p2 = _proxy(), _proxy()
+    p2.mark_burned(10001)
+    p2.mark_burned(10002)
+    avail, total = _proxy_ports_snapshot([_Ad(p1), _Ad(p2), object()])   # 프록시 없는 어댑터 혼재 OK
+    assert (avail, total) == (8, 10)               # 가장 소진된 워커(p2) 기준
+
+
+def test_proxy_ports_snapshot_none_without_proxy():
+    from worker_main import _proxy_ports_snapshot
+    assert _proxy_ports_snapshot([object()]) is None
+    assert _proxy_ports_snapshot(None) is None

lps/tests/test_job_queue.py

@@ -91,3 +91,26 @@ def test_backoff_is_exponential_capped():
     assert compute_backoff(2, base=5) == 10
     assert compute_backoff(3, base=5) == 20
     assert compute_backoff(100, base=5, cap=600) == 600
+
+
+async def test_ops_counts_deadline_and_cost(q):
+    # 완료 잡 2건의 검색원가 합산 — claim 반환 잡을 완료(순서 의존 제거)
+    for code, cost in (("b", 0.01), ("c", 0.02)):
+        await q.enqueue(JobType.SEARCH.value, {"q": code})
+        job = await q.claim("w1")
+        await q.complete(job["job_id"], "w1", {"metrics": {"cost": {"total_usd": cost}}})
+    # 데드라인 강제종료 — 재큐(PENDING)로 살아나도 deadline_1h 에 잡혀야 한다(dead 와 별개 축)
+    j1 = await q.enqueue(JobType.SEARCH.value, {"q": "a"})
+    await q.claim("w1")
+    await q.fail(j1, "w1", "JobDeadlineExceeded: 300s", backoff_sec=0)
+    snap = await q.ops()
+    assert snap["deadline_1h"] == 1
+    assert abs(snap["cost_1h_usd"] - 0.03) < 1e-9
+
+
+async def test_ops_cost_ignores_jobs_without_metrics(q):
+    jid = await q.enqueue(JobType.SEARCH.value, {"q": "x"})
+    await q.claim("w1")
+    await q.complete(jid, "w1", {"count": 3})   # metrics 없는 결과 — 합산에서 무시(0)
+    snap = await q.ops()
+    assert snap["cost_1h_usd"] == 0

lps/tests/test_lps_api.py

@@ -89,6 +89,8 @@ async def test_ops_snapshot(client, clean_jobs):
     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"):
+    for k in ("pending", "running", "done", "dead", "dead_1h", "stuck_running", "oldest_pending_sec",
+              "blocks_1h", "deadline_1h", "pool_pct"):
         assert k in j and isinstance(j[k], int)
+    assert isinstance(j["cost_1h_usd"], (int, float))
     assert j["pending"] == 1

lps/worker_main.py

@@ -114,7 +114,22 @@ async def _warmup_worker(worker_adapters, tries: int = 3, attempt_timeout: float
                     LOG.w(f"[warmup:{ad.source}] {tries}회 실패(첫 잡에서 재시도): {type(ex).__name__}")
 
 
-async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None):
+def _proxy_ports_snapshot(adapters) -> tuple[int, int] | None:
+    """워커 프록시들의 가용 포트 현황 — (최소 가용 수, 전체 포트 수). 프록시 미사용이면 None.
+    쿨다운 맵은 프록시 인스턴스(워커)별이라 가장 소진된 워커 기준(min)으로 본다."""
+    proxies = {}
+    for ad in (adapters or []):
+        p = getattr(ad, "_proxy", None)
+        if p is not None and p.enabled:
+            proxies[id(p)] = p
+    if not proxies:
+        return None
+    any_p = next(iter(proxies.values()))
+    total = any_p.port_end - any_p.port_start + 1
+    return min(p.available_ports() for p in proxies.values()), total
+
+
+async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None, ip_log=None):
     """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가
     행/좀비 워커 감지) (2) 큐/차단/DB풀/소스별 실패 지표 점검 → AlertManager 로 발화
     (룰별 쿨다운으로 스팸 방지, 조건 해소 시 회복 알림)."""
@@ -124,7 +139,12 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters
     th_lag = int(os.environ.get("LPS_ALERT_QUEUE_LAG_SEC", "300"))
     th_pool = int(os.environ.get("LPS_ALERT_POOL_PCT", "90"))
     th_srcfail = int(os.environ.get("LPS_ALERT_SOURCE_FAIL_30M", "5"))
+    th_deadline = int(os.environ.get("LPS_ALERT_DEADLINE_1H", "5"))
+    th_cost = float(os.environ.get("LPS_ALERT_COST_1H_USD", "1.0"))
+    th_ports = int(os.environ.get("LPS_ALERT_PORTS_LOW_PCT", "30"))
+    th_leak = int(os.environ.get("LPS_ALERT_BLOCK_SESSIONS_6H", "1"))
     alerts = alerts or AlertManager(origin="worker")
+    ip_log = ip_log or IpSessionLog()
     while not stop.is_set():
         try:
             with open(hb_path, "w") as f:
@@ -142,6 +162,22 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters
             await alerts.check("stuck", snap["stuck_running"] > 0, f"stuck={snap['stuck_running']}", snap)
             await alerts.check("db_pool", pool["pct"] >= th_pool,
                                f"DB 풀 포화 {pool['pct']}% (checked_out {pool['checked_out']}/{pool['capacity']})", snap)
+            await alerts.check("deadline", snap["deadline_1h"] >= th_deadline,
+                               f"잡 데드라인 강제종료 1h={snap['deadline_1h']} — 크롤 행 반복 신호", snap)
+            await alerts.check("cost", snap["cost_1h_usd"] >= th_cost,
+                               f"검색원가 1h=${snap['cost_1h_usd']} — 비용 폭주(리소스차단 풀림·재시도 루프) 점검", snap)
+            # 가용 프록시 포트 고갈 — 쿨다운 격리 누적. blocks_1h 보다 먼저 우는 대규모 차단 조기 신호.
+            ports = _proxy_ports_snapshot(adapters)
+            if ports:
+                avail, total = ports
+                snap["proxy_ports_avail"], snap["proxy_ports_total"] = avail, total
+                await alerts.check("proxy_ports_low", avail * 100 <= total * th_ports,
+                                   f"가용 프록시 포트 {avail}/{total} — 대규모 차단 진행 신호", snap)
+            # 예산 누수 — 요청 예산을 지켰는데도 차단된 IP 세션 발생 = 현재 예산이 안전하지 않다는 신호.
+            block_sessions = (await ip_log.recent_stats(360)).get("block", 0)
+            snap["block_sessions_6h"] = block_sessions
+            await alerts.check("budget_leak", block_sessions >= th_leak,
+                               f"예산 회전에도 차단된 IP 세션 6h={block_sessions} — LPS_IP_REQUEST_BUDGET 하향 검토", snap)
             # 소스별 장기 실패 — 최근 30분간 시도는 있는데 성공이 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호)
             per_source: dict[str, list[int]] = {}
             for ad in (adapters or []):