ops(ci): boot_smoke 스키마 어서션 max_migration 361→378 (현재 마이그 헤드)

지난 감사(361) 이후 마이그가 378(이번 publish_outbox attempts/failed 포함)까지 전진 →
boot_smoke 스키마 게이트의 하드코딩 기대값 갱신. purge/cand/uq 기대는 동일.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/api/ingest_study.py

@@ -23,6 +23,7 @@ from datetime import datetime, timezone
 from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.config import settings
@@ -66,6 +67,22 @@ class IngestBody(BaseModel):
     attempts: list[IngestAttempt]
 
 
+def _already_ingested(rows) -> dict:
+    """이미 적재된 세션들의 캐시 요약(멱등 응답). 최초 멱등체크 + 동시경합 흡수 양쪽에서 사용."""
+    return {
+        "status": "already_ingested",
+        "sessions": [
+            {
+                "topic_id": s.study_topic_id,
+                "correct": s.correct_count,
+                "wrong": s.wrong_count,
+                "unsure": s.unsure_count,
+            }
+            for s in rows
+        ],
+    }
+
+
 def _parse_answered_at(s: str | None, now: datetime) -> datetime:
     if not s:
         return now
@@ -98,18 +115,7 @@ async def ingest_attempts(
         )
     ).scalars().all()
     if existing:
-        return {
-            "status": "already_ingested",
-            "sessions": [
-                {
-                    "topic_id": s.study_topic_id,
-                    "correct": s.correct_count,
-                    "wrong": s.wrong_count,
-                    "unsure": s.unsure_count,
-                }
-                for s in existing
-            ],
-        }
+        return _already_ingested(existing)
 
     # pub_id → source_id(내부 질문 id) 해소. deleted tombstone 제외.
     pub_ids = list({a.question_pub_id for a in body.attempts})
@@ -156,73 +162,92 @@ async def ingest_attempts(
     if not by_topic:
         raise HTTPException(status_code=404, detail="해소된 attempt 없음")
 
-    summaries = []
-    for topic_id, items in by_topic.items():
-        qids = [q.id for (_, q) in items]
-        qs = StudyQuizSession(
-            user_id=user_id,
-            study_topic_id=topic_id,
-            question_ids=qids,
-            subject_distribution={},
-            status="done",
-            cursor=len(qids),
-            source="viewer",
-            client_session_uuid=body.client_session_uuid,
-            finished_at=now,
-            created_at=now,
-            updated_at=now,
-        )
-        session.add(qs)
-        await session.flush()  # qs.id
+    try:
+        summaries = []
+        for topic_id, items in by_topic.items():
+            qids = [q.id for (_, q) in items]
+            qs = StudyQuizSession(
+                user_id=user_id,
+                study_topic_id=topic_id,
+                question_ids=qids,
+                subject_distribution={},
+                status="done",
+                cursor=len(qids),
+                source="viewer",
+                client_session_uuid=body.client_session_uuid,
+                finished_at=now,
+                created_at=now,
+                updated_at=now,
+            )
+            session.add(qs)
+            await session.flush()  # qs.id
 
-        c = w = u = 0
-        for a, q in items:
-            try:
-                sel, is_corr, outcome = derive_outcome(a.selected_choice, a.is_unsure, q.correct_choice)
-            except ValueError:
-                skipped.append(a.question_pub_id)  # 선택 없고 unsure 아님 = 무효 → skip
-                continue
-            if outcome == "correct":
-                c += 1
-            elif outcome == "wrong":
-                w += 1
-            elif outcome == "unsure":
-                u += 1
-            session.add(
-                StudyQuestionAttempt(
-                    user_id=user_id,
-                    study_question_id=q.id,
-                    study_topic_id=topic_id,
-                    selected_choice=sel,
-                    correct_choice=q.correct_choice,
-                    is_correct=is_corr,
-                    outcome=outcome,
-                    quiz_session_id=qs.id,
-                    answered_at=_parse_answered_at(a.answered_at, now),
+            c = w = u = 0
+            for a, q in items:
+                try:
+                    sel, is_corr, outcome = derive_outcome(a.selected_choice, a.is_unsure, q.correct_choice)
+                except ValueError:
+                    skipped.append(a.question_pub_id)  # 선택 없고 unsure 아님 = 무효 → skip
+                    continue
+                if outcome == "correct":
+                    c += 1
+                elif outcome == "wrong":
+                    w += 1
+                elif outcome == "unsure":
+                    u += 1
+                session.add(
+                    StudyQuestionAttempt(
+                        user_id=user_id,
+                        study_question_id=q.id,
+                        study_topic_id=topic_id,
+                        selected_choice=sel,
+                        correct_choice=q.correct_choice,
+                        is_correct=is_corr,
+                        outcome=outcome,
+                        quiz_session_id=qs.id,
+                        answered_at=_parse_answered_at(a.answered_at, now),
+                    )
+                )
+            qs.correct_count, qs.wrong_count, qs.unsure_count = c, w, u
+            await session.flush()
+
+            # finalize 무수정 재생(progress/SR/pattern + 4-A/4-B enqueue). 그 후 멱등 마커.
+            summary = await finalize_session(
+                session, user_id=user_id, study_topic_id=topic_id, quiz_session_id=qs.id
+            )
+            qs.finalized_at = now
+            summaries.append(
+                {
+                    "topic_id": topic_id,
+                    "quiz_session_id": qs.id,
+                    "correct": summary.correct,
+                    "wrong": summary.wrong,
+                    "unsure": summary.unsure,
+                    "newly_correct": summary.newly_correct,
+                    "relapsed": summary.relapsed,
+                    "recovered": summary.recovered,
+                }
+            )
+
+        await session.commit()
+    except IntegrityError:
+        # 동시 같은 client_session_uuid 경합 — 상대가 먼저 commit → (client_session_uuid,
+        # study_topic_id) uq(mig376) 위반. 데이터는 안전(원자 1-tx 전체 롤백 → SR 이중 advance
+        # 없음). 승자 결과로 graceful 수렴(500 대신 already_ingested). uuid 경합이 아닌 진짜
+        # 무결성 오류면 재조회가 비어 → re-raise 로 표면화.
+        await session.rollback()
+        winner = (
+            await session.execute(
+                select(StudyQuizSession).where(
+                    StudyQuizSession.client_session_uuid == body.client_session_uuid
                 )
             )
-        qs.correct_count, qs.wrong_count, qs.unsure_count = c, w, u
-        await session.flush()
+        ).scalars().all()
+        if not winner:
+            raise
+        logger.info("study_ingest uuid=%s 동시경합 흡수 → already_ingested", body.client_session_uuid)
+        return _already_ingested(winner)
 
-        # finalize 무수정 재생(progress/SR/pattern + 4-A/4-B enqueue). 그 후 멱등 마커.
-        summary = await finalize_session(
-            session, user_id=user_id, study_topic_id=topic_id, quiz_session_id=qs.id
-        )
-        qs.finalized_at = now
-        summaries.append(
-            {
-                "topic_id": topic_id,
-                "quiz_session_id": qs.id,
-                "correct": summary.correct,
-                "wrong": summary.wrong,
-                "unsure": summary.unsure,
-                "newly_correct": summary.newly_correct,
-                "relapsed": summary.relapsed,
-                "recovered": summary.recovered,
-            }
-        )
-
-    await session.commit()
     logger.info(
         "study_ingest uuid=%s user=%s sessions=%s skipped=%s",
         body.client_session_uuid, user_id, len(summaries), len(skipped),

app/models/published.py

@@ -56,5 +56,9 @@ class PublishOutbox(Base):
         DateTime(timezone=True), default=datetime.now, nullable=False
     )
     processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # mig378: 행별 격리 재시도/terminal. attempts=savepoint 실패 누적, failed_at=MAX 초과 terminal
+    # (set 시 워커 select 에서 제외 → head-of-line block 방지).
+    attempts: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=0)
+    failed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
 
     # 미처리 부분 인덱스 idx(id) WHERE processed_at IS NULL = mig372.

app/services/study/publish_enqueue.py

@@ -68,10 +68,10 @@ async def enqueue_question_publish(session: AsyncSession, q: Any) -> None:
         await enqueue_publish(session, kind=KIND_EXPLANATION, source_id=q.id, payload=expl)
 
 
-async def backfill_publish_questions(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int:
+async def backfill_publish_questions(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> tuple[int, int]:
     """active(미삭제) 문항을 id>after_id 부터 bounded 로 outbox 적재.
 
-    반환 = enqueue 한 문항 수(0 이면 끝). 큰 셋은 마지막 id 로 페이지 반복. caller commit.
+    반환 = (enqueue 수, 마지막 처리 id). caller 는 수==limit 면 last_id 로 다음 페이지. caller commit.
     """
     rows = (
         await session.execute(
@@ -83,7 +83,7 @@ async def backfill_publish_questions(session: AsyncSession, *, after_id: int = 0
     ).scalars().all()
     for q in rows:
         await enqueue_question_publish(session, q)
-    return len(rows)
+    return len(rows), (rows[-1].id if rows else after_id)
 
 
 async def enqueue_topic_publish(session: AsyncSession, topic: Any) -> None:
@@ -91,10 +91,10 @@ async def enqueue_topic_publish(session: AsyncSession, topic: Any) -> None:
     await enqueue_publish(session, kind=KIND_TOPIC, source_id=topic.id, payload=project_topic(topic))
 
 
-async def backfill_publish_topics(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int:
+async def backfill_publish_topics(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> tuple[int, int]:
     """active(미삭제) 주제를 id>after_id 부터 bounded 로 outbox 적재(S-1 초기 백필).
 
-    반환 = enqueue 한 주제 수(0 이면 끝). 큰 셋은 마지막 id 로 페이지 반복. caller commit.
+    반환 = (enqueue 수, 마지막 처리 id). caller 는 수==limit 면 last_id 로 다음 페이지. caller commit.
     멱등 = 발행 워커의 (payload_hash, deleted) 디둡이 no-op 재투영 흡수(중복 enqueue 무해).
     """
     rows = (
@@ -107,7 +107,7 @@ async def backfill_publish_topics(session: AsyncSession, *, after_id: int = 0, l
     ).scalars().all()
     for t in rows:
         await enqueue_topic_publish(session, t)
-    return len(rows)
+    return len(rows), (rows[-1].id if rows else after_id)
 
 
 async def enqueue_card_publish(session: AsyncSession, card: Any) -> None:
@@ -123,10 +123,10 @@ async def enqueue_card_publish(session: AsyncSession, card: Any) -> None:
         await enqueue_publish(session, kind=KIND_CARD, source_id=card.id, payload=project_card(card))
 
 
-async def backfill_publish_cards(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int:
+async def backfill_publish_cards(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> tuple[int, int]:
     """검수완료(needs_review=False)·미삭제 카드를 id>after_id 부터 bounded 로 outbox 적재(S-2 초기 백필).
 
-    반환 = enqueue 한 카드 수(0 이면 끝). 멱등 = 워커 (payload_hash, deleted) 디둡. caller commit.
+    반환 = (enqueue 수, 마지막 처리 id). caller 는 수==limit 면 last_id 로 다음 페이지. 멱등 = 워커 디둡. caller commit.
     """
     rows = (
         await session.execute(
@@ -142,7 +142,7 @@ async def backfill_publish_cards(session: AsyncSession, *, after_id: int = 0, li
     ).scalars().all()
     for c in rows:
         await enqueue_card_publish(session, c)
-    return len(rows)
+    return len(rows), (rows[-1].id if rows else after_id)
 
 
 async def enqueue_card_progress_publish(session: AsyncSession, progress: Any) -> None:
@@ -155,11 +155,11 @@ async def enqueue_card_progress_publish(session: AsyncSession, progress: Any) ->
     )
 
 
-async def backfill_publish_card_progress(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> int:
+async def backfill_publish_card_progress(session: AsyncSession, *, after_id: int = 0, limit: int = 200) -> tuple[int, int]:
     """모든 card progress row 를 id>after_id 부터 bounded 로 outbox 적재(S-4 초기 백필).
 
     ★필터 없음 = ALL row(due_at NULL sentinel·terminal 포함) — due-only 백필은 sentinel 누락.
-    반환 = enqueue 한 row 수(0 이면 끝). 멱등 = 워커 디둡. caller commit.
+    반환 = (enqueue 수, 마지막 처리 id). caller 는 수==limit 면 last_id 로 다음 페이지. 멱등 = 워커 디둡. caller commit.
     """
     rows = (
         await session.execute(
@@ -171,4 +171,4 @@ async def backfill_publish_card_progress(session: AsyncSession, *, after_id: int
     ).scalars().all()
     for p in rows:
         await enqueue_card_progress_publish(session, p)
-    return len(rows)
+    return len(rows), (rows[-1].id if rows else after_id)

app/workers/classify_worker.py

@@ -608,7 +608,9 @@ async def process(
             except Exception as exc:
                 if legacy_cfg is not None and is_deferrable_error(exc):
                     raise StageDeferred(f"macbook_unavailable:{type(exc).__name__}") from exc
-                logger.warning(f"[summary-fallback] id={document_id}: {exc}")
+                # ai_summary=NULL 로 완료되면 digest/briefing 이 조용히 제외 → ERROR 로 가시화
+                # (best-effort 강등 자체는 유지, 운영 추적성만 보강).
+                logger.error(f"[summary-fallback] id={document_id} ai_summary 미생성: {exc}")
 
     finally:
         await client.close()

app/workers/csb_collector.py

@@ -140,7 +140,8 @@ async def _download_pdf(url: str, dest: Path) -> int:
     if len(resp.content) > _MAX_PDF_BYTES:
         raise FeedError(f"PDF 크기 초과 ({len(resp.content)} bytes): {url}")
     dest.parent.mkdir(parents=True, exist_ok=True)
-    dest.write_bytes(resp.content)
+    # 최대 50MB PDF write 는 동기 blocking — 이벤트루프 점유 회피 to_thread (R5 동형).
+    await asyncio.to_thread(dest.write_bytes, resp.content)
     return len(resp.content)
 
 
@@ -190,9 +191,11 @@ async def _ingest_pdf(session, page_slug: str, pdf_url: str) -> bool:
 
     dest = Path(settings.nas_mount_path) / rel_path
     size = await _download_pdf(pdf_url, dest)
+    # 50MB PDF read + sha256 는 동기 blocking(I/O+CPU) — 이벤트루프 점유 회피 to_thread (R5 동형).
+    file_hash = await asyncio.to_thread(lambda: hashlib.sha256(dest.read_bytes()).hexdigest())
     doc = Document(
         file_path=rel_path,
-        file_hash=hashlib.sha256(dest.read_bytes()).hexdigest(),
+        file_hash=file_hash,
         file_format="pdf",
         file_size=size,
         file_type="immutable",

app/workers/file_watcher.py

@@ -251,104 +251,118 @@ async def watch_inbox():
     for extra_path in settings.additional_watch_targets:
         targets.append((extra_path, "library"))
 
-    async with async_session() as session:
-        # ─── Web/ 트랙 (devonagent) — DEVONthink Smart Rule 이 떨군 .html 만 진입 ───
-        if web_root.exists():
-            # rglob NFS 디렉토리 walk(blocking stat 다발)를 off-thread 로 수집 (R5).
-            for file_path in await asyncio.to_thread(lambda: list(web_root.rglob("*.html"))):
-                if not file_path.is_file() or should_skip(file_path):
-                    continue
-                rel_path = str(file_path.relative_to(nas_root))
-                added, _ = await _ingest_web_file(session, file_path, rel_path)
+    # 파일별 독립 세션+commit 으로 격리 — 한 파일 실패(예: rglob↔stat 사이 삭제로 FileNotFoundError,
+    # flush 오류)가 watch_inbox 전체를 raise·롤백해 그 사이클 등록분을 모두 잃거나, 결정적 poison
+    # 파일이 매 사이클 같은 지점에서 중단시키는 것을 차단 (news_collector/csb_collector 와 동형).
+    # ─── Web/ 트랙 (devonagent) — DEVONthink Smart Rule 이 떨군 .html 만 진입 ───
+    if web_root.exists():
+        # rglob NFS 디렉토리 walk(blocking stat 다발)를 off-thread 로 수집 (R5).
+        for file_path in await asyncio.to_thread(lambda: list(web_root.rglob("*.html"))):
+            if not file_path.is_file() or should_skip(file_path):
+                continue
+            rel_path = str(file_path.relative_to(nas_root))
+            try:
+                async with async_session() as session:
+                    added, _ = await _ingest_web_file(session, file_path, rel_path)
+                    await session.commit()
                 new_count += added
-
-        # ─── PKM 트랙 (기존 drive_sync) ─────────────────────────────────────────
-        for sub, expected_category in targets:
-            scan_root = pkm_root / sub
-            if not scan_root.exists():
+            except Exception as e:
+                logger.warning("[Web] 파일 처리 실패 skip path=%s: %s", rel_path, e)
                 continue
 
-            # 안전 자료실 A-2/B-4 — 타깃 폴더 기반 (material, jurisdiction, license)
-            target_mt, target_jur, target_license = _TARGET_AXIS.get(
-                Path(sub).name, (None, None, None)
+    # ─── PKM 트랙 (기존 drive_sync) ─────────────────────────────────────────
+    for sub, expected_category in targets:
+        scan_root = pkm_root / sub
+        if not scan_root.exists():
+            continue
+
+        # 안전 자료실 A-2/B-4 — 타깃 폴더 기반 (material, jurisdiction, license)
+        target_mt, target_jur, target_license = _TARGET_AXIS.get(
+            Path(sub).name, (None, None, None)
+        )
+
+        # NFS 디렉토리 walk(blocking) off-thread 수집 (R5).
+        for file_path in await asyncio.to_thread(lambda: list(scan_root.rglob("*"))):
+            if not file_path.is_file() or should_skip(file_path):
+                continue
+
+            category, needs_conversion, next_stage = _route_media(
+                file_path, expected_category
             )
 
-            # NFS 디렉토리 walk(blocking) off-thread 수집 (R5).
-            for file_path in await asyncio.to_thread(lambda: list(scan_root.rglob("*"))):
-                if not file_path.is_file() or should_skip(file_path):
-                    continue
+            # audio/video 폴더에 엉뚱한 확장자가 들어왔거나 Inbox 에
+            # audio/video 가 잘못 떨어진 경우 — 이 라운드에서 아예 skip
+            if category is None and next_stage is None:
+                continue
 
-                category, needs_conversion, next_stage = _route_media(
-                    file_path, expected_category
-                )
-
-                # audio/video 폴더에 엉뚱한 확장자가 들어왔거나 Inbox 에
-                # audio/video 가 잘못 떨어진 경우 — 이 라운드에서 아예 skip
-                if category is None and next_stage is None:
-                    continue
-
-                rel_path = str(file_path.relative_to(nas_root))
+            rel_path = str(file_path.relative_to(nas_root))
+            try:
                 # GB 파일 SHA-256 은 이벤트 루프를 점유 → 같은 루프의 모든 1분 주기 consumer
                 # + FastAPI 요청이 수십초~분 동시 정지. to_thread 오프로드. 스캔 루프가 이미
                 # 순차라 file_hash 는 한 번에 하나만 실행(직렬화) — 병렬 해싱 X = NFS 2.5GbE
-                # 대역폭·버퍼 메모리 blowup 방지 (R5).
+                # 대역폭·버퍼 메모리 blowup 방지 (R5). 세션 밖에서 계산(커넥션 미점유).
                 fhash = await asyncio.to_thread(file_hash, file_path)
 
-                result = await session.execute(
-                    select(Document).where(Document.file_path == rel_path)
-                )
-                existing = result.scalar_one_or_none()
-
-                if existing is None:
-                    ext = file_path.suffix.lstrip(".").lower() or "unknown"
-                    doc = Document(
-                        file_path=rel_path,
-                        file_hash=fhash,
-                        file_format=ext,
-                        file_size=file_path.stat().st_size,
-                        file_type="immutable",
-                        title=file_path.stem,
-                        source_channel="drive_sync",
-                        category=category,
-                        needs_conversion=needs_conversion,
-                        # 안전 자료실 A-2/B-4 — watch 타깃 매핑 (KGS=law/KR 등, 비대상=NULL)
-                        material_type=target_mt,
-                        jurisdiction=target_jur,
+                async with async_session() as session:
+                    result = await session.execute(
+                        select(Document).where(Document.file_path == rel_path)
                     )
-                    # B-4 — 타깃 폴더 license 주입(restricted 포함, 비대상=미주입). classify 는
-                    # material_type IS NULL 일 때만 제안 + extract_meta 미기록이라 주입 보존.
-                    if target_license:
-                        doc.extract_meta = {"license": dict(target_license)}
-                    session.add(doc)
-                    await session.flush()
+                    existing = result.scalar_one_or_none()
 
-                    if next_stage:
-                        await enqueue_stage(session, doc.id, next_stage)
-                    new_count += 1
+                    if existing is None:
+                        ext = file_path.suffix.lstrip(".").lower() or "unknown"
+                        doc = Document(
+                            file_path=rel_path,
+                            file_hash=fhash,
+                            file_format=ext,
+                            file_size=file_path.stat().st_size,
+                            file_type="immutable",
+                            title=file_path.stem,
+                            source_channel="drive_sync",
+                            category=category,
+                            needs_conversion=needs_conversion,
+                            # 안전 자료실 A-2/B-4 — watch 타깃 매핑 (KGS=law/KR 등, 비대상=NULL)
+                            material_type=target_mt,
+                            jurisdiction=target_jur,
+                        )
+                        # B-4 — 타깃 폴더 license 주입(restricted 포함, 비대상=미주입). classify 는
+                        # material_type IS NULL 일 때만 제안 + extract_meta 미기록이라 주입 보존.
+                        if target_license:
+                            doc.extract_meta = {"license": dict(target_license)}
+                        session.add(doc)
+                        await session.flush()
 
-                elif existing.file_hash != fhash:
-                    existing.file_hash = fhash
-                    existing.file_size = file_path.stat().st_size
-                    # 기존 문서에 category/quarantine flag 가 비어있으면 보정
-                    if existing.category is None and category is not None:
-                        existing.category = category
-                    if needs_conversion and not getattr(existing, "needs_conversion", False):
-                        existing.needs_conversion = True
-                    # B-4 — 축/license 보정(B-4 이전 적재분이 재변경 시): material 미설정 시 주입,
-                    # license 부재 시에만 merge 주입(clobber 회피 — 기존 extract_meta 키 보존).
-                    if existing.material_type is None and target_mt is not None:
-                        existing.material_type = target_mt
-                        existing.jurisdiction = target_jur
-                    if target_license and not (existing.extract_meta or {}).get("license"):
-                        meta = dict(existing.extract_meta or {})
-                        meta["license"] = dict(target_license)
-                        existing.extract_meta = meta
+                        if next_stage:
+                            await enqueue_stage(session, doc.id, next_stage)
+                        await session.commit()
+                        new_count += 1
 
-                    if next_stage:
-                        await enqueue_stage(session, existing.id, next_stage)
-                    changed_count += 1
+                    elif existing.file_hash != fhash:
+                        existing.file_hash = fhash
+                        existing.file_size = file_path.stat().st_size
+                        # 기존 문서에 category/quarantine flag 가 비어있으면 보정
+                        if existing.category is None and category is not None:
+                            existing.category = category
+                        if needs_conversion and not getattr(existing, "needs_conversion", False):
+                            existing.needs_conversion = True
+                        # B-4 — 축/license 보정(B-4 이전 적재분이 재변경 시): material 미설정 시 주입,
+                        # license 부재 시에만 merge 주입(clobber 회피 — 기존 extract_meta 키 보존).
+                        if existing.material_type is None and target_mt is not None:
+                            existing.material_type = target_mt
+                            existing.jurisdiction = target_jur
+                        if target_license and not (existing.extract_meta or {}).get("license"):
+                            meta = dict(existing.extract_meta or {})
+                            meta["license"] = dict(target_license)
+                            existing.extract_meta = meta
 
-        await session.commit()
+                        if next_stage:
+                            await enqueue_stage(session, existing.id, next_stage)
+                        await session.commit()
+                        changed_count += 1
+                    # else: 무변경 → 쓰기 없음 (세션 자동 닫힘, commit 불요)
+            except Exception as e:
+                logger.warning("[PKM] 파일 처리 실패 skip path=%s: %s", rel_path, e)
+                continue
 
     if new_count or changed_count:
         logger.info(f"[Inbox+§3] 새 파일 {new_count}건, 변경 파일 {changed_count}건 등록")

app/workers/marker_worker.py

@@ -300,6 +300,11 @@ async def _process_single(
             f"[marker] transient error id={document_id} kind={type(exc).__name__}: {exc}"
         )
         raise
+    except json.JSONDecodeError as exc:
+        # 200 응답의 truncated/malformed body — 연결 흔들림 등 transient. _fail(non-retryable)
+        # 로 박지 말고 raise → queue retry (max_attempts 까지). 진짜 손상이면 재시도 후 failed.
+        logger.warning(f"[marker] malformed json body (200) id={document_id}: {exc}")
+        raise
     except Exception as exc:
         logger.exception(f"[marker] unexpected error id={document_id}: {exc}")
         await _fail(session, document_id, str(exc)[:1000])

app/workers/presegment_worker.py

@@ -497,12 +497,18 @@ async def process(document_id: int, session: AsyncSession) -> None:
         logger.warning(f"[presegment] id={document_id} file not found ({raw}) → extract")
         return
 
+    import asyncio
+
     import fitz  # PyMuPDF — extract_worker/marker_worker 와 동일 의존
 
+    def _read_toc(path: str):
+        # fitz open/get_toc 는 동기 blocking — live 스테이지라 이벤트루프(같은 루프의 1분 consumer +
+        # FastAPI 요청) 점유 회피 위해 to_thread 오프로드(거대/손상 PDF 파싱 수백 ms~초).
+        with fitz.open(path) as pdf:
+            return pdf.page_count, (pdf.get_toc(simple=True) or [])
+
     try:
-        with fitz.open(str(source)) as pdf:
-            page_count = pdf.page_count
-            toc = pdf.get_toc(simple=True) or []
+        page_count, toc = await asyncio.to_thread(_read_toc, str(source))
     except Exception as exc:
         # PDF 손상 등 — 분할 불가. 단일문서로 통과(extract 가 PyMuPDF/OCR 로 재시도하며 가시화).
         logger.warning(

app/workers/study_publish_worker.py

@@ -28,6 +28,8 @@ logger = setup_logger("study_publish_worker")
 BATCH_SIZE = 500
 # pg_advisory_xact_lock 전역 단일 라이터 키(발행 워커 전용 임의 상수, 타 advisory 락과 비충돌).
 ADVISORY_LOCK_KEY = 838201
+# 행별 격리 재시도 상한 — 초과 시 failed_at 스탬프(terminal)로 select 에서 제외.
+MAX_OUTBOX_ATTEMPTS = 5
 
 
 async def consume_publish_outbox() -> None:
@@ -46,11 +48,15 @@ async def consume_publish_outbox() -> None:
             max_rev = int(
                 (await session.execute(select(func.coalesce(func.max(Published.rev), 0)))).scalar() or 0
             )
-            # 3) 미처리 outbox 를 커밋순(id)으로.
+            # 3) 미처리 outbox 를 커밋순(id)으로. failed_at(terminal) 은 제외 — poison 행이
+            #    head-of-line 을 영구 점유하지 않게 함.
             rows = (
                 await session.execute(
                     select(PublishOutbox)
-                    .where(PublishOutbox.processed_at.is_(None))
+                    .where(
+                        PublishOutbox.processed_at.is_(None),
+                        PublishOutbox.failed_at.is_(None),
+                    )
                     .order_by(PublishOutbox.id.asc())
                     .limit(BATCH_SIZE)
                 )
@@ -60,59 +66,86 @@ async def consume_publish_outbox() -> None:
 
             now = datetime.now(timezone.utc)
             published_count = 0
+            failed_count = 0
             for ob in rows:
-                existing = (
-                    await session.execute(
-                        select(Published).where(
-                            Published.kind == ob.kind,
-                            Published.source_id == ob.source_id,
-                        )
-                    )
-                ).scalar_one_or_none()
+                try:
+                    # 행 단위 savepoint 격리 — 한 행의 예외가 배치 전체(앞 행 processed_at 포함)를
+                    # 롤백해 poison 행이 다음 사이클에 다시 최저 id 로 선택되는 무한 재선택을 차단.
+                    async with session.begin_nested():
+                        existing = (
+                            await session.execute(
+                                select(Published).where(
+                                    Published.kind == ob.kind,
+                                    Published.source_id == ob.source_id,
+                                )
+                            )
+                        ).scalar_one_or_none()
 
-                # (payload_hash, deleted) 디둡 — no-op 재투영은 rev 안 올림.
-                if (
-                    existing is not None
-                    and existing.payload_hash == ob.payload_hash
-                    and existing.deleted == ob.deleted
-                ):
-                    ob.processed_at = now
+                        # (payload_hash, deleted) 디둡 — no-op 재투영은 rev 안 올림.
+                        is_noop = (
+                            existing is not None
+                            and existing.payload_hash == ob.payload_hash
+                            and existing.deleted == ob.deleted
+                        )
+                        if is_noop:
+                            ob.processed_at = now
+                        else:
+                            new_rev = max_rev + 1
+                            if existing is None:
+                                session.add(
+                                    Published(
+                                        kind=ob.kind,
+                                        source_id=ob.source_id,
+                                        pub_id=uuid.uuid4().hex,
+                                        payload=ob.payload,
+                                        payload_hash=ob.payload_hash,
+                                        schema_version=ob.schema_version,
+                                        rev=new_rev,
+                                        deleted=ob.deleted,
+                                        created_at=now,
+                                        updated_at=now,
+                                    )
+                                )
+                            else:
+                                existing.payload = ob.payload
+                                existing.payload_hash = ob.payload_hash
+                                existing.schema_version = ob.schema_version
+                                existing.deleted = ob.deleted
+                                existing.rev = new_rev
+                                existing.updated_at = now
+                            ob.processed_at = now
+                            # 배치 내 동일 (kind, source_id) 후속 행이 직전 반영을 보도록 flush(최신 승).
+                            await session.flush()
+                except Exception as row_err:
+                    # savepoint 롤백 = 이 행의 쓰기(processed_at 포함) 취소. attempts/failed_at 만
+                    # 바깥 트랜잭션에 누적돼 최종 commit 으로 영속(영구 재선택 방지).
+                    ob.attempts = (ob.attempts or 0) + 1
+                    if ob.attempts >= MAX_OUTBOX_ATTEMPTS:
+                        ob.failed_at = now
+                        failed_count += 1
+                        logger.error(
+                            "publish_outbox_row_terminal id=%s kind=%s source_id=%s attempts=%s: %s",
+                            ob.id, ob.kind, ob.source_id, ob.attempts, row_err,
+                        )
+                    else:
+                        logger.warning(
+                            "publish_outbox_row_retry id=%s kind=%s source_id=%s attempts=%s: %s",
+                            ob.id, ob.kind, ob.source_id, ob.attempts, row_err,
+                        )
                     continue
-
-                max_rev += 1
-                if existing is None:
-                    session.add(
-                        Published(
-                            kind=ob.kind,
-                            source_id=ob.source_id,
-                            pub_id=uuid.uuid4().hex,
-                            payload=ob.payload,
-                            payload_hash=ob.payload_hash,
-                            schema_version=ob.schema_version,
-                            rev=max_rev,
-                            deleted=ob.deleted,
-                            created_at=now,
-                            updated_at=now,
-                        )
-                    )
                 else:
-                    existing.payload = ob.payload
-                    existing.payload_hash = ob.payload_hash
-                    existing.schema_version = ob.schema_version
-                    existing.deleted = ob.deleted
-                    existing.rev = max_rev
-                    existing.updated_at = now
-
-                ob.processed_at = now
-                # 배치 내 동일 (kind, source_id) 후속 행이 직전 반영을 보도록 flush(최신 승).
-                await session.flush()
-                published_count += 1
+                    # savepoint 커밋 성공 시에만 rev 카운터 전진(실패 행은 rev 미소모 → 드물게 gap,
+                    # 단일 라이터·커밋순 부여라 viewer since-rev 증분 동기 정합엔 무해).
+                    if not is_noop:
+                        max_rev = new_rev
+                        published_count += 1
 
             await session.commit()
             logger.info(
-                "publish_outbox_drained scanned=%s published=%s max_rev=%s",
+                "publish_outbox_drained scanned=%s published=%s failed=%s max_rev=%s",
                 len(rows),
                 published_count,
+                failed_count,
                 max_rev,
             )
         except Exception as e:

docker-compose.yml

@@ -3,7 +3,13 @@ services:
     image: pgvector/pgvector:pg16
     volumes:
       - pgdata:/var/lib/postgresql/data
-      - ./migrations:/docker-entrypoint-initdb.d
+      # ★ 2026-06-29 fresh-DB/DR 부팅 fix: initdb.d 마운트 제거(기존 `./migrations:/docker-entrypoint-initdb.d`).
+      # 빈 볼륨 첫 기동 시 postgres 엔트리포인트가 migrations/*.sql(001~) 을 psql autocommit 으로 실행해
+      # 스키마는 만들되 schema_migrations 스탬프는 안 남김(runner 만 생성) → fastapi init_db 가 documents
+      # 존재로 'fresh' 를 오판해 baseline(_load_baseline_if_fresh) 로드를 건너뛰고, 빈 schema_migrations
+      # 로 001 부터 재replay → `CREATE TABLE users`(IF NOT EXISTS 없음) 충돌 → 부팅 크래시(DR/신규환경).
+      # fresh-boot 은 init_db 의 baseline 적재 + migration runner 단일 경로로 일원화(설계 의도). 기존 prod
+      # 볼륨은 비어있지 않아 init scripts 가 애초에 미발동 → 무영향.
     environment:
       POSTGRES_DB: pkm
       POSTGRES_USER: pkm

docs/architecture.md

@@ -1094,7 +1094,7 @@ services:
     image: pgvector/pgvector:pg16
     volumes:
       - pgdata:/var/lib/postgresql/data
-      - ./migrations:/docker-entrypoint-initdb.d
+      # initdb.d 마운트 제거(2026-06-29): fresh-boot 은 fastapi init_db+baseline 단일 경로.
     environment:
       POSTGRES_DB: pkm
       POSTGRES_USER: pkm

docs/storage_layout.md

@@ -71,7 +71,7 @@ GPU 서버의 NFS mount (`/proc/mounts` 실측):
 
 | 컨테이너 | 마운트 | 모드 | 비고 |
 |---|---|---|---|
-| postgres | `pgdata:/var/lib/postgresql/data` + `./migrations:/docker-entrypoint-initdb.d` | rw | DB 본체 named volume |
+| postgres | `pgdata:/var/lib/postgresql/data` | rw | DB 본체 named volume (initdb.d 마운트는 2026-06-29 제거 — 아래 관찰) |
 | kordoc-service | `${NAS}/Document_Server:/documents` | **ro** | PDF/HWP parse |
 | ocr-service | `${NAS}/Document_Server:/documents` + `ocr_models:/root/.cache` | **ro** + rw | |
 | marker-service | `${NAS}/Document_Server:/documents` + `marker_models:/models` | **ro** + rw | PDF→markdown |
@@ -84,7 +84,7 @@ GPU 서버의 NFS mount (`/proc/mounts` 실측):
 **관찰**:
 - worker 컨테이너 (kordoc/ocr/marker/stt) 는 모두 NAS **read-only** 마운트 → 원본 안전.
 - fastapi 만 NAS **rw** → 업로드/preview/extracted_images 쓰기 단일 책임.
-- `./migrations` 이 postgres 의 `docker-entrypoint-initdb.d` 와 fastapi 의 `/app/migrations` 양쪽에 마운트. 단 실제 migration runner 는 fastapi `init_db()` 만 사용 (postgres init scripts 는 첫 생성 시만 실행 → 효과 X, 안전).
+- `./migrations` 은 fastapi 의 `/app/migrations` 에만 마운트. migration runner 는 fastapi `init_db()` 단일 경로. (~2026-06-29: postgres `docker-entrypoint-initdb.d` 마운트 제거. 기존엔 "첫 생성 시만 실행 → 효과 X" 로 봤으나, 빈 볼륨 첫 기동 시 postgres 가 migrations/*.sql 을 실제 실행해 스키마는 만들되 schema_migrations 스탬프를 안 남겨 → init_db 의 baseline fresh 판정을 깨고 부팅 크래시 유발. fresh-DB/DR 부팅을 init_db+baseline 단일 경로로 일원화.)
 
 ## 정책 정리
 

migrations/378_publish_outbox_attempts_failed.sql

@@ -0,0 +1,9 @@
+-- 378_publish_outbox_attempts_failed.sql
+-- (번호: 멀티세션 중 prod 가 377_domain_bucket 을 선점 → 378 로 리넘버.)
+-- publish_outbox poison row head-of-line block 차단. 발행 워커가 행별 savepoint 격리 후
+-- 예외 시 attempts++ 하고 MAX 초과 시 failed_at 스탬프(terminal) → 그 행을 select 에서 제외해
+-- 후속 발행이 막히지 않게 함. 기존 미처리 행은 attempts=0 / failed_at=NULL 로 정상 재처리.
+-- (단일 ALTER = 1 statement = asyncpg prepared 호환.)
+ALTER TABLE publish_outbox
+    ADD COLUMN IF NOT EXISTS attempts SMALLINT NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS failed_at TIMESTAMPTZ;

scripts/backfill_publish_card_progress.py

@@ -31,8 +31,8 @@ from core.database import async_session
 from models.study_memo_card_progress import StudyMemoCardProgress
 from services.study.publish_enqueue import backfill_publish_card_progress
 
-# 개인 학습툴 progress row 대비 넉넉. 도달 시 가드 경보.
-PAGE = 100000
+# 페이지 배치 크기 — after_id 루프로 전량 처리(bounded tx).
+PAGE = 5000
 
 
 async def run(dry_run: bool) -> None:
@@ -50,13 +50,17 @@ async def run(dry_run: bool) -> None:
         print("[dry-run] 적재 안 함. 실제 실행은 --dry-run 제거.")
         return
 
-    async with async_session() as session:
-        n = await backfill_publish_card_progress(session, after_id=0, limit=PAGE)
-        await session.commit()
+    total = 0
+    after = 0
+    while True:
+        async with async_session() as session:
+            n, after = await backfill_publish_card_progress(session, after_id=after, limit=PAGE)
+            await session.commit()
+        total += n
+        if n < PAGE:
+            break
 
-    print(f"\n[ok] outbox 적재 {n}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
-    if n >= PAGE:
-        print(f"[warn] PAGE({PAGE}) 도달 — progress 가 더 있을 수 있음. after_id 페이징 추가 필요.")
+    print(f"\n[ok] outbox 적재 {total}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
 
 
 def main() -> None:

scripts/backfill_publish_cards.py

@@ -31,8 +31,8 @@ from core.database import async_session
 from models.study_memo_card import StudyMemoCard
 from services.study.publish_enqueue import backfill_publish_cards
 
-# 개인 학습툴 카드 수 대비 넉넉(단일 outbox 적재 tx, 워커는 BATCH_SIZE 로 drain). 도달 시 가드 경보.
-PAGE = 100000
+# 페이지 배치 크기 — after_id 루프로 전량 처리(bounded tx). 워커는 BATCH_SIZE 로 drain.
+PAGE = 5000
 
 
 async def run(dry_run: bool) -> None:
@@ -55,13 +55,17 @@ async def run(dry_run: bool) -> None:
         print("[dry-run] 적재 안 함. 실제 실행은 --dry-run 제거.")
         return
 
-    async with async_session() as session:
-        n = await backfill_publish_cards(session, after_id=0, limit=PAGE)
-        await session.commit()
+    total = 0
+    after = 0
+    while True:
+        async with async_session() as session:
+            n, after = await backfill_publish_cards(session, after_id=after, limit=PAGE)
+            await session.commit()
+        total += n
+        if n < PAGE:
+            break
 
-    print(f"\n[ok] outbox 적재 {n}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
-    if n >= PAGE:
-        print(f"[warn] PAGE({PAGE}) 도달 — 카드가 더 있을 수 있음. after_id 페이징 추가 필요.")
+    print(f"\n[ok] outbox 적재 {total}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
 
 
 def main() -> None:

scripts/backfill_publish_topics.py

@@ -37,7 +37,7 @@ from core.database import async_session
 from models.study_topic import StudyTopic
 from services.study.publish_enqueue import backfill_publish_topics
 
-# 개인 학습툴 주제 수 대비 넉넉. 도달 시 overflow 가드가 경보.
+# 페이지 배치 크기 — after_id 루프로 전량 처리(bounded tx).
 PAGE = 5000
 
 
@@ -58,13 +58,17 @@ async def run(dry_run: bool) -> None:
         print("[dry-run] 적재 안 함. 실제 실행은 --dry-run 제거.")
         return
 
-    async with async_session() as session:
-        n = await backfill_publish_topics(session, after_id=0, limit=PAGE)
-        await session.commit()
+    total = 0
+    after = 0
+    while True:
+        async with async_session() as session:
+            n, after = await backfill_publish_topics(session, after_id=after, limit=PAGE)
+            await session.commit()
+        total += n
+        if n < PAGE:
+            break
 
-    print(f"\n[ok] outbox 적재 {n}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
-    if n >= PAGE:
-        print(f"[warn] PAGE({PAGE}) 도달 — 주제가 더 있을 수 있음. after_id 페이징 추가 필요.")
+    print(f"\n[ok] outbox 적재 {total}건 — 발행 워커가 drain(flag on 시) 하며 rev 부여.")
 
 
 def main() -> None:

scripts/ci/boot_smoke.py

@@ -106,7 +106,7 @@ async def main() -> None:
             "SELECT count(*) FROM pg_indexes WHERE indexname='uq_attempt_session_question'"))).scalar()
         mx = (await s.execute(text("SELECT max(version) FROM schema_migrations"))).scalar()
     print(f"SCHEMA OK — max_migration={mx} documents={docs} purge_col={purge} cand_qwen={cand} attempt_uq={uq}")
-    assert docs and purge == 1 and cand == 0 and uq == 1 and mx == 361, "FAIL: 기대 스키마 상태 불일치"
+    assert docs and purge == 1 and cand == 0 and uq == 1 and mx == 378, "FAIL: 기대 스키마 상태 불일치"
 
     # ── 5) /health 직접 호출 ──────────────────────────────────────────────
     health = await main.health_check()

services/mineru/server.py

@@ -59,6 +59,11 @@ MAX_IMAGES_PER_DOC = int(os.getenv("MINERU_MAX_IMAGES_PER_DOC", "200"))
 MAX_BYTES_PER_IMAGE = int(os.getenv("MINERU_MAX_BYTES_PER_IMAGE", str(10 * 1024 * 1024)))
 MAX_PAGES_HARD = int(os.getenv("MINERU_MAX_PAGES_HARD", "200"))  # 1-shot max_pages 안전장치
 
+# self-timeout — 변환/워밍이 vLLM 행으로 _engine_lock 을 영구 점유해 서비스가 wedge 되는 것을 차단.
+# (클라이언트 marker_worker 는 300s 로 포기하나 서버측 inflight 는 자동 취소 안 됨 → 서버 자체 상한 필요.)
+PARSE_TIMEOUT_S = float(os.getenv("MINERU_PARSE_TIMEOUT_S", "600"))
+WARMUP_TIMEOUT_S = float(os.getenv("MINERU_WARMUP_TIMEOUT_S", "1200"))  # 최초 모델 다운로드(~2.4GB) 여유
+
 _PRELOAD = os.getenv("MINERU_PRELOAD", "1") != "0"
 
 # ---- 엔진 상태 ---------------------------------------------------------------
@@ -68,6 +73,15 @@ _warmup_error: str | None = None
 _engine_lock = asyncio.Lock()
 
 
+def _is_engine_fatal(exc: BaseException) -> bool:
+    """OOM/CUDA 류 = 엔진 상태 오염 가능 → 재워밍 강제 대상(타임아웃은 호출측에서 별도 판정)."""
+    s = f"{type(exc).__name__} {exc}".lower()
+    return any(
+        k in s
+        for k in ("out of memory", "oom", "cuda", "cublas", "device-side", "illegal memory")
+    )
+
+
 async def _run_mineru(pdf_bytes: bytes, lang: str) -> tuple[str, list[dict]]:
     """슬라이스된 PDF bytes → (markdown, 이미지 dict 리스트). **async 엔진 경로.**
 
@@ -148,7 +162,7 @@ async def _ensure_warmup() -> None:
             page.insert_text((72, 72), "MinerU warmup.")
             warmup_bytes = doc.tobytes()
             doc.close()
-            await _run_mineru(warmup_bytes, MINERU_LANG)
+            await asyncio.wait_for(_run_mineru(warmup_bytes, MINERU_LANG), timeout=WARMUP_TIMEOUT_S)
             _warmup_done = True
             _warmup_error = None
             logger.info(f"[mineru-service] warmup done engine_version={_engine_version}")
@@ -274,6 +288,7 @@ def _serialize_images(images: list[dict], src_path: str) -> tuple[list[ConvertIm
 
 @app.post("/convert", response_model=ConvertResponse)
 async def convert(req: ConvertRequest):
+    global _warmup_done
     p = _resolve_path(req.file_path)
     if p is None or not p.is_file():
         raise HTTPException(404, detail={"code": "file_not_found", "message": req.file_path})
@@ -288,10 +303,18 @@ async def convert(req: ConvertRequest):
     async with _engine_lock:                # 실제 변환 직렬화(단일 GPU)
         start = time.monotonic()
         try:
-            md_text, raw_images = await _run_mineru(pdf_bytes, MINERU_LANG)
+            md_text, raw_images = await asyncio.wait_for(
+                _run_mineru(pdf_bytes, MINERU_LANG), timeout=PARSE_TIMEOUT_S
+            )
         except HTTPException:
             raise
         except Exception as exc:
+            # 타임아웃(엔진 행) 또는 OOM/CUDA 류면 엔진 오염 가능 → 다음 요청이 재워밍하도록 리셋.
+            # 재워밍까지 실패하면 _ensure_warmup 이 _warmup_error 설정 → /ready 503 → healthcheck
+            # 재시작으로 escalate(영구 degradation 차단). 일시 OOM 이면 재워밍 성공 후 정상화.
+            if isinstance(exc, (asyncio.TimeoutError, TimeoutError)) or _is_engine_fatal(exc):
+                _warmup_done = False
+                logger.error("[mineru-service] engine reset (timeout/fatal) path=%s: %s", p, exc)
             logger.exception(f"[mineru-service] conversion failed path={p}: {exc}")
             raise HTTPException(422, detail={"code": "conversion_failed",
                 "message": f"{type(exc).__name__}: {exc}"}) from exc