feat(BA-6155): add metadata-driven chunk-store upload session engine

Introduce the concurrency-safe upload session engine that replaces the
single-file-append TUS model. Each chunk is written to its own file under
chunks/, named by absolute byte offset; info.json is the source of truth,
updated only via atomic rename under a short fcntl.flock window. Heavy chunk
payload writes happen lock-free in per-request temp files, so multiple Storage
Proxy replicas sharing an NFS mount can upload concurrently without the TOCTOU
corruption of the stat-check-then-append model.

This commit combines what were previously two stacked PRs — the pure session
state model and the on-disk storage class — into a single cohesive unit. They
both live in tus_session.py and the model has no consumer on its own, so
splitting them across PRs hurt reviewability rather than helping it.

Contents:
- errors/upload.py: ChunkConflictError (409), UploadSessionCorruptedError (500)
- services/upload/tus_session.py:
  - ChunkRecord / SessionState (committed_offset as the largest contiguous
    prefix, missing_ranges, progress_percent, find_at_offset) / ChunkAcceptance
  - TusUploadSession: ensure_initialized, read_state, open_temp_chunk,
    commit_chunk (idempotent on duplicate (offset,length,sha256), 409 on
    conflict, no-op on already-completed), assemble, cleanup
  - stream_chunk_to_temp: bounded-memory streaming + sha256
- cleanup() keeps the completed info.json marker and reclaims only chunk files
  under the lock, so a late duplicate PATCH that raced completion on another
  replica observes status=="completed" and no-ops instead of crashing on a
  vanished temp/target (surfaced by the real-NFS multi-proxy reproduction).
- Unit tests for both the model and the on-disk/concurrency behavior.

Supersedes the separate state-model change (was BA-6154).

Resolves BA-6155.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: jopemachine

changes/11766.enhance.md

@@ -0,0 +1 @@
+Introduce `TusUploadSession`, a concurrency-safe metadata-driven chunk store for resumable (TUS) uploads, as the foundation for corruption-free uploads across storage-proxy replicas that share a filesystem.

src/ai/backend/storage/errors/__init__.py

@@ -62,6 +62,10 @@
     QuotaScopeNotFoundError,
     QuotaTreeNotFoundError,
 )
+from .upload import (
+    ChunkConflictError,
+    UploadSessionCorruptedError,
+)
 from .vfolder import (
     InvalidSubpathError,
     VFolderNotFoundError,
@@ -90,6 +94,9 @@
     "InvalidDataLengthError",
     "ServiceNotInitializedError",
     "UploadOffsetMismatchError",
+    # upload
+    "ChunkConflictError",
+    "UploadSessionCorruptedError",
     # vfolder
     "VFolderNotFoundError",
     "InvalidSubpathError",

src/ai/backend/storage/errors/upload.py

@@ -0,0 +1,49 @@
+"""
+Upload session related exceptions.
+"""
+
+from __future__ import annotations
+
+from aiohttp import web
+
+from ai.backend.common.exception import (
+    BackendAIError,
+    ErrorCode,
+    ErrorDetail,
+    ErrorDomain,
+    ErrorOperation,
+)
+
+
+class ChunkConflictError(BackendAIError, web.HTTPConflict):
+    """
+    Raised when an incoming chunk targets an offset that already holds a
+    different chunk in the upload session (409 Conflict).
+    """
+
+    error_type = "https://api.backend.ai/probs/storage/chunk-conflict"
+    error_title = "Upload Chunk Conflict"
+
+    def error_code(self) -> ErrorCode:
+        return ErrorCode(
+            domain=ErrorDomain.STORAGE_PROXY,
+            operation=ErrorOperation.UPDATE,
+            error_detail=ErrorDetail.CONFLICT,
+        )
+
+
+class UploadSessionCorruptedError(BackendAIError, web.HTTPInternalServerError):
+    """
+    Raised when the on-disk upload session metadata cannot be parsed or is
+    structurally invalid.
+    """
+
+    error_type = "https://api.backend.ai/probs/storage/upload-session-corrupted"
+    error_title = "Upload Session Corrupted"
+
+    def error_code(self) -> ErrorCode:
+        return ErrorCode(
+            domain=ErrorDomain.STORAGE_PROXY,
+            operation=ErrorOperation.READ,
+            error_detail=ErrorDetail.INTERNAL_ERROR,
+        )

src/ai/backend/storage/services/upload/__init__.py

@@ -0,0 +1,3 @@
+"""
+Chunk-based TUS upload session services.
+"""