diff --git a/lps/Dockerfile b/lps/Dockerfile index 345fa0b..83b9d3f 100644 --- a/lps/Dockerfile +++ b/lps/Dockerfile @@ -11,6 +11,6 @@ COPY . . # 항상 APP_ENV=local 로 실행 → config.local.toml 사용. ENV APP_ENV=local -EXPOSE 9400 +EXPOSE 9600 CMD ["python", "web_main.py"] diff --git a/lps/README.md b/lps/README.md index 8413e49..026d8e8 100644 --- a/lps/README.md +++ b/lps/README.md @@ -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`) diff --git a/lps/common/database/model/models.py b/lps/common/database/model/models.py index 45ee030..80f70ea 100644 --- a/lps/common/database/model/models.py +++ b/lps/common/database/model/models.py @@ -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"), + ), + ) diff --git a/lps/common/enums.py b/lps/common/enums.py index 2cd248f..f601b66 100644 --- a/lps/common/enums.py +++ b/lps/common/enums.py @@ -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 결과 전송(재시도 엔진 공유) — 가벼움 diff --git a/lps/config/config.local.toml.example b/lps/config/config.local.toml.example index 4fbf00e..d185ee7 100644 --- a/lps/config/config.local.toml.example +++ b/lps/config/config.local.toml.example @@ -3,7 +3,7 @@ # 모든 서버는 APP_ENV=local 로 띄우며 이 파일을 읽는다. [WebServerConfig] server_name = "LpsServer" -port = 9400 +port = 9600 process_count = 1 is_ssl = false is_test = true diff --git a/lps/crud/job_crud.py b/lps/crud/job_crud.py new file mode 100644 index 0000000..9404e50 --- /dev/null +++ b/lps/crud/job_crud.py @@ -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) diff --git a/lps/run_local_server.sh b/lps/run_local_server.sh index 7dd33c9..64f307c 100755 --- a/lps/run_local_server.sh +++ b/lps/run_local_server.sh @@ -8,7 +8,7 @@ cd "$(dirname "$0")" # lps/ VENV=".venv" PY="$VENV/bin/python" -PORT=9400 +PORT=9600 # 1) venv + 의존성 보장 if [[ ! -d "$VENV" ]]; then 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/web_main.py b/lps/web_main.py index e312a77..310d11b 100644 --- a/lps/web_main.py +++ b/lps/web_main.py @@ -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