Compare commits

..

1 Commits

+32
View File
@@ -23,6 +23,15 @@ logger = setup_logger("queue_consumer")
# pipeline.held_stages 안내 로그는 1분 사이클마다 반복하지 않고 최초 1회만.
_hold_logged = False
# PR3 후속(2026-07-03): 영구 실패 알람 — 사람이 개입해야 풀리는 상태라 Chat 웹훅 발화.
# allowlist 로 소음 제한: embed/chunk 류 대량 배치가 일제히 실패하면 문서 수만큼 알람이
# 쏟아지므로, 건당 가치가 높고 발생률이 낮은 LLM 스테이지만 기본 대상으로 한다.
_ALERT_FAIL_STAGES = {
s.strip()
for s in os.getenv("ALERT_FAIL_STAGES", "deep_summary,summarize").split(",")
if s.strip()
}
# stage별 배치 크기
# stt 는 GPU 단일 점유 + 회의 30분짜리도 가능 → 배치 1. thumbnail 은 ffmpeg subprocess 로 가벼움.
# deep_summary (PR-B B-1) 는 MLX 26B 단일 Semaphore(1) 경유 → 배치 1.
@@ -353,6 +362,8 @@ async def _process_stage(stage, worker_fn):
except Exception as e:
# 실패 처리
permanently_failed = False
doc_title = None
async with async_session() as session:
item = await session.get(ProcessingQueue, queue_id)
if not item:
@@ -363,7 +374,16 @@ async def _process_stage(stage, worker_fn):
item.error_message = err_text[:500]
if item.attempts >= item.max_attempts:
item.status = "failed"
permanently_failed = True
logger.error(f"[{stage}] document_id={document_id} 영구 실패: {e}")
if stage in _ALERT_FAIL_STAGES:
# 알람용 제목 best-effort — 실패해도 알람 자체는 발화한다.
try:
from models.document import Document
_doc = await session.get(Document, document_id)
doc_title = getattr(_doc, "title", None)
except Exception:
doc_title = None
# B3: marker_worker 는 변환 시작 시 doc.md_status='processing' 으로 표시한다.
# 변환이 _fail()/_set_skipped() 를 거치지 않고 예외로 죽으면(예: 대형
# batch ReadTimeout) doc.md_status 가 'processing' 에 영구 고착 = orphan
@@ -385,6 +405,18 @@ async def _process_stage(stage, worker_fn):
item.started_at = None
logger.warning(f"[{stage}] document_id={document_id} 재시도 예정 ({item.attempts}/{item.max_attempts}): {e}")
await session.commit()
if permanently_failed and stage in _ALERT_FAIL_STAGES:
# 영구 실패 = 무인 파이프라인이 스스로 못 푸는 상태 → 유인 전환 알람.
# send_alert 는 절대 raise 하지 않음(no-op/실패 = False 반환뿐).
from services.alerts import send_alert
await send_alert(
f"[DS] {stage} 영구 실패 — doc {document_id}",
(
f"{doc_title or '(제목 미상)'}\n"
f"에러: {err_text[:300]}\n"
f"확인: scripts/presegment_attended.py list (보류/거부 사유) 또는 큐 재큐"
),
)
async def consume_queue():