feat(packs-lab): chunked resumable upload (offset-based) 추가

기존 single-shot POST /upload는 그대로 유지하고, 5GB+ 안정성을 위한
chunk upload 5-endpoint를 추가했다.

- POST /upload/init — mint-token jti consume + 세션 디렉토리 생성
- PUT /upload/{sid}/chunk?offset=N — offset 매칭 후 .part 파일 append
  · 불일치 시 409 + X-Current-Offset 헤더로 재개 지점 통보
- GET /upload/{sid}/status — 현재 written / expected_size 조회
- POST /upload/{sid}/complete — atomic rename + Supabase INSERT
- DELETE /upload/{sid} — 세션 중단 + 부분파일 정리

auth.py: verify_upload_token_no_consume() 추가 — chunk/complete/abort/status
는 동일 mint-token을 재사용해야 하므로 jti consume 없이 시그니처+만료만 검증.

models.py: InitUploadResponse, ChunkUploadResponse 추가.

세션 state: PACK_BASE_DIR/.uploads/{jti}/meta.json + data.part (파일시스템
영속, 단일 컨테이너 가정).

chunk 크기 상한: PACK_CHUNK_MAX_SIZE env (기본 64MB).

tests: chunk upload 시나리오 8종 — full-flow / offset mismatch / status /
abort / wrong token / incomplete complete / filename collision / host path
저장. 전체 37 테스트 pass.

CLAUDE.md: packs-lab API 표에 chunk 5-endpoint + 사용 패턴 보강.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-12 02:36:20 +09:00
parent 448dbd5f48
commit b4dd21e67a
5 changed files with 489 additions and 7 deletions

View File

@@ -55,8 +55,8 @@ def mint_upload_token(payload: dict) -> str:
return base64.urlsafe_b64encode(body).decode() + "." + sig
def verify_upload_token(token: str) -> dict:
"""업로드 토큰 검증 + jti 사용 마킹."""
def _decode_upload_token(token: str) -> dict:
"""토큰 시그니처 + 만료 + jti 존재만 검증. JTI 마킹 없음."""
try:
b64, sig = token.split(".", 1)
body = base64.urlsafe_b64decode(b64.encode())
@@ -72,13 +72,25 @@ def verify_upload_token(token: str) -> dict:
if int(time.time()) > expires_at:
raise HTTPException(status_code=401, detail="토큰 만료")
jti = payload.get("jti")
if not jti:
if not payload.get("jti"):
raise HTTPException(status_code=401, detail="jti 누락")
return payload
def verify_upload_token(token: str) -> dict:
"""업로드 토큰 검증 + jti 사용 마킹. single-shot 업로드와 chunked init에서만 사용."""
payload = _decode_upload_token(token)
jti = payload["jti"]
with _jti_lock:
if jti in _used_jti:
raise HTTPException(status_code=409, detail="이미 사용된 토큰")
_used_jti.add(jti)
return payload
def verify_upload_token_no_consume(token: str) -> dict:
"""업로드 토큰 검증만 (jti consume 없음). chunked upload chunk/complete/abort/status에 사용."""
return _decode_upload_token(token)