feat(lps): 작업 큐 엔진 — PostgreSQL 원자적 claim + lease + dead-letter
레퍼런스의 DB-큐 반면교사를 전부 뒤집는 큐 엔진을 구축한다. - 원자적 claim: FOR UPDATE SKIP LOCKED 서브쿼리 + 같은 UPDATE + RETURNING → 워커 다수여도 이중 할당 원천 불가(fetch/claim 미분리) - 모든 전이는 조건부 CAS(status/worker_id 가드) + RETURNING - 복구는 timeout 추측이 아닌 lease 만료 소유권 기반(reaper 회수) - 재시도 지수백오프 + max_attempts + dead-letter 를 큐에 내장(스크립트 난립 제거) - dedupe_key 부분 유니크로 활성 중복 차단, counts()로 관측(수동 psql 대체) - enums: JobStatus(PENDING/RUNNING/DONE/DEAD)·JobType(SEARCH/OUTBOX) - models: job 테이블(무FK·SMALLINT코드·TIMESTAMPTZ, claim/lease/dedupe 인덱스) - crud/job_crud: JobQueue(enqueue/claim/complete/fail/renew_lease/reap/counts) - tests: 동시 8워커 이중할당0·lease회수·재시도→dead·소유권가드 등 8건(실 lps_db) 부수: 포트 9400→9600 (negodata 9400 충돌 회피) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
민헌
parent
e27b73c018
commit
da9bd7750c
@ -11,6 +11,6 @@ COPY . .
|
||||
# 항상 APP_ENV=local 로 실행 → config.local.toml 사용.
|
||||
ENV APP_ENV=local
|
||||
|
||||
EXPOSE 9400
|
||||
EXPOSE 9600
|
||||
|
||||
CMD ["python", "web_main.py"]
|
||||
|
||||
@ -17,7 +17,7 @@
|
||||
lps/
|
||||
├── web_main.py # 진입점
|
||||
├── requirements.txt / Dockerfile / .dockerignore
|
||||
├── run_local_server.sh # 로컬 실행(대화형), 포트 9400
|
||||
├── run_local_server.sh # 로컬 실행(대화형), 포트 9600
|
||||
├── pytest.ini / conftest.py
|
||||
├── config/
|
||||
│ ├── config_loader.py / config_models.py / server_configs.py
|
||||
@ -38,7 +38,7 @@ lps/
|
||||
## 로컬 실행
|
||||
```bash
|
||||
cp config/config.local.toml.example config/config.local.toml # 값 채우기
|
||||
./run_local_server.sh # → http://localhost:9400/docs
|
||||
./run_local_server.sh # → http://localhost:9600/docs
|
||||
```
|
||||
|
||||
## 테스트
|
||||
@ -47,7 +47,7 @@ python -m pytest # tests/ (기본 healthz 스모크)
|
||||
```
|
||||
|
||||
## 포트
|
||||
- backend 9300 / agent 9500 과 겹치지 않도록 **LPS 는 9400** 사용.
|
||||
- backend 9300 / negodata 9400 / agent 9500 과 겹치지 않도록 **LPS 는 9600** 사용.
|
||||
|
||||
## 새 도메인 추가 순서 (backend 컨벤션)
|
||||
1. `common/database/model/models.py` 에 테이블 정의(+ `DBType`)
|
||||
|
||||
@ -1,8 +1,52 @@
|
||||
from sqlalchemy import Column, Index, SmallInteger, String, Text, DateTime
|
||||
from sqlalchemy.dialects.postgresql import UUID, JSONB
|
||||
from sqlalchemy.orm import declarative_base
|
||||
from sqlalchemy.sql import text
|
||||
|
||||
from common.enums import DBType
|
||||
|
||||
# 모든 ORM 모델의 베이스. insert 시 isinstance 체크에도 사용된다.
|
||||
# 도메인 테이블이 생기면 아래에 backend/common/database/model/models.py 스타일로 정의한다.
|
||||
# - @staticmethod DBType() 로 소속 논리 DB(common.enums.DBType)를 반환
|
||||
# - 코드값(status/type 등)은 SMALLINT 정수 코드(앱 enum 매핑)
|
||||
# - 소프트 삭제(deleted), created_at/updated_at 컨벤션 유지
|
||||
MAIN_BASE = declarative_base()
|
||||
|
||||
|
||||
class job(MAIN_BASE):
|
||||
"""작업 큐. PostgreSQL 을 '제대로' 큐로 쓴다 — 원자적 CAS claim + lease 소유권 + dead-letter.
|
||||
코드값(status/type)은 SMALLINT 정수 코드(common.enums 매핑), 시각은 전 구간 TIMESTAMPTZ, 무 FK.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def DBType():
|
||||
return DBType.MAIN.value
|
||||
|
||||
__tablename__ = "job"
|
||||
|
||||
job_id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
|
||||
job_type = Column(SmallInteger, nullable=False) # JobType
|
||||
status = Column(SmallInteger, nullable=False, server_default=text("1")) # JobStatus (1=PENDING)
|
||||
priority = Column(SmallInteger, nullable=False, server_default=text("100")) # 낮을수록 우선
|
||||
payload = Column(JSONB, nullable=False, server_default=text("'{}'::jsonb")) # 잡 입력
|
||||
result = Column(JSONB, nullable=True) # 잡 출력(완료 시)
|
||||
dedupe_key = Column(String(200), nullable=True) # 활성 중복 방지 키(부분 유니크)
|
||||
attempts = Column(SmallInteger, nullable=False, server_default=text("0")) # 시도 횟수(claim 시 +1)
|
||||
max_attempts = Column(SmallInteger, nullable=False, server_default=text("3"))
|
||||
run_after = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) # 이 시각 이후에만 claim(백오프)
|
||||
lease_until = Column(DateTime(timezone=True), nullable=True) # 소유권 임대 만료(reaper 회수 기준)
|
||||
worker_id = Column(String(80), nullable=True) # 현재 점유 워커
|
||||
run_started_at = Column(DateTime(timezone=True), nullable=True) # RUNNING 진입 시각(할당시각과 분리)
|
||||
last_error = Column(Text, nullable=True)
|
||||
created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"))
|
||||
updated_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()"), onupdate=text("now()"))
|
||||
|
||||
__table_args__ = (
|
||||
# claim 정렬/필터용: PENDING 중 run_after 지난 것을 priority·생성순으로
|
||||
Index("ix_job_claim", "status", "run_after", "priority", "created_at"),
|
||||
# reaper: 만료된 RUNNING lease 회수용
|
||||
Index("ix_job_lease", "status", "lease_until"),
|
||||
# 활성 중복 방지: 같은 dedupe_key 는 PENDING/RUNNING 중 하나만 존재 가능
|
||||
Index(
|
||||
"uq_job_dedupe_active",
|
||||
"dedupe_key",
|
||||
unique=True,
|
||||
postgresql_where=text("status IN (1, 2) AND dedupe_key IS NOT NULL"),
|
||||
),
|
||||
)
|
||||
|
||||
@ -49,3 +49,20 @@ class DBWRType(Enum):
|
||||
|
||||
DB_READ = 1
|
||||
DB_WRITE = 2
|
||||
|
||||
|
||||
class JobStatus(Enum):
|
||||
"""작업 큐 상태. 전이는 전부 조건부 원자 UPDATE(CAS)로만 한다.
|
||||
실패는 재시도 가능하면 PENDING(run_after=백오프)으로 되돌리고, 소진되면 DEAD(dead-letter)."""
|
||||
|
||||
PENDING = 1 # 대기(claim 가능). run_after <= now() 일 때만 실제 claim 대상
|
||||
RUNNING = 2 # 워커가 점유 중(lease_until 까지 소유). 만료 시 reaper 가 회수
|
||||
DONE = 3 # 완료
|
||||
DEAD = 4 # dead-letter — max_attempts 소진(수동 개입/알림 대상)
|
||||
|
||||
|
||||
class JobType(Enum):
|
||||
"""작업 종류. 무거운 잡(SEARCH=브라우저)과 가벼운 잡을 구분해 워커/동시성을 분리한다."""
|
||||
|
||||
SEARCH = 1 # 최저가 검색(쿠팡=브라우저) — 무거움
|
||||
OUTBOX = 2 # 외부 API 결과 전송(재시도 엔진 공유) — 가벼움
|
||||
|
||||
@ -3,7 +3,7 @@
|
||||
# 모든 서버는 APP_ENV=local 로 띄우며 이 파일을 읽는다.
|
||||
[WebServerConfig]
|
||||
server_name = "LpsServer"
|
||||
port = 9400
|
||||
port = 9600
|
||||
process_count = 1
|
||||
is_ssl = false
|
||||
is_test = true
|
||||
|
||||
174
lps/crud/job_crud.py
Normal file
174
lps/crud/job_crud.py
Normal file
@ -0,0 +1,174 @@
|
||||
"""작업 큐 CRUD — PostgreSQL 을 '제대로' 큐로 쓴다.
|
||||
|
||||
레퍼런스의 반면교사를 전부 뒤집는다:
|
||||
- 할당은 **단일 문장 원자 claim**: FOR UPDATE SKIP LOCKED 서브쿼리 + 같은 UPDATE + RETURNING.
|
||||
→ 워커/디스패처가 몇이든 같은 잡 이중 할당이 원천 불가. fetch 와 claim 을 분리하지 않는다.
|
||||
- 모든 전이는 **조건부 CAS**(WHERE 에 status/worker_id 가드) + RETURNING.
|
||||
- 복구는 timeout 추측이 아니라 **lease 만료 소유권**(reaper 가 회수).
|
||||
- 재시도/백오프/dead-letter 를 큐에 내장(스크립트 난립 제거). 외부전송도 OUTBOX 잡으로.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from common.database.db_session_manager import DB_SESSION_MNG
|
||||
from common.enums import DBType, DBWRType, JobStatus
|
||||
|
||||
|
||||
def compute_backoff(attempts: int, base: float = 5.0, cap: float = 600.0) -> float:
|
||||
"""지수 백오프(초). attempts 회 시도 후 다음 재시도까지 대기 = base * 2^(attempts-1), cap 상한."""
|
||||
return min(cap, base * (2 ** max(0, attempts - 1)))
|
||||
|
||||
|
||||
class JobQueue:
|
||||
DB = DBType.MAIN.value
|
||||
|
||||
async def _tx(self, fn):
|
||||
"""쓰기 트랜잭션(commit/rollback 은 여기서 책임)."""
|
||||
s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value)
|
||||
try:
|
||||
res = await fn(s)
|
||||
await s.commit()
|
||||
return res
|
||||
except Exception:
|
||||
await s.rollback()
|
||||
raise
|
||||
finally:
|
||||
await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)
|
||||
|
||||
# ---- 적재 -----------------------------------------------------------
|
||||
async def enqueue(self, job_type: int, payload: dict, priority: int = 100, dedupe_key: str | None = None, max_attempts: int = 3):
|
||||
"""잡 적재. dedupe_key 가 활성(PENDING/RUNNING) 중복이면 삽입 없이 None 반환."""
|
||||
sql = text("""
|
||||
INSERT INTO job (job_type, priority, payload, dedupe_key, max_attempts)
|
||||
VALUES (:t, :p, CAST(:payload AS jsonb), :dk, :ma)
|
||||
ON CONFLICT (dedupe_key) WHERE status IN (1, 2) AND dedupe_key IS NOT NULL
|
||||
DO NOTHING
|
||||
RETURNING job_id
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
row = (await s.execute(sql, {
|
||||
"t": job_type, "p": priority, "payload": json.dumps(payload),
|
||||
"dk": dedupe_key, "ma": max_attempts,
|
||||
})).first()
|
||||
return str(row[0]) if row else None
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
# ---- 원자적 claim ---------------------------------------------------
|
||||
async def claim(self, worker_id: str, lease_sec: int = 120):
|
||||
"""대기 잡 1건을 원자적으로 점유. 없으면 None.
|
||||
FOR UPDATE SKIP LOCKED 로 잠근 행을 같은 UPDATE 에서 RUNNING 으로 전이 → 이중 할당 불가."""
|
||||
sql = text("""
|
||||
UPDATE job SET
|
||||
status = 2,
|
||||
worker_id = :wid,
|
||||
lease_until = now() + make_interval(secs => :lease),
|
||||
run_started_at = now(),
|
||||
attempts = attempts + 1,
|
||||
updated_at = now()
|
||||
WHERE job_id = (
|
||||
SELECT job_id FROM job
|
||||
WHERE status = 1 AND run_after <= now()
|
||||
ORDER BY priority ASC, created_at ASC
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
RETURNING job_id, job_type, payload, attempts, max_attempts
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
row = (await s.execute(sql, {"wid": worker_id, "lease": lease_sec})).mappings().first()
|
||||
if not row:
|
||||
return None
|
||||
d = dict(row)
|
||||
d["job_id"] = str(d["job_id"])
|
||||
if isinstance(d.get("payload"), str):
|
||||
d["payload"] = json.loads(d["payload"])
|
||||
return d
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
# ---- 완료/실패 (소유권 가드) ---------------------------------------
|
||||
async def complete(self, job_id: str, worker_id: str, result: dict | None = None) -> bool:
|
||||
sql = text("""
|
||||
UPDATE job SET status = 3, result = CAST(:result AS jsonb),
|
||||
lease_until = NULL, worker_id = NULL, updated_at = now()
|
||||
WHERE job_id = :id AND status = 2 AND worker_id = :wid
|
||||
RETURNING job_id
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "result": json.dumps(result) if result is not None else None})).first()
|
||||
return row is not None
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
async def fail(self, job_id: str, worker_id: str, error: str, backoff_sec: float = 5.0) -> int | None:
|
||||
"""실패 처리. 시도 남으면 PENDING(run_after=백오프)으로 재큐, 소진되면 DEAD(dead-letter).
|
||||
전이 후 status(JobStatus 값)를 반환. 소유 불일치면 None."""
|
||||
sql = text("""
|
||||
UPDATE job SET
|
||||
status = CASE WHEN attempts >= max_attempts THEN 4 ELSE 1 END,
|
||||
run_after = CASE WHEN attempts >= max_attempts THEN run_after
|
||||
ELSE now() + make_interval(secs => :backoff) END,
|
||||
last_error = :err,
|
||||
lease_until = NULL,
|
||||
worker_id = NULL,
|
||||
updated_at = now()
|
||||
WHERE job_id = :id AND status = 2 AND worker_id = :wid
|
||||
RETURNING status
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "err": error[:2000], "backoff": backoff_sec})).first()
|
||||
return int(row[0]) if row else None
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
# ---- lease 갱신(heartbeat) / 회수(reaper) ---------------------------
|
||||
async def renew_lease(self, job_id: str, worker_id: str, lease_sec: int = 120) -> bool:
|
||||
sql = text("""
|
||||
UPDATE job SET lease_until = now() + make_interval(secs => :lease), updated_at = now()
|
||||
WHERE job_id = :id AND worker_id = :wid AND status = 2
|
||||
RETURNING job_id
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
row = (await s.execute(sql, {"id": job_id, "wid": worker_id, "lease": lease_sec})).first()
|
||||
return row is not None
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
async def reap(self) -> list[str]:
|
||||
"""만료된 lease(워커 사망 등)의 RUNNING 잡을 회수. 시도 남으면 즉시 재큐, 소진되면 DEAD.
|
||||
회수된 job_id 목록 반환."""
|
||||
sql = text("""
|
||||
UPDATE job SET
|
||||
status = CASE WHEN attempts >= max_attempts THEN 4 ELSE 1 END,
|
||||
run_after = now(),
|
||||
last_error = COALESCE(last_error, '') || ' [lease-expired reclaim]',
|
||||
lease_until = NULL,
|
||||
worker_id = NULL,
|
||||
updated_at = now()
|
||||
WHERE status = 2 AND lease_until IS NOT NULL AND lease_until < now()
|
||||
RETURNING job_id
|
||||
""")
|
||||
|
||||
async def run(s):
|
||||
rows = (await s.execute(sql)).all()
|
||||
return [str(r[0]) for r in rows]
|
||||
|
||||
return await self._tx(run)
|
||||
|
||||
# ---- 관측(관리 API/메트릭용) ---------------------------------------
|
||||
async def counts(self) -> dict[str, int]:
|
||||
"""상태별 잡 개수(관리 API·알림용). 수동 psql 스크립트를 대체한다."""
|
||||
async def run(s):
|
||||
rows = (await s.execute(text("SELECT status, count(*) FROM job GROUP BY status"))).all()
|
||||
by_val = {int(st): int(c) for st, c in rows}
|
||||
return {js.name: by_val.get(js.value, 0) for js in JobStatus}
|
||||
|
||||
return await self._tx(run)
|
||||
@ -8,7 +8,7 @@ cd "$(dirname "$0")" # lps/
|
||||
|
||||
VENV=".venv"
|
||||
PY="$VENV/bin/python"
|
||||
PORT=9400
|
||||
PORT=9600
|
||||
|
||||
# 1) venv + 의존성 보장
|
||||
if [[ ! -d "$VENV" ]]; then
|
||||
|
||||
93
lps/tests/test_job_queue.py
Normal file
93
lps/tests/test_job_queue.py
Normal file
@ -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
|
||||
@ -4,7 +4,7 @@
|
||||
# APP_ENV=dev python web_main.py # 환경 지정
|
||||
#
|
||||
# 또는 uvicorn 직접 실행:
|
||||
# uvicorn router.router:app --reload --host=0.0.0.0 --port=9400
|
||||
# uvicorn router.router:app --reload --host=0.0.0.0 --port=9600
|
||||
|
||||
import uvicorn
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user