ca7f057e41
협의로 선정한 조기 신호 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>
117 lines
4.7 KiB
Python
117 lines
4.7 KiB
Python
"""작업 큐 엔진 테스트 — 원자적 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
|
|
|
|
|
|
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
|