da9bd7750c
레퍼런스의 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>
53 lines
3.1 KiB
Python
53 lines
3.1 KiB
Python
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 체크에도 사용된다.
|
|
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"),
|
|
),
|
|
)
|