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

[jopemachine]

📚 Stacked PRs

This PR is part of a 5-PR stack implementing BA-3974 (epic: BA-6153). Merge in order:

1. 👉 feat(BA-6155): add metadata-driven chunk-store upload session engine #11766 — feat(BA-6155): add Valkey-backed chunk-store upload session engine ← you are here
2. ⬇️ fix(BA-6156): rewire TUS PATCH/HEAD to chunk-based store #11767 — fix(BA-6156): rewire TUS PATCH/HEAD to chunk-based store (actual user-visible fix)
3. ⬇️ test(BA-6157): add multi-proxy NFS race regression test #11768 — test(BA-6157): add multi-proxy NFS race regression test
4. ⬇️ feat(BA-6158): support TUS Checksum extension #11769 — feat(BA-6158): support TUS Checksum extension
5. ⬇️ feat(BA-6159): add /upload/status endpoint and progress headers #11770 — feat(BA-6159): add /upload/status endpoint + progress headers

Summary

Introduces the concurrency-safe TUS upload session engine. Per-session metadata
is the source of truth in Valkey (keyed by session id) and is guarded by a
per-session distributed lock produced by an injected DistributedLockFactory.
Chunk payload bytes live as individual files under chunks/chunk_<offset>.dat
on the shared filesystem; they are content-addressed by (offset, sha256) and
idempotent, so the heavy write path needs no coordination at all. Only the small
metadata read-modify-write window is serialized — and across Storage Proxy
replicas, since the lock is distributed.

Note: This PR combines two originally stacked issues — the pure
session-state model (BA-6154) and the on-disk storage engine (BA-6155). They
live in the same package and the model has no consumer on its own, so splitting
them across PRs hurt reviewability. Both issues are resolved here.

Architecture

The handler (PR #11767) only adapts the request body; all chunk file I/O and
metadata management lives in the engine (this PR).

flowchart TB
    classDef pr66 fill:#e6f7ff,stroke:#1890ff,color:#000
    classDef pr67 fill:#fff7e6,stroke:#fa8c16,color:#000
    classDef redis fill:#fff1f0,stroke:#cf1322,color:#000
    classDef disk fill:#f6ffed,stroke:#52c41a,color:#000

    subgraph L1["PATCH handler (PR #11767)"]
        REQ["request.content"]:::pr67
        ADP["TusChunkUploadStreamReader"]:::pr67
        REQ --> ADP
    end

    subgraph L2["Upload engine (PR #11766)"]
        WTC["write_temp_chunk"]:::pr66
        CC["commit_chunk"]:::pr66
        FIN["assemble + cleanup"]:::pr66
        WTC --> CC --> FIN
    end

    subgraph L3R["Valkey · source of truth"]
        STATE["tus.upload.session:id"]:::redis
        LOCK["tus.upload.lock:id"]:::redis
    end

    subgraph L3F["shared FS · chunk payloads"]
        TMP["chunk_off.tok.tmp"]:::disk
        DAT["chunk_off.dat"]:::disk
        OUT["assembled file"]:::disk
    end

    ADP --> WTC
    WTC -->|"lock-free write"| TMP
    CC -->|"acquire"| LOCK
    CC -->|"get / set"| STATE
    CC -->|"atomic rename"| DAT
    FIN -->|"concat in order"| OUT
    FIN -->|"mark COMPLETED"| STATE
    FIN -->|"unlink chunks"| DAT

Loading

Data model

classDiagram
    class TusUploadSession {
        +ensure_initialized()
        +read_state() SessionState
        +open_temp_chunk(offset) Path
        +write_temp_chunk(offset, reader) WrittenChunk
        +commit_chunk(offset, path, length, sha256) ChunkCommitResult
        +assemble(target)
        +cleanup()
    }
    class SessionState {
        +session_id
        +total_size
        +committed_chunks : ChunkMetadata[]
        +status : UploadStatus
        +committed_offset
        +find_at_offset(offset)
        +append_chunk(chunk)
    }
    class ChunkMetadata {
        +offset
        +length
        +sha256
    }
    class WrittenChunk {
        +path
        +length
        +sha256
    }
    class ChunkCommitResult {
        +state : SessionState
        +committed : bool
        +is_final_commit : bool
    }
    class TusUploadSessionArgs {
        +session_dir
        +session_id
        +total_size
        +valkey_client : ValkeyTusClient
        +lock_factory : DistributedLockFactory
    }
    SessionState o-- ChunkMetadata
    TusUploadSession ..> SessionState
    TusUploadSession ..> ChunkCommitResult
    TusUploadSession ..> WrittenChunk
    TusUploadSession ..> TusUploadSessionArgs

Loading

One PATCH lifecycle

The expensive body write is lock-free; only the short metadata read-modify-write
is serialized via the distributed lock, so concurrent replicas never corrupt
each other.

sequenceDiagram
    autonumber
    participant H as tus_upload_part
    participant S as TusUploadSession
    participant V as Valkey
    participant D as shared FS

    H->>S: ensure_initialized()
    S->>V: acquire lock + SET state if missing
    H->>S: write_temp_chunk(offset, reader)
    loop async for chunk in reader.read()
        S->>D: write chunk_off.tok.tmp (lock-free) + sha256
    end
    S-->>H: WrittenChunk(path, length, sha256)
    H->>S: commit_chunk(offset, path, length, sha256)
    Note over S,V: distributed lock window (short)
    alt status == COMPLETED
        S-->>H: no-op (late duplicate from another replica)
    else duplicate (same offset, length, sha256)
        S-->>H: committed=False (idempotent)
    else conflict (same offset, different content)
        S-->>H: ChunkConflictError 409
    else new chunk
        S->>D: atomic rename tmp to chunk_off.dat
        S->>V: update SessionState with new record
        S-->>H: ChunkCommitResult(is_final_commit?)
    end
    opt is_final_commit
        H->>S: assemble(target)
        S->>D: concat chunks in offset order
        S->>V: mark status=COMPLETED
        H->>S: cleanup()
        S->>D: unlink chunk files (state kept for TTL)
    end
    H-->>H: 204 No Content

Loading

Locking & HA semantics

Session state is NOT held in memory and NOT on the shared filesystem.
TusUploadSession is stateless in-process — every operation reads/writes the
SessionState in Valkey, and chunk payload files are content-addressed.

This makes it safe for the production topology of multiple storage-proxy
processes/replicas sharing an NFS mount: no in-process state to synchronize,
no dependency on filesystem lock semantics, no dependency on NFS attribute-cache
coherence.

How replicas coordinate:

• Source of truth = Valkey, fetched per request (no in-process cache).
• Per-session distributed lock (RedisLock via DistributedLockFactory)
  serializes the metadata read-modify-write window. The factory pattern mirrors
  the manager's DistributedLockFactory — the lock backend lives in
  RootContext.tus_lock_factory and is injected into the engine.
• Atomic rename (Path.replace) promotes a temp chunk file
  (chunk_<offset>.<token>.tmp) to its canonical name (chunk_<offset>.dat).
  Readers see old-or-new, never a partial file.
• Per-offset content-addressed chunk files — duplicates are idempotent
  no-ops, mismatched content surfaces as a 409 ChunkConflictError.
• Stateless JWT upload tokens — any PATCH may land on any replica with no
  session affinity, and still resolves to the same Valkey key and same chunk
  file path.

Stale or abandoned sessions are reclaimed by the Valkey state's TTL (24h), so
there is no separate GC sweep.

Resolves BA-6154, BA-6155.