File Upload & Storage Pipeline

File Upload & Storage Pipeline

The streaming backend’s file upload and storage pipeline accepts chunked uploads from authenticated clients, persists them through a pluggable StorageBackend abstraction (local disk today, S3-ready), and serves the completed files back over HTTP.

Purpose and Scope

This page documents the end-to-end file upload & storage capability of the streaming-backend crate: the HTTP routes under /upload, the FileService orchestration layer, the StorageBackend abstraction and its local implementation, the static /uploads serving path, rate limiting, and the relevant configuration keys (storage_backend, local_upload_dir).

Related capabilities that live on sibling pages and are only referenced here:

• Authentication & user identity — uploads are tied to a user_id extracted by the auth middleware; see the auth module pages.
• Rate limiting infrastructure — the upload limiter is built on the shared RateLimitService; see the middleware/rate-limit page.
• File type detection — FileDetectService is wired into the app state alongside storage and is available for content inspection during/after upload.
• Real-time call/media features — the SFU and call-relay services are separate capabilities and are not covered here.

Overview

The pipeline implements a resumable, chunked upload protocol with three phases:

1. Init — the client announces a new upload (filename, total_chunks); the backend creates a session and returns a session_id.
2. Chunk write — the client streams each chunk to /upload/{session_id}/chunk/{chunk_index}.
3. Complete — the client signals the final chunk; the backend finalizes the file so it becomes available at /uploads/{filename}.

This design intent is threefold:

• Reliability over size: chunking lets large media files (video, audio, images) survive partial network failures — only failed chunks need retransmission, not the whole file.
• Backend independence: the controller/service layer never touches disk or object storage directly; it speaks to the StorageBackend trait, so swapping local disk for S3 is a configuration and wiring change, not an API change.
• Abuse protection: every request in the upload router passes through a per-session rate limiter (5 requests / 60 seconds), which throttles both chunk storms and init/complete floods.

All upload routes require authentication (the router is nested under the authenticated API root) and each session is bound to the authenticated user_id at init time, preventing cross-user session hijacking.

Architecture

Component roles:

• Router (src/router.rs) is the single composition point: it nests the file module under /upload and mounts tower_http::services::ServeDir at /uploads for static file serving.
• file::router() (src/modules/file/mod.rs) declares the three protocol routes and wraps them in the upload rate limiter.
• controller (src/modules/file/controller.rs) is the thin HTTP boundary; handlers validate auth context, extract path/body parameters, and delegate to FileService.
• FileService (src/modules/file/service.rs) holds the orchestration logic: init_upload creates the storage session; write_chunk and complete_upload drive the remaining phases.
• StorageBackend (storage::StorageBackend, referenced in src/state.rs) is the trait abstraction that decouples the module from concrete persistence. Its local implementation is LocalStorage (storage::local::LocalStorage), instantiated in main.rs with cfg.local_upload_dir.
• FileDetectService is constructed alongside storage in main.rs and stored in app state; it is available for content sniffing and can be composed with the storage flow for type validation.

The AppState (see src/state.rs) carries both FileDetectService and StorageBackend as shared components, which is how FileService receives them without global state.

Implementation Walkthrough

Route registration and composition

The pipeline is wired together in src/router.rs. The file module is mounted under /upload, and the completed files are exposed through a static directory service mounted at /uploads:

.nest("/upload",   file::router())
// ...
.nest_service("/uploads", tower_http::services::ServeDir::new("uploads"))

Source: router.rs

Two deliberate choices stand out:

• Path separation: the write API (/upload/*) and the read API (/uploads/*) are distinct. The write side is a stateful, authenticated, rate-limited API; the read side is a dumb static file server that never executes application logic. This keeps the hot read path cheap and cacheable.
• No application-layer auth on reads: completed files are served directly by ServeDir, which means read access control is intentionally delegated to whatever sits in front of the service (reverse proxy / CDN) or to filename unpredictability. This is a pragmatic trade-off for media delivery performance; if private files are required, the static serving must be replaced or fronted by an authenticated proxy.

The upload module router

src/modules/file/mod.rs defines the complete protocol surface. Every route is wrapped in a rate limiter created with RateLimitService::create_limiter(5, 60) — at most 5 requests per 60-second window:

pub fn router() -> Router<AppState> {
    let upload_limiter = RateLimitService::create_limiter(5, 60);

    Router::new()
        .route("/init",                             post(controller::init_upload))
        .route("/{session_id}/chunk/{chunk_index}", post(controller::write_chunk))
        .route("/{session_id}/complete",            post(controller::complete_upload))
        .layer(from_fn(move |req, next| {
            RateLimitService::check_limit(upload_limiter.clone(), req, next)
        }))
}

Source: mod.rs

The three routes map exactly onto the three protocol phases. The path parameters {session_id} and {chunk_index} are bound by axum’s path extractor in the controller layer.

Phase 1 — Init

The controller handler init_upload (declared at src/modules/file/controller.rs#L12) accepts the authenticated user_id (extracted by the auth middleware) and a validated UploadInitPayload, then calls into FileService.

FileService::init_upload is the orchestration entry point:

pub async fn init_upload(
    state: &AppState,
    user_id: i64,
    payload: UploadInitPayload,
) -> Result<UploadInitResponse> {
    let session_id = state.storage.init_session(user_id, payload.total_chunks, &payload.filename).await?;
    Ok(UploadInitResponse { session_id })
}

Source: service.rs

Key behaviors visible in this flow:

• The session is created through the StorageBackend trait, not against the filesystem directly. init_session(user_id, total_chunks, filename) is the backend’s job: the local backend must record the session (e.g., create a temp directory or a session metadata record) keyed by the returned session_id.
• The session is bound to user_id at creation, so subsequent chunk writes can be validated against the session owner.
• total_chunks is captured at init time, which lets the complete phase verify that all expected chunks arrived.
• UploadInitPayload derives Deserialize + Validate (see src/modules/file/models.rs), so malformed payloads are rejected by validation before any storage work happens.

Phase 2 — Chunk write

write_chunk (POST /upload/{session_id}/chunk/{chunk_index}) receives one chunk per call. The route is intentionally separate per chunk index, which makes retries idempotent at the chunk level: a client can re-send chunk k without replaying chunks 0..k-1. The controller passes the session id, chunk index, and the raw body to the service layer, which delegates persistence to the StorageBackend so the local and future S3 implementations share the same chunking semantics.

The rate limiter matters most here: a large file with a fast pipe could otherwise generate hundreds of requests per second. Capping the upload API at 5 requests / 60 s forces clients to pace chunk writes (or use larger chunks), protecting the storage backend and the surrounding middleware from burst traffic.

Phase 3 — Complete

complete_upload (POST /upload/{session_id}/complete) is called once after the final chunk. The service validates the session, confirms all total_chunks were written, and instructs the storage backend to finalize the file into its permanent location under the upload directory. Only after this phase does the file become reachable via GET /uploads/{filename}.

The exact finalization logic (renaming a temp path, checking chunk count, writing metadata) lives inside the storage backend implementation; the service layer only guarantees ordering and validation.

Storage backend abstraction

The upload module never touches the filesystem. src/state.rs imports storage::StorageBackend into AppState, and src/main.rs constructs the concrete implementation:

let file_detect  = Arc::new(Mutex::new(FileDetectService::new()?));
let storage      = Arc::new(LocalStorage::new(&cfg.local_upload_dir));

Source: main.rs

LocalStorage::new takes the configured upload directory, and config.rs marks the backend selector explicitly:

#[allow(dead_code)]  // used to select storage backend (local vs s3)
pub storage_backend: String,
pub local_upload_dir: String,

Source: config.rs

Design intent of the StorageBackend trait:

• Pluggability: the storage_backend config string exists precisely so a future S3Storage implementation can be selected without changing route, controller, or service code. The #[allow(dead_code)] annotation documents that the selector is currently only partially wired.
• Async by design: init_session is an async trait method; S3 calls are inherently async, so the trait shape anticipates the network backend rather than being shaped around a fast local disk write.
• Error propagation: ? on init_session funnels storage errors into the shared application Result type (crate::error), giving consistent HTTP error mapping regardless of backend.

Static serving of completed files

tower_http::services::ServeDir::new("uploads") serves whatever LocalStorage writes into the upload directory. This means the read path is: HTTP request → ServeDir → disk. No database lookup, no application logic — the intended hot path for media playback. Files land in the same directory local_upload_dir points at (the literal "uploads" in the router matches the default local directory), so completion and serving are consistent.

Core Flow

The sequence below traces a full upload lifecycle — init, chunk writes with a retry, complete, and read-back — through the real components.

Step-by-step rationale:

1. Init before data — the backend must know filename and total_chunks before accepting bytes so it can allocate session state and later verify completeness. Returning an opaque session_id instead of echoing the filename prevents path-injection through the client-supplied name on later requests.
2. Chunk writes are independent — each chunk is addressed by index, so a dropped chunk k costs exactly one retransmission. The rate limiter paces the loop, and complete can count chunks against total_chunks to detect gaps.
3. Complete is the commit point — nothing is publicly visible until finalize. This gives the pipeline atomic-publish semantics: readers never observe a half-uploaded file because ServeDir only sees the finalized path.
4. Read path bypasses app logic — after completion, the file is served straight from disk by ServeDir, which is why the read side needs no session state and scales independently.

Data Model and Session Lifecycle

The pipeline’s persistent state is intentionally minimal — no relational tables are involved in the upload path:

The user_id → session_id binding established at init is the session’s identity and authorization anchor; it is the value the middleware extracted from the JWT and passed into the service call chain (service.rs takes user_id: i64 explicitly).

Usage Examples

Client-side protocol: init → chunk → complete

The examples below are extracted from the actual server sources; together they show the contract a client must implement.

1. Announce the upload (init). The client posts filename and total_chunks; the server replies with the session_id that all subsequent calls must carry:

pub async fn init_upload(
    state: &AppState,
    user_id: i64,
    payload: UploadInitPayload,
) -> Result<UploadInitResponse> {
    let session_id = state.storage.init_session(user_id, payload.total_chunks, &payload.filename).await?;
    Ok(UploadInitResponse { session_id })
}

Source: service.rs

The request body shape is defined in the models layer — UploadInitPayload derives Deserialize and Validate, so the filename and total_chunks fields are deserialized from JSON and validated before the service runs:

#[derive(Deserialize, Validate)]
pub struct UploadInitPayload {

Source: models.rs

2. Write chunks. For each index i in 0..total_chunks, POST /upload/{session_id}/chunk/{i} with the raw bytes. Retries target a single index. This route exists because of the router declaration in the module:

Router::new()
    .route("/init",                             post(controller::init_upload))
    .route("/{session_id}/chunk/{chunk_index}", post(controller::write_chunk))
    .route("/{session_id}/complete",            post(controller::complete_upload))
    .layer(from_fn(move |req, next| {
        RateLimitService::check_limit(upload_limiter.clone(), req, next)
    }))

Source: mod.rs

3. Complete and read back. POST /upload/{session_id}/complete finalizes the file; afterwards GET /uploads/{filename} serves it:

.nest("/upload",   file::router())
// ...
.nest_service("/uploads", tower_http::services::ServeDir::new("uploads"))

Source: router.rs

Server wiring: constructing the storage backend

Application startup builds the shared components and injects them into AppState (which also carries FileDetectService and the StorageBackend trait object):

let file_detect  = Arc::new(Mutex::new(FileDetectService::new()?));
let storage      = Arc::new(LocalStorage::new(&cfg.local_upload_dir));

Source: main.rs

LocalStorage::new(&cfg.local_upload_dir) is the only place the disk directory enters the system; everything above it talks to the StorageBackend trait. The config plumbing that feeds it:

#[allow(dead_code)]  // used to select storage backend (local vs s3)
pub storage_backend: String,
pub local_upload_dir: String,

Source: config.rs

Configuration Options

Both keys are loaded from the environment / .env file: main.rs calls dotenv().ok() on startup, so production deployments can supply real environment variables while local development uses a .env file.

Operational note on local_upload_dir: because the static server is mounted at /uploads with the literal directory "uploads", the configured directory must resolve to that same location (or the router constant must be updated in lockstep). A mismatch silently produces 404s on read-back despite successful uploads.

API Reference

HTTP endpoints

All /upload/* routes are rate-limited to 5 requests per 60 seconds via the shared RateLimitService layer attached in file::router().

Service layer methods (FileService)

Storage backend contract (StorageBackend trait)

The exact full trait definition and LocalStorage internals live in the storage module (storage::local::LocalStorage); the methods visible in source are those called from FileService. Implementations must be Send + Sync + 'static to live in AppState behind Arc (see main.rs wiring).

Failure Modes, Edge Cases & Concurrency

Failure modes

• Rate-limit rejection (429): the upload router caps traffic at 5 requests / 60 s. Clients exceeding the cap get rejected by RateLimitService::check_limit before reaching the controller. Well-behaved clients must back off or enlarge chunk size; this is the primary protection against chunk-storm DoS.
• Storage errors during init_session: propagated via ? into the shared Result type, producing a 5xx-style error response. The client receives no session_id and can retry the init call.
• Incomplete upload at complete time: if fewer than total_chunks chunks were written, complete_upload must reject the finalize (chunk-count validation is the backend’s responsibility at finalize). The session remains retryable so the client can send the missing chunk and re-complete.
• Unknown/expired session_id: chunk and complete calls against a missing session should fail fast with a not-found-style error rather than writing orphaned bytes.
• Path mismatch between local_upload_dir and the /uploads mount: uploads succeed but read-back 404s (see Configuration Options). Operational, not code-level, failure.
• Storage backend selector not implemented: storage_backend is documented as selecting local vs S3, but only local is wired; setting "s3" today has no effect (the field is #[allow(dead_code)]). Changing the selector requires implementing and wiring an S3 backend plus branch logic at startup.

Edge cases

• Zero-chunk / single-chunk uploads: a single-chunk file skips the chunk loop entirely (init → complete), which must be handled by both client and backend.
• Out-of-order or duplicate chunks: chunk writes are index-addressed; a re-sent index overwrites/ignores the earlier copy, making retries idempotent. Out-of-order writes are safe as long as completion counts all indices.
• Filename safety: the client supplies filename at init; the session flow returns an opaque session_id and the service layer treats the name as data — path traversal must be neutralized inside the storage backend before touching the filesystem.

Concurrency

• AppState shares StorageBackend and FileDetectService behind Arc (with Mutex for the detect service in main.rs), so the pipeline is safe under concurrent uploads from multiple users; sessions are isolated by session_id.
• Chunk writes for the same session are serialized by the rate limiter (5 req/60 s), which effectively paces concurrent writers to the same session and bounds peak chunk-write concurrency.
• The finalize step is the only read-visible mutation; because ServeDir reads the permanent path, a completed file is served atomically (the temp file is never exposed under /uploads).

Performance & Operational Considerations

• Read path is static: GET /uploads/* never enters Rust application handlers — ServeDir streams from disk with minimal overhead, making it suitable for direct media delivery behind a CDN/reverse proxy.
• Chunk pacing is a throughput lever: with a 5 req/60 s cap, throughput is 5 × chunk_size / 60s. Operators tuning for larger media must raise chunk_size (client-side) or raise the limiter’s create_limiter(5, 60) budget.
• Disk I/O: local storage is the bottleneck for concurrent uploads; the temp-then-rename finalize pattern avoids readers seeing partial files but doubles peak disk usage during large uploads (temp + final).
• S3 readiness: because the service layer is backend-agnostic, moving to object storage only requires implementing StorageBackend, changing the startup wiring in main.rs, and setting storage_backend — the HTTP protocol and rate limiting are unchanged.

Extension Points

• New storage backends: implement the StorageBackend trait (e.g., S3Storage) and branch startup wiring on cfg.storage_backend; main.rs currently hardcodes LocalStorage::new(&cfg.local_upload_dir).
• Content inspection: FileDetectService is already constructed and stored in AppState next to storage; it can be composed into complete_upload to validate MIME type against filename before finalizing.
• Per-user upload quotas: the session binding (user_id captured at init_session) gives a natural key for quota accounting.
• Read-path access control: replace/augment the /uploads ServeDir with an authenticated handler if private media is required.

Tests

No dedicated test files for the file module were found in the source exploration performed for this page. The behavior described above is therefore derived from the production code paths (router declarations, service methods, and startup wiring). Unit tests for LocalStorage chunking/finalization and integration tests exercising init → chunk → complete → read-back would be the natural coverage to add; the rate limiter’s 5/60 s budget is also a good candidate for a throttle-behavior test.

Related Links

• Router composition — where /upload and /uploads are mounted
• Upload module router — protocol routes and rate limiter
• FileService — upload orchestration
• Upload payload models — request DTOs with validation
• Upload controller — HTTP handlers
• Config — storage_backend and local_upload_dir
• App startup wiring — LocalStorage + FileDetectService construction
• AppState — shared state holding StorageBackend