Hyungi Ahn
751cdc5be8
기존 UNIQUE(document_id, stage, status)는 pending+processing 동시 존재를 허용해서 stale 복구 시 충돌 발생. 2-layer 방어로 근본 차단: 1) DB: partial unique index uq_queue_active — 활성 행(pending/processing)은 (document_id, stage)당 최대 1개만 허용 2) App: enqueue_stage() 중앙 함수 — INSERT ON CONFLICT DO NOTHING으로 모든 9개 경로의 check-then-insert TOCTOU race 제거 migration 117은 guard check 포함 — 활성 중복이 남아있으면 RAISE EXCEPTION 으로 중단, 수동 정리 유도. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""processing_queue 테이블 ORM (비동기 가공 큐)"""
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import BigInteger, DateTime, Enum, ForeignKey, SmallInteger, Text, text
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from core.database import Base
|
|
|
|
|
|
class ProcessingQueue(Base):
|
|
__tablename__ = "processing_queue"
|
|
|
|
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
|
|
document_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("documents.id"), nullable=False)
|
|
stage: Mapped[str] = mapped_column(
|
|
Enum("extract", "classify", "summarize", "embed", "chunk", "preview", name="process_stage"), nullable=False
|
|
)
|
|
status: Mapped[str] = mapped_column(
|
|
Enum("pending", "processing", "completed", "failed", name="process_status"),
|
|
default="pending"
|
|
)
|
|
attempts: Mapped[int] = mapped_column(SmallInteger, default=0)
|
|
max_attempts: Mapped[int] = mapped_column(SmallInteger, default=3)
|
|
error_message: Mapped[str | None] = mapped_column(Text)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), default=datetime.now
|
|
)
|
|
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
|
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
|
|
|
|
# DB 제약은 partial unique index uq_queue_active로 관리 (migration 117)
|
|
|
|
|
|
async def enqueue_stage(
|
|
session: AsyncSession, document_id: int, stage: str, *, status: str = "pending",
|
|
) -> bool:
|
|
"""ProcessingQueue에 행 추가 (DB 레벨 중복 방어).
|
|
|
|
같은 (document_id, stage)에 활성 행(pending/processing)이 이미 있으면
|
|
아무것도 하지 않고 False 반환.
|
|
"""
|
|
stmt = (
|
|
pg_insert(ProcessingQueue)
|
|
.values(document_id=document_id, stage=stage, status=status)
|
|
.on_conflict_do_nothing(
|
|
index_elements=["document_id", "stage"],
|
|
index_where=text("status IN ('pending', 'processing')"),
|
|
)
|
|
)
|
|
result = await session.execute(stmt)
|
|
return result.rowcount > 0
|