diff --git a/docs/delta-upload-protocol.md b/docs/delta-upload-protocol.md new file mode 100644 index 00000000..31fd58ff --- /dev/null +++ b/docs/delta-upload-protocol.md @@ -0,0 +1,139 @@ +# Delta-Upload Protocol + +Upload only what changed. The server's dedup store already splits every +file into content-defined chunks (FastCDC, 64 KB – 1 MiB, avg 256 KB, +BLAKE3-addressed) and shares unchanged chunks between file versions — +but a classic upload still transfers every byte just for the server to +discard the known ones. This protocol moves the "which chunks are new?" +question to the client, so unchanged bytes never cross the wire. + +Editing a few bytes of a 500 MB file re-uploads ~1 MiB instead of +500 MB. + +## Who can use it + +Any authenticated API client. The OxiCloud web frontend adopts it in a +later phase; generic WebDAV/NextCloud clients cannot (their protocols +have no delta concept) — they keep uploading full bytes, and the server +keeps deduplicating those on write. + +Chunk boundaries are the **client's choice**: matching the server's +FastCDC parameters maximizes cross-version sharing, but any split with +chunks of 1 byte … 1 MiB is valid — correctness is guaranteed by +server-side verification, not by the chunking scheme. + +## The three steps + +### 1. `POST /api/files/delta/negotiate` + +```json +{ "chunks": [ { "h": "", "s": 262144 }, … ] } +``` + +Response — the distinct chunk hashes the caller must upload, in +first-occurrence order: + +```json +{ "missing": [ "", … ] } +``` + +The answer is **user-scoped**: a chunk counts as available only when one +of the *caller's own* (non-trashed) files already references it. The +endpoint is purely advisory — the commit re-checks entitlement +atomically, so a stale or spoofed answer can never leak content. + +### 2. `PUT /api/files/delta/chunks` + +Body: `application/octet-stream`, a sequence of frames + +``` +[u32 length, big-endian][length bytes] …repeated… +``` + +- one frame per chunk, each 1 byte … 1 MiB (the CDC maximum), +- whole request capped by `OXICLOUD_CHUNK_MAX_BYTES` (default 100 MB) — + split larger deltas across requests. + +The server **recomputes BLAKE3 of every frame itself** (a declared hash +is never trusted for content addressing) and registers the chunks as +unreferenced orphans (`ref_count = 0`). Response: + +```json +{ "received": [ { "h": "", "s": 262144 }, … ] } +``` + +Compare against your own hashes to catch corruption before committing. +Abandoned uploads need no cleanup call: the periodic GC sweeps +zero-reference chunks. + +### 3. `POST /api/files/delta/commit` + +```json +{ + "file_hash": "", + "chunks": [ { "h": "…", "s": 262144 }, … ], // full sequence, in order + "name": "video.mp4", "folder_id": "" // create mode + // — or — + "file_id": "" // update (replace content) +} +``` + +Server-side, in order: + +1. **AuthZ** — `Create` on the folder (create mode) or `Update` on the + file (update mode); quota on the logical size. +2. **Pin** — one atomic `UPDATE … RETURNING` takes a reference on every + distinct chunk the caller is *entitled* to: chunks reachable through + the caller's own files, or unreferenced orphans (the just-uploaded + state). Anything else → `409 { "still_missing": […] }`: upload + exactly those and retry the same commit. +3. **Verify** — the pinned sequence is re-read and the whole-file BLAKE3 + recomputed. A mismatch releases the pins and returns 400 (and an + audit event): the declared `file_hash` is never trusted, because a + forged manifest would poison future whole-file dedup hits for *other + users* uploading the genuine content. +4. **Attach** — the manifest is inserted with the same accounting as the + streaming byte path (a concurrent identical commit resolves via + `ON CONFLICT`: the loser's references are released and it becomes a + dedup hit). +5. **Row** — the file is created (`201`, body = FileDto) or its content + swapped (`200`). + +If the caller already owns the exact `file_hash`, the commit +short-circuits to a pure reference bump — chunks aren't even looked at +(same as `POST /api/files/by-hash`). + +## Security model + +- **No content oracle.** Possession is proven per chunk: without bytes + you can only claim what your own files already reference. Probing + someone else's chunk hashes yields `still_missing`, indistinguishable + from the hash never existing. +- **No manifest poisoning.** `file_hash` and every chunk hash are + recomputed server-side before becoming addressable. +- **Bounded resources.** Per-frame cap 1 MiB, per-request cap + `OXICLOUD_CHUNK_MAX_BYTES`, whole-file cap `OXICLOUD_MAX_UPLOAD_SIZE`, + per-caller rate limit (240 delta requests/min), quota enforced at + commit. Orphan chunks are GC-swept. +- **Audit.** Rejections emit `delta_upload.rejected` with stable + `reason` keys: `rate_limited`, `chunk_verification_failed`, + `file_hash_mismatch`. AuthZ denials surface as the engine's standard + `authz.denied`. + +## Error summary + +| Status | Meaning | Client action | +|---|---|---| +| 400 | malformed framing/hashes/sizes, or `file_hash` mismatch | fix and retry from step 1 | +| 404 | folder/file not found or not accessible | — | +| 409 | `{"still_missing": […]}` | PUT those chunks, retry the commit | +| 429 | rate limited | back off | +| 507 | quota exceeded | — | + +## Cost notes + +- `negotiate` is one indexed query (GIN over manifest chunk arrays). +- `commit` performs one sequential server-side read of the full logical + file for verification — cheap on local backends, a full object read on + S3/Azure. Still strictly cheaper than receiving the bytes, and the + client's bandwidth saving is unaffected. diff --git a/migrations/20260628000000_delta_upload_gin_index.sql b/migrations/20260628000000_delta_upload_gin_index.sql new file mode 100644 index 00000000..ef742b16 --- /dev/null +++ b/migrations/20260628000000_delta_upload_gin_index.sql @@ -0,0 +1,13 @@ +-- Delta-upload protocol: chunk-level ownership lookups. +-- +-- The negotiate/commit endpoints answer "which of these N chunk hashes may +-- this caller claim without uploading bytes?" — a chunk is claimable when a +-- manifest of one of the caller's (non-trashed) files contains it. That +-- containment test (`chunk_hashes @> ARRAY[hash]`) would be a sequential +-- scan over storage.chunk_manifests without an index; GIN makes each probe +-- an index lookup. +-- +-- Plan B if this disappoints at scale: a normalized +-- storage.manifest_chunks(file_hash, chunk_hash) join table. +CREATE INDEX IF NOT EXISTS idx_chunk_manifests_chunk_hashes_gin + ON storage.chunk_manifests USING GIN (chunk_hashes); diff --git a/src/application/services/delta_upload_service.rs b/src/application/services/delta_upload_service.rs new file mode 100644 index 00000000..68a463e0 --- /dev/null +++ b/src/application/services/delta_upload_service.rs @@ -0,0 +1,477 @@ +//! Delta-upload protocol — "upload only what changed". +//! +//! The CDC dedup store already shares unchanged chunks between file +//! versions *after* the bytes arrive; this protocol moves that detection +//! to the client side so unchanged bytes never cross the wire: +//! +//! 1. `negotiate`: the client sends the chunk hashes that compose its +//! file; the server answers which of them it cannot claim — only those +//! need uploading. +//! 2. `chunks`: the client uploads the missing chunks (raw frames). The +//! server recomputes every hash itself and registers the chunks as +//! unreferenced (`ref_count = 0`) orphans — pinned by the commit that +//! follows, or swept by the periodic GC if the client never returns. +//! 3. `commit`: the server pins one reference per distinct chunk (only +//! chunks the caller owns or unreferenced orphans — see the security +//! notes in `dedup_service.rs`), **re-reads the proposed sequence and +//! recomputes the whole-file BLAKE3** (a declared hash is never +//! trusted: a forged manifest would poison future whole-file dedup +//! hits for other users), attaches the manifest with the same +//! accounting as the streaming ingest, and creates or updates the +//! file row. +//! +//! Stateless by design: there is no session table. Every step re-derives +//! its facts from the chunk store, and the GC reclaims anything a client +//! abandons mid-protocol. + +use std::collections::HashSet; +use std::sync::Arc; + +use bytes::Bytes; +use futures::Stream; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; +use uuid::Uuid; + +use crate::application::dtos::file_dto::FileDto; +use crate::application::ports::authorization_ports::AuthorizationEngine; +use crate::application::ports::file_ports::{FileUploadUseCase, StoredBlob}; +use crate::application::ports::storage_ports::StorageUsagePort; +use crate::application::services::file_upload_service::FileUploadService; +use crate::application::services::storage_usage_service::StorageUsageService; +use crate::common::errors::DomainError; +use crate::common::mime_detect::{MAGIC_BYTES_LEN, refine_content_type}; +use crate::domain::services::authorization::{Permission, Resource, Subject}; +use crate::infrastructure::services::dedup_service::{CDC_MAX_CHUNK, DedupService}; +use crate::infrastructure::services::pg_acl_engine::PgAclEngine; + +// ── Wire DTOs ──────────────────────────────────────────────────────────────── + +/// One chunk reference: `h` = BLAKE3 hex (64 chars), `s` = size in bytes. +/// Field names are deliberately terse — a 10 GB file is ~40 000 of these. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct ChunkRef { + /// BLAKE3 hash of the chunk (64 hex chars). + pub h: String, + /// Chunk size in bytes (1 ..= 1 MiB). + pub s: u64, +} + +/// Request body of `POST /api/files/delta/negotiate`. +#[derive(Debug, Deserialize, ToSchema)] +pub struct DeltaNegotiateRequest { + /// The file's chunks, in order (duplicates allowed — repeated content). + pub chunks: Vec, +} + +/// Response of `POST /api/files/delta/negotiate`. +#[derive(Debug, Serialize, ToSchema)] +pub struct DeltaNegotiateResponse { + /// Distinct chunk hashes the caller must upload (first-occurrence order). + pub missing: Vec, +} + +/// Response of `PUT /api/files/delta/chunks` — the server-computed identity +/// of every received frame, in wire order. Clients compare against their +/// own hashes to detect corruption before committing. +#[derive(Debug, Serialize, ToSchema)] +pub struct DeltaChunksResponse { + pub received: Vec, +} + +/// Request body of `POST /api/files/delta/commit`. +/// +/// Exactly one of (`name` + `folder_id`) or `file_id` selects the mode: +/// create a new file, or replace an existing file's content. +#[derive(Debug, Deserialize, ToSchema)] +pub struct DeltaCommitRequest { + /// BLAKE3 of the complete file (verified server-side, never trusted). + pub file_hash: String, + /// Full chunk sequence, in file order (per occurrence). + pub chunks: Vec, + /// Create mode: file name (basename). + pub name: Option, + /// Create mode: target folder (caller needs Create permission). + pub folder_id: Option, + /// Update mode: file whose content is replaced (caller needs Write). + pub file_id: Option, +} + +/// Resolved commit mode after request validation. +enum CommitMode { + Create { name: String, folder_id: String }, + Update { file_id: String }, +} + +/// Outcome of a commit attempt. +pub enum DeltaCommitOutcome { + /// The file row exists; `created` distinguishes 201 from 200. + Done { file: FileDto, created: bool }, + /// Some chunks could not be pinned (GC race, skipped negotiate, or + /// chunks the caller may not claim). The client uploads exactly these + /// and retries the same commit. + StillMissing(Vec), +} + +// ── Service ────────────────────────────────────────────────────────────────── + +/// Orchestrates the three delta-upload steps. All authorization lives here +/// (service layer), per the project's AuthZ rule; handlers only +/// authenticate, rate-limit and translate the wire format. +pub struct DeltaUploadService { + dedup: Arc, + uploads: Arc, + quota: Arc, + authz: Arc, + /// Whole-file ceiling — same `max_upload_size` that bounds byte uploads. + max_total_size: u64, +} + +impl DeltaUploadService { + pub fn new( + dedup: Arc, + uploads: Arc, + quota: Arc, + authz: Arc, + max_total_size: u64, + ) -> Self { + Self { + dedup, + uploads, + quota, + authz, + max_total_size, + } + } + + /// Most chunks a single request may reference: the whole-file ceiling + /// divided by the smallest possible CDC chunk, with headroom for + /// fixed-size client chunkers. + fn max_chunk_count(&self) -> usize { + (self.max_total_size as usize + / crate::infrastructure::services::dedup_service::CDC_MIN_CHUNK) + .saturating_mul(2) + .max(1024) + } + + /// Shape-validate a chunk list: hash format, per-chunk size bounds, + /// count and total ceilings. Returns the total size. + fn validate_chunk_list(&self, chunks: &[ChunkRef]) -> Result { + if chunks.len() > self.max_chunk_count() { + return Err(DomainError::validation_error(format!( + "Too many chunks: {} (maximum {})", + chunks.len(), + self.max_chunk_count() + ))); + } + let mut total: u64 = 0; + for chunk in chunks { + if !is_valid_hash(&chunk.h) { + return Err(DomainError::validation_error( + "Invalid chunk hash format. Expected BLAKE3 (64 hex characters)", + )); + } + if chunk.s == 0 || chunk.s > CDC_MAX_CHUNK as u64 { + return Err(DomainError::validation_error(format!( + "Chunk size {} out of bounds (1 ..= {CDC_MAX_CHUNK})", + chunk.s + ))); + } + total = total.saturating_add(chunk.s); + } + if total > self.max_total_size { + return Err(DomainError::validation_error(format!( + "Declared total of {total} bytes exceeds the {}-byte upload ceiling", + self.max_total_size + ))); + } + Ok(total) + } + + /// Step 1: which of these chunks must the caller upload? + /// + /// Purely advisory and user-scoped — the commit re-checks entitlement + /// atomically, so a stale answer can never leak content. + pub async fn negotiate_with_perms( + &self, + caller_id: Uuid, + request: &DeltaNegotiateRequest, + ) -> Result { + self.validate_chunk_list(&request.chunks)?; + + let distinct = distinct_hashes(&request.chunks); + let claimable = self.dedup.claimable_chunks(caller_id, &distinct).await?; + let missing = distinct + .into_iter() + .filter(|h| !claimable.contains(h)) + .collect(); + Ok(DeltaNegotiateResponse { missing }) + } + + /// Step 2: store uploaded chunk frames. Hashes are computed + /// server-side; chunks land as unreferenced orphans awaiting a commit. + pub async fn receive_chunks(&self, frames: S) -> Result + where + S: Stream> + Send, + { + let received = self.dedup.store_loose_chunks(frames).await?; + Ok(DeltaChunksResponse { + received: received + .into_iter() + .map(|(h, s)| ChunkRef { h, s }) + .collect(), + }) + } + + /// Step 3: pin → verify → attach manifest → create/update the file row. + pub async fn commit_with_perms( + &self, + caller_id: Uuid, + request: DeltaCommitRequest, + ) -> Result { + // ── Shape ───────────────────────────────────────────────── + if !is_valid_hash(&request.file_hash) { + return Err(DomainError::validation_error( + "Invalid file_hash format. Expected BLAKE3 (64 hex characters)", + )); + } + let total_size = self.validate_chunk_list(&request.chunks)?; + + let mode = match (&request.name, &request.folder_id, &request.file_id) { + (Some(name), Some(folder_id), None) => { + let name = sanitize_file_name(name)?; + CommitMode::Create { + name, + folder_id: folder_id.clone(), + } + } + (None, None, Some(file_id)) => CommitMode::Update { + file_id: file_id.clone(), + }, + _ => { + return Err(DomainError::validation_error( + "Provide either name + folder_id (create) or file_id (update)", + )); + } + }; + + // ── AuthZ first: nothing is pinned for callers who may not write ── + match &mode { + CommitMode::Create { folder_id, .. } => { + let folder_uuid = Uuid::parse_str(folder_id) + .map_err(|_| DomainError::not_found("Folder", folder_id.clone()))?; + self.authz + .require( + Subject::User(caller_id), + Permission::Create, + Resource::Folder(folder_uuid), + ) + .await?; + } + CommitMode::Update { file_id } => { + let file_uuid = Uuid::parse_str(file_id) + .map_err(|_| DomainError::not_found("File", file_id.clone()))?; + self.authz + .require( + Subject::User(caller_id), + Permission::Update, + Resource::File(file_uuid), + ) + .await?; + } + } + + // ── Quota on the logical size (same semantics as a byte upload) ── + self.quota + .check_storage_quota(caller_id, total_size) + .await?; + + // ── Whole-file fast path: caller already owns this exact content ── + // Mirrors the instant-upload endpoint: a reference bump, no chunk + // work at all. Ownership is required — an existing-but-foreign + // manifest must be earned through the pin + verify path below. + if self + .dedup + .user_owns_blob_reference(&request.file_hash, &caller_id.to_string()) + .await + && let Some(metadata) = self.dedup.get_blob_metadata(&request.file_hash).await + { + self.dedup.add_reference(&request.file_hash).await?; + let blob = StoredBlob { + hash: request.file_hash.clone(), + size: metadata.size, + is_new_blob: false, + }; + let file = self + .register_row(caller_id, &mode, metadata.content_type, blob) + .await?; + return Ok(DeltaCommitOutcome::Done { + file, + created: matches!(mode, CommitMode::Create { .. }), + }); + } + + // ── Pin: atomically take one reference per distinct entitled chunk ── + let distinct = distinct_hashes(&request.chunks); + let pinned = self + .dedup + .pin_claimable_chunks(caller_id, &distinct) + .await?; + if pinned.len() != distinct.len() { + let still_missing: Vec = distinct + .iter() + .filter(|h| !pinned.contains(*h)) + .cloned() + .collect(); + let pinned_vec: Vec = pinned.into_iter().collect(); + self.dedup.release_pinned_chunks(&pinned_vec).await; + tracing::debug!( + "Delta commit: {} of {} chunks not claimable — client must upload them", + still_missing.len(), + distinct.len() + ); + return Ok(DeltaCommitOutcome::StillMissing(still_missing)); + } + + // ── Verify: the declared file_hash is recomputed from the pinned + // bytes before any manifest row can exist. ── + let verification = self + .dedup + .hash_chunk_sequence( + &request + .chunks + .iter() + .map(|c| (c.h.clone(), c.s)) + .collect::>(), + MAGIC_BYTES_LEN, + ) + .await; + let (computed_hash, head) = match verification { + Ok(v) => v, + Err(e) => { + self.dedup.release_pinned_chunks(&distinct).await; + tracing::info!( + target: "audit", + event = "delta_upload.rejected", + reason = "chunk_verification_failed", + caller_id = %caller_id, + file_hash = %request.file_hash, + "👮🏻‍♂️ Delta commit rejected: chunk sequence failed verification read", + ); + return Err(e); + } + }; + if computed_hash != request.file_hash { + self.dedup.release_pinned_chunks(&distinct).await; + tracing::info!( + target: "audit", + event = "delta_upload.rejected", + reason = "file_hash_mismatch", + caller_id = %caller_id, + declared_hash = %request.file_hash, + computed_hash = %computed_hash, + "👮🏻‍♂️ Delta commit rejected: declared file_hash does not match the chunk sequence", + ); + return Err(DomainError::validation_error( + "file_hash does not match the chunk sequence", + )); + } + + // ── Attach the manifest (shared accounting with the byte path) ── + let display_name = match &mode { + CommitMode::Create { name, .. } => name.clone(), + CommitMode::Update { file_id } => file_id.clone(), + }; + let content_type = match refine_content_type(&head, &display_name, "") { + ct if ct.is_empty() => "application/octet-stream".to_string(), + ct => ct, + }; + let chunk_hashes: Vec = request.chunks.iter().map(|c| c.h.clone()).collect(); + let chunk_sizes: Vec = request.chunks.iter().map(|c| c.s).collect(); + let attached = self + .dedup + .attach_manifest( + &request.file_hash, + &chunk_hashes, + &chunk_sizes, + total_size, + Some(content_type.clone()), + &distinct, + ) + .await?; + + let blob = StoredBlob { + hash: request.file_hash.clone(), + size: attached.size(), + is_new_blob: !attached.was_deduplicated(), + }; + let file = self + .register_row(caller_id, &mode, Some(content_type), blob) + .await?; + Ok(DeltaCommitOutcome::Done { + file, + created: matches!(mode, CommitMode::Create { .. }), + }) + } + + /// Create or update the file row against a blob reference the commit + /// already holds (the registration paths release it on failure). + async fn register_row( + &self, + caller_id: Uuid, + mode: &CommitMode, + content_type: Option, + blob: StoredBlob, + ) -> Result { + match mode { + CommitMode::Create { name, folder_id } => { + let content_type = + content_type.unwrap_or_else(|| "application/octet-stream".to_string()); + self.uploads + .upload_file_streaming( + name.clone(), + Some(folder_id.clone()), + content_type, + blob, + ) + .await + } + CommitMode::Update { file_id } => { + self.uploads + .update_file_content_by_id_with_perms(caller_id, file_id, blob) + .await + } + } + } +} + +/// 64 lowercase/uppercase hex characters. +fn is_valid_hash(hash: &str) -> bool { + hash.len() == 64 && hash.chars().all(|c| c.is_ascii_hexdigit()) +} + +/// Basename only — same path-traversal guard as the upload handlers. +fn sanitize_file_name(name: &str) -> Result { + let base = name + .rsplit('/') + .next() + .unwrap_or(name) + .rsplit('\\') + .next() + .unwrap_or(name) + .trim(); + if base.is_empty() { + return Err(DomainError::validation_error("File name must not be empty")); + } + Ok(base.to_string()) +} + +/// Distinct hashes in first-occurrence order. +fn distinct_hashes(chunks: &[ChunkRef]) -> Vec { + let mut seen = HashSet::new(); + chunks + .iter() + .filter(|c| seen.insert(c.h.as_str())) + .map(|c| c.h.clone()) + .collect() +} diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 02ac99a0..fabcb5c1 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -238,6 +238,71 @@ impl FileUploadService { Ok(dto) } + /// Swap an existing file's content to an already-ingested blob — the + /// update mode of the delta-upload commit. The caller needs `Write` + /// permission on the file; the blob reference is consumed (released on + /// failure by the write port, like every other registration path). + pub async fn update_file_content_by_id_with_perms( + &self, + caller_id: Uuid, + file_id: &str, + blob: StoredBlob, + ) -> Result { + let Some(InstantUploadDeps { authz, .. }) = &self.instant_upload else { + return Err(DomainError::internal_error( + "FileUpload", + "instant upload is not wired (authz/dedup/quota missing)", + )); + }; + let Some(file_read) = &self.file_read else { + return Err(DomainError::internal_error( + "FileUpload", + "read port is not wired", + )); + }; + + let file_uuid = Uuid::parse_str(file_id) + .map_err(|_| DomainError::not_found("File", file_id.to_string()))?; + authz + .require( + Subject::User(caller_id), + Permission::Update, + Resource::File(file_uuid), + ) + .await?; + + let file = file_read.get_file(file_id).await?; + let (new_hash, updated_at) = self + .file_write + .update_file_content_with_blob(file_id, &blob.hash, blob.size, None) + .await?; + // The file maps to a different blob now — stale cached content must + // never be served for the rest of its TTI window. + if let Some(cc) = &self.content_cache { + cc.invalidate(file_id).await; + } + + let parts = file.into_parts(); + let updated = crate::domain::entities::file::File::with_timestamps_and_blob_hash( + parts.id, + parts.name, + parts.storage_path, + blob.size, + parts.mime_type, + parts.folder_id, + parts.created_at, + updated_at as u64, + parts.owner_id, + new_hash, + ) + .map_err(|e| DomainError::internal_error("FileUpload", format!("rebuild entity: {e}")))?; + let dto = FileDto::from(updated); + if let Some(hook) = &self.file_lifecycle_hook { + hook.on_file_updated(file_id, &dto.content_hash, &dto.mime_type); + } + Ok(dto) + } + // ── private helpers ────────────────────────────────────────── /// Optionally update storage usage after a successful upload. diff --git a/src/application/services/mod.rs b/src/application/services/mod.rs index 7fa82755..7006ef9f 100644 --- a/src/application/services/mod.rs +++ b/src/application/services/mod.rs @@ -5,6 +5,7 @@ pub mod batch_operations; pub mod blob_lifecycle_service; pub mod calendar_service; pub mod contact_service; +pub mod delta_upload_service; pub mod device_auth_service; pub mod external_identity_service; pub mod favorites_service; diff --git a/src/common/di.rs b/src/common/di.rs index 49e9bf87..4e9a94ca 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -461,6 +461,18 @@ impl AppServiceFactory { ), ); + // Delta-upload protocol — chunk negotiation over the same dedup + // store. Bounded by the same whole-file ceiling as byte uploads. + let delta_upload_service = Arc::new( + crate::application::services::delta_upload_service::DeltaUploadService::new( + core.dedup_service.clone(), + file_upload_service.clone(), + storage_usage.clone(), + authz.clone(), + self.config.storage.max_upload_size as u64, + ), + ); + let file_retrieval_service = Arc::new(FileRetrievalService::new_with_cache( repos.file_read_repository.clone(), core.file_content_cache.clone(), @@ -505,6 +517,7 @@ impl AppServiceFactory { // Traits for abstraction folder_service, file_upload_service, + delta_upload_service, file_retrieval_service, file_management_service, file_use_case_factory, @@ -1040,6 +1053,13 @@ impl AppServiceFactory { user_profile_rate_limiter: Arc::new( crate::interfaces::middleware::rate_limit::RateLimiter::new(60, 60, 50_000), ), + // Delta upload: 240 requests / minute / caller. Generous for a + // real client (chunk PUTs carry up to 100 MB each) while + // stopping pin/negotiate floods; 50 000 tracked callers bound + // the memory like the other limiters. + delta_upload_rate_limiter: Arc::new( + crate::interfaces::middleware::rate_limit::RateLimiter::new(240, 60, 50_000), + ), // PR 12 — per-sharer email-invite ceiling: caller_id-keyed. // Defends against a compromised account spamming external // invites (each invite mints a new external user + email). @@ -1377,6 +1397,8 @@ pub struct ApplicationServices { // Traits for abstraction pub folder_service: Arc, pub file_upload_service: Arc, + pub delta_upload_service: + Arc, pub file_retrieval_service: Arc, pub file_management_service: Arc, pub file_use_case_factory: Arc, @@ -1496,6 +1518,9 @@ pub struct AppState { /// authenticated caller covers any legitimate UI rendering while /// throttling enumeration. pub user_profile_rate_limiter: Arc, + /// Per-caller flood guard for the delta-upload endpoints + /// (negotiate / chunks / commit share one budget). + pub delta_upload_rate_limiter: Arc, /// Per-sharer ceiling on `POST /api/grants` invitations whose /// subject is `{ type: "email" }`. 50 per hour keyed on /// `caller_id`. Anonymous attackers can't reach this code path diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 8f7ff957..5b6105d9 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -65,11 +65,11 @@ use crate::domain::errors::{DomainError, ErrorKind}; // ── CDC Constants ──────────────────────────────────────────────────────────── /// Minimum CDC chunk size (64 KB). -const CDC_MIN_CHUNK: usize = 65_536; +pub const CDC_MIN_CHUNK: usize = 65_536; /// Average CDC chunk size (256 KB). -const CDC_AVG_CHUNK: usize = 262_144; +pub const CDC_AVG_CHUNK: usize = 262_144; /// Maximum CDC chunk size (1 MB). -const CDC_MAX_CHUNK: usize = 1_048_576; +pub const CDC_MAX_CHUNK: usize = 1_048_576; // ── CDC helper types ───────────────────────────────────────────────────────── @@ -402,10 +402,42 @@ impl DedupService { S: Stream> + Send, { let outcome = self.ingest_chunks_from_stream(source).await?; + tracing::debug!( + "CDC stream ingested: {} ({} bytes, {} chunks, {} written)", + &outcome.file_hash[..12], + outcome.total_size, + outcome.chunk_hashes.len(), + outcome.newly_written, + ); let distinct = outcome.distinct_hashes(); - let total_size = outcome.total_size; - let file_hash = outcome.file_hash.clone(); + self.attach_manifest( + &outcome.file_hash, + &outcome.chunk_hashes, + &outcome.chunk_sizes, + outcome.total_size, + content_type, + &distinct, + ) + .await + } + /// Attach a manifest to chunk references the caller already holds (one + /// per distinct chunk hash) — the shared accounting tail of both + /// [`store_from_stream`] and the delta-upload commit. + /// + /// On a lost insert race or an already-existing manifest, the existing + /// manifest's ref_count is bumped FIRST and only then are the held chunk + /// references released (`distinct_held`); the reverse order could leave + /// the caller's file row without any manifest reference behind it. + pub async fn attach_manifest( + &self, + file_hash: &str, + chunk_hashes: &[String], + chunk_sizes: &[u64], + total_size: u64, + content_type: Option, + distinct_held: &[String], + ) -> Result { // A bounded retry covers the rare interleaving where the manifest // that beat our INSERT is deleted again before our ref bump lands. for _ in 0..3 { @@ -415,17 +447,11 @@ impl DedupService { VALUES ($1, $2, $3, $4, $5, $6, 1) ON CONFLICT (file_hash) DO NOTHING", ) - .bind(&file_hash) - .bind(&outcome.chunk_hashes) - .bind( - outcome - .chunk_sizes - .iter() - .map(|s| *s as i64) - .collect::>(), - ) + .bind(file_hash) + .bind(chunk_hashes) + .bind(chunk_sizes.iter().map(|s| *s as i64).collect::>()) .bind(total_size as i64) - .bind(outcome.chunk_hashes.len() as i32) + .bind(chunk_hashes.len() as i32) .bind(&content_type) .execute(self.pool.as_ref()) .await @@ -436,46 +462,253 @@ impl DedupService { if inserted > 0 { tracing::info!( - "NEW BLOB (CDC stream): {} ({} bytes, {} chunks, {} written)", + "NEW BLOB (CDC): {} ({} bytes, {} chunks)", &file_hash[..12], total_size, - outcome.chunk_hashes.len(), - outcome.newly_written, + chunk_hashes.len(), ); - self.fire_blob_creation_hooks(&file_hash, content_type.as_deref()); + self.fire_blob_creation_hooks(file_hash, content_type.as_deref()); return Ok(DedupResultDto::NewBlob { - hash: file_hash, + hash: file_hash.to_string(), size: total_size, }); } // The manifest already exists — either this exact content was // stored before or an identical concurrent upload just won the - // race. Bump ITS ref_count first and only then hand back this - // session's chunk references; the reverse order could leave the - // caller's file row without any manifest reference behind it. - if let Some(existing_size) = self.bump_manifest_if_exists(&file_hash).await? { - self.release_chunk_refs(self.pool.as_ref(), &distinct).await; + // race. Bump ITS ref_count, then hand back the held references. + if let Some(existing_size) = self.bump_manifest_if_exists(file_hash).await? { + self.release_chunk_refs(self.pool.as_ref(), distinct_held) + .await; tracing::info!( "DEDUP HIT (manifest): {} ({} bytes saved)", &file_hash[..12], existing_size, ); return Ok(DedupResultDto::ExistingBlob { - hash: file_hash, + hash: file_hash.to_string(), size: existing_size as u64, saved_bytes: existing_size as u64, }); } } - self.release_chunk_refs(self.pool.as_ref(), &distinct).await; + self.release_chunk_refs(self.pool.as_ref(), distinct_held) + .await; Err(DomainError::internal_error( "Dedup", format!("Manifest insert/bump kept racing for {file_hash}"), )) } + // ── Delta-upload primitives ────────────────────────────────── + // + // The delta protocol ("upload only what changed") lets a client claim + // chunks by hash instead of sending their bytes. Two invariants keep + // that from becoming a content oracle or a poisoning vector: + // + // 1. **Ownership**: without bytes, a caller may only claim chunks that + // are already reachable through their own (non-trashed) files, or + // unreferenced orphans (ref_count = 0 — i.e. "I just uploaded it"). + // Everything else must be uploaded; the store dedups it on write. + // 2. **Verification**: a declared file_hash is never trusted — the + // commit re-reads the proposed chunk sequence server-side and + // recomputes BLAKE3 before any manifest row exists. A forged hash + // would otherwise poison future whole-file dedup hits for OTHER + // users uploading the genuine content. + + /// Of `hashes` (distinct), the subset `caller_id` may claim without + /// uploading bytes: chunks referenced by manifests of the caller's + /// non-trashed files, or directly referenced as (legacy) whole-file + /// blobs. Backed by the GIN index on `chunk_manifests.chunk_hashes`. + pub async fn claimable_chunks( + &self, + caller_id: uuid::Uuid, + hashes: &[String], + ) -> Result, DomainError> { + if hashes.is_empty() { + return Ok(HashSet::new()); + } + sqlx::query_scalar::<_, String>( + "SELECT c.h FROM UNNEST($1::text[]) AS c(h) + WHERE EXISTS ( + SELECT 1 + FROM storage.files f + JOIN storage.chunk_manifests m ON m.file_hash = f.blob_hash + WHERE f.user_id = $2 AND NOT f.is_trashed + AND m.chunk_hashes @> ARRAY[c.h] + ) + OR EXISTS ( + SELECT 1 FROM storage.files f2 + WHERE f2.user_id = $2 AND NOT f2.is_trashed + AND f2.blob_hash = c.h + )", + ) + .bind(hashes) + .bind(caller_id) + .fetch_all(self.pool.as_ref()) + .await + .map(|rows| rows.into_iter().collect()) + .map_err(|e| DomainError::internal_error("Dedup", format!("claimable_chunks query: {e}"))) + } + + /// Pin one reference on each of `hashes` (distinct) that the caller is + /// entitled to claim — owned chunks (see [`claimable_chunks`]) or + /// unreferenced orphans (`ref_count = 0`, the just-uploaded state). + /// One statement: entitlement check and bump are atomic per row, so a + /// concurrent last-reference delete can never be resurrected and a + /// non-entitled hash is simply not returned. + /// + /// Returns the set actually pinned; the caller compares against its + /// input and reports the difference as `still_missing`. + pub async fn pin_claimable_chunks( + &self, + caller_id: uuid::Uuid, + hashes: &[String], + ) -> Result, DomainError> { + if hashes.is_empty() { + return Ok(HashSet::new()); + } + sqlx::query_scalar::<_, String>( + "UPDATE storage.blobs b + SET ref_count = ref_count + 1 + WHERE b.hash = ANY($1) + AND ( b.ref_count = 0 + OR EXISTS ( + SELECT 1 + FROM storage.files f + JOIN storage.chunk_manifests m ON m.file_hash = f.blob_hash + WHERE f.user_id = $2 AND NOT f.is_trashed + AND m.chunk_hashes @> ARRAY[b.hash::text] + ) + OR EXISTS ( + SELECT 1 FROM storage.files f2 + WHERE f2.user_id = $2 AND NOT f2.is_trashed + AND f2.blob_hash = b.hash + ) ) + RETURNING b.hash", + ) + .bind(hashes) + .bind(caller_id) + .fetch_all(self.pool.as_ref()) + .await + .map(|rows| rows.into_iter().collect()) + .map_err(|e| { + DomainError::internal_error("Dedup", format!("pin_claimable_chunks query: {e}")) + }) + } + + /// Release one reference per distinct hash — the public counterpart of + /// [`pin_claimable_chunks`] for aborted commits. Best-effort. + pub async fn release_pinned_chunks(&self, hashes: &[String]) { + self.release_chunk_refs(self.pool.as_ref(), hashes).await; + } + + /// Store client-provided loose chunks (delta upload, step 2). + /// + /// Each element of `frames` is one chunk's raw bytes (the wire framing + /// is the interface layer's concern). The hash is ALWAYS computed + /// server-side — a declared hash is never trusted for content + /// addressing. Chunks are written unsynced, made durable with one + /// batched sweep, then registered at `ref_count = 0`: unreferenced + /// orphans that either get pinned by a following commit or swept by + /// the periodic GC if the client never returns. `ON CONFLICT DO + /// NOTHING` keeps existing rows' reference counts untouched. + /// + /// Returns `(hash, size)` per frame, in input order. + pub async fn store_loose_chunks(&self, frames: S) -> Result, DomainError> + where + S: Stream> + Send, + { + futures::pin_mut!(frames); + + let mut received: Vec<(String, u64)> = Vec::new(); + let mut new_rows: Vec<(String, i64)> = Vec::new(); + let mut seen: HashSet = HashSet::new(); + + while let Some(frame) = frames.next().await { + let data = frame?; + if data.len() > CDC_MAX_CHUNK { + return Err(DomainError::validation_error(format!( + "Chunk frame of {} bytes exceeds the {CDC_MAX_CHUNK}-byte maximum", + data.len() + ))); + } + let hash = blake3::hash(&data).to_hex().to_string(); + received.push((hash.clone(), data.len() as u64)); + if seen.insert(hash.clone()) { + let len = data.len() as i64; + self.backend + .put_blob_from_bytes_unsynced(&hash, data) + .await?; + new_rows.push((hash, len)); + } + } + + if !new_rows.is_empty() { + // Durability before visibility — same invariant as the ingest + // engine: no PG row may ever point at unsynced bytes. + let hashes: Vec = new_rows.iter().map(|(h, _)| h.clone()).collect(); + let sizes: Vec = new_rows.iter().map(|(_, s)| *s).collect(); + self.backend.sync_blobs(&hashes).await?; + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count) + SELECT h, s, 0 FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) + ON CONFLICT (hash) DO NOTHING", + ) + .bind(&hashes) + .bind(&sizes) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to register chunks: {e}")) + })?; + } + + Ok(received) + } + + /// Verification read for the delta commit: stream the proposed chunk + /// sequence from the backend, recompute the whole-file BLAKE3 and + /// capture the first bytes for MIME sniffing. The caller must hold a + /// pin on every chunk (so a concurrent GC cannot pull bytes out from + /// under the read). Also validates each chunk's actual size against + /// the declared one — the manifest's Range arithmetic depends on it. + pub async fn hash_chunk_sequence( + &self, + chunks: &[(String, u64)], + sniff_len: usize, + ) -> Result<(String, Vec), DomainError> { + let mut hasher = blake3::Hasher::new(); + let mut head: Vec = Vec::with_capacity(sniff_len.min(16 * 1024)); + + for (hash, declared_size) in chunks { + let mut stream = self.backend.get_blob_stream(hash).await?; + let mut actual: u64 = 0; + while let Some(part) = stream.next().await { + let part = part.map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Verification read of chunk {hash}: {e}"), + ) + })?; + actual += part.len() as u64; + hasher.update(&part); + if head.len() < sniff_len { + let take = (sniff_len - head.len()).min(part.len()); + head.extend_from_slice(&part[..take]); + } + } + if actual != *declared_size { + return Err(DomainError::validation_error(format!( + "Chunk {hash} is {actual} bytes, manifest declares {declared_size}" + ))); + } + } + + Ok((hasher.finalize().to_hex().to_string(), head)) + } + /// Bump a manifest's ref_count if it exists; returns its total_size. /// Single statement — no window between the existence check and the bump. async fn bump_manifest_if_exists(&self, file_hash: &str) -> Result, DomainError> { @@ -2702,3 +2935,325 @@ mod rechunk_integration_tests { cleanup(&pool, &hash, &files).await; } } + +// ───────────────────────────────────────────────────────────────────────────── +// Integration tests for the delta-upload primitives — the entitlement and +// verification rules the chunk-negotiation protocol stands on. Same gating +// and DB conventions as the re-chunk suite above. +// ───────────────────────────────────────────────────────────────────────────── +#[cfg(integration_tests)] +#[allow(dead_code)] +mod delta_upload_integration_tests { + use super::*; + use crate::infrastructure::services::local_blob_backend::LocalBlobBackend; + use crate::integration_test_support::{ensure_clean_test_db, test_db_url}; + use sqlx::Row; + use sqlx::postgres::PgPoolOptions; + use tempfile::TempDir; + use uuid::Uuid; + + async fn test_pool() -> Arc { + let pool = PgPoolOptions::new() + .max_connections(4) + .connect(&test_db_url()) + .await + .expect("connect to test DB — run tests/common/spawn-db.sh first"); + ensure_clean_test_db(&pool).await; + Arc::new(pool) + } + + async fn seed_user(pool: &PgPool) -> Uuid { + sqlx::query("SELECT id FROM auth.users LIMIT 1") + .fetch_one(pool) + .await + .map(|r| r.get::("id")) + .expect("auth.users must be seeded (init-test-schema.sh)") + } + + async fn local_svc(pool: &Arc, dir: &TempDir) -> DedupService { + let backend = Arc::new(LocalBlobBackend::new(&dir.path().join("blobs"))); + backend.initialize().await.expect("init backend"); + DedupService::new(backend, pool.clone(), pool.clone()) + } + + /// Store `data` through the streaming path and give `user_id` a file + /// row referencing it — making its chunks claimable by that user. + async fn seed_owned_content( + svc: &DedupService, + pool: &PgPool, + user_id: Uuid, + data: &[u8], + label: &str, + ) -> (String, Vec, Uuid) { + let source = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::copy_from_slice(data))]); + let stored = svc + .store_from_stream(source, Some("application/octet-stream".into())) + .await + .expect("store"); + let file_hash = stored.hash().to_string(); + let chunks: Vec = sqlx::query_scalar( + "SELECT UNNEST(chunk_hashes) FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(&file_hash) + .fetch_all(pool) + .await + .expect("chunks"); + + let file_id: Uuid = sqlx::query_scalar( + "INSERT INTO storage.files (name, user_id, blob_hash, size) + VALUES ($1, $2, $3, $4) RETURNING id", + ) + .bind(format!( + "rust-test-delta-{label}-{}", + &Uuid::new_v4().to_string()[..8] + )) + .bind(user_id) + .bind(&file_hash) + .bind(data.len() as i64) + .fetch_one(pool) + .await + .expect("file row"); + (file_hash, chunks, file_id) + } + + async fn blob_ref(pool: &PgPool, hash: &str) -> Option { + sqlx::query_scalar("SELECT ref_count FROM storage.blobs WHERE hash = $1") + .bind(hash) + .fetch_optional(pool) + .await + .expect("blob query") + } + + async fn cleanup(pool: &PgPool, file_hash: &str, file_id: Uuid, extra_hashes: &[String]) { + let chunks: Option> = sqlx::query_scalar( + "SELECT chunk_hashes FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(file_hash) + .fetch_optional(pool) + .await + .unwrap_or(None); + let _ = sqlx::query("DELETE FROM storage.files WHERE id = $1") + .bind(file_id) + .execute(pool) + .await; + let _ = sqlx::query("DELETE FROM storage.chunk_manifests WHERE file_hash = $1") + .bind(file_hash) + .execute(pool) + .await; + let mut to_drop = chunks.unwrap_or_default(); + to_drop.push(file_hash.to_string()); + to_drop.extend_from_slice(extra_hashes); + let _ = sqlx::query("DELETE FROM storage.blobs WHERE hash = ANY($1)") + .bind(&to_drop) + .execute(pool) + .await; + } + + fn content(len: usize, salt: u8) -> Vec { + let mut data: Vec = (0..len) + .map(|i| { + ((i % 251) as u8) + .wrapping_add(salt) + .wrapping_add((i / 7919) as u8) + }) + .collect(); + data.extend_from_slice(Uuid::new_v4().as_bytes()); + data + } + + // ── Entitlement: claimable vs pin ──────────────────────────── + #[tokio::test] + async fn claim_and_pin_respect_ownership_and_orphans() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + let user = seed_user(&pool).await; + + // Owned content (multi-chunk), one foreign chunk (ref 1, no file + // row for this user), one orphan (ref 0), one unknown hash. + let data = content(3 * 1024 * 1024, 21); + let (file_hash, owned_chunks, file_id) = + seed_owned_content(&svc, &pool, user, &data, "claim").await; + assert!(owned_chunks.len() >= 3, "3 MiB must split into ≥3 chunks"); + + let foreign = blake3::hash(format!("foreign-{}", Uuid::new_v4()).as_bytes()) + .to_hex() + .to_string(); + let orphan = blake3::hash(format!("orphan-{}", Uuid::new_v4()).as_bytes()) + .to_hex() + .to_string(); + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count) VALUES ($1, 10, 1), ($2, 10, 0)", + ) + .bind(&foreign) + .bind(&orphan) + .execute(pool.as_ref()) + .await + .expect("seed foreign+orphan"); + let unknown = blake3::hash(format!("unknown-{}", Uuid::new_v4()).as_bytes()) + .to_hex() + .to_string(); + + let mut probe: Vec = owned_chunks.clone(); + probe.push(foreign.clone()); + probe.push(orphan.clone()); + probe.push(unknown.clone()); + + // claimable: only the owned chunks (advisory view — orphans are + // intentionally NOT advertised; the commit pin may still take them). + let claimable = svc.claimable_chunks(user, &probe).await.expect("claimable"); + for c in &owned_chunks { + assert!(claimable.contains(c), "owned chunk {c} must be claimable"); + } + assert!( + !claimable.contains(&foreign), + "foreign chunk must not be claimable" + ); + assert!( + !claimable.contains(&unknown), + "unknown chunk must not be claimable" + ); + + // pin: owned + orphan succeed; foreign and unknown are refused. + let pinned = svc.pin_claimable_chunks(user, &probe).await.expect("pin"); + for c in &owned_chunks { + assert!(pinned.contains(c), "owned chunk {c} must pin"); + } + assert!( + pinned.contains(&orphan), + "ref-0 orphan must pin (just-uploaded state)" + ); + assert!( + !pinned.contains(&foreign), + "foreign owned chunk must NOT pin" + ); + assert!(!pinned.contains(&unknown), "unknown hash must NOT pin"); + + // Ref counts moved exactly where they should. + assert_eq!(blob_ref(&pool, &orphan).await, Some(1), "orphan 0→1"); + assert_eq!( + blob_ref(&pool, &foreign).await, + Some(1), + "foreign untouched" + ); + assert_eq!( + blob_ref(&pool, &owned_chunks[0]).await, + Some(2), + "owned chunk 1→2 (manifest + pin)" + ); + + // Release restores the original counts (clamped at 0). + let pinned_vec: Vec = pinned.into_iter().collect(); + svc.release_pinned_chunks(&pinned_vec).await; + assert_eq!(blob_ref(&pool, &orphan).await, Some(0)); + assert_eq!(blob_ref(&pool, &owned_chunks[0]).await, Some(1)); + + cleanup(&pool, &file_hash, file_id, &[foreign, orphan]).await; + } + + // ── Loose chunk store ──────────────────────────────────────── + #[tokio::test] + async fn loose_chunks_register_as_orphans_without_touching_existing_refs() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + let user = seed_user(&pool).await; + + // An owned chunk that the client redundantly re-uploads. + let data = content(100 * 1024, 22); + let (file_hash, owned_chunks, file_id) = + seed_owned_content(&svc, &pool, user, &data, "loose").await; + let owned_chunk_bytes = { + let mut stream = svc.read_blob_stream(&file_hash).await.expect("stream"); + let mut out = Vec::new(); + while let Some(part) = stream.next().await { + out.extend_from_slice(&part.expect("part")); + } + out + }; + + let fresh = content(50 * 1024, 23); + let frames = stream::iter(vec![ + Ok::<_, DomainError>(Bytes::from(fresh.clone())), + Ok(Bytes::from(fresh.clone())), // duplicate frame + Ok(Bytes::from(owned_chunk_bytes.clone())), // already-referenced chunk + ]); + + let received = svc.store_loose_chunks(frames).await.expect("store loose"); + assert_eq!(received.len(), 3, "every frame is answered, in order"); + assert_eq!( + received[0].0, received[1].0, + "duplicate frames share a hash" + ); + let fresh_hash = received[0].0.clone(); + + assert_eq!( + blob_ref(&pool, &fresh_hash).await, + Some(0), + "fresh chunk lands as an unreferenced orphan" + ); + assert_eq!( + blob_ref(&pool, &owned_chunks[0]).await, + Some(1), + "re-uploading an existing chunk must not disturb its refs" + ); + + // The orphan's bytes are really there and addressable. + assert_eq!( + svc.backend().blob_exists(&fresh_hash).await.unwrap(), + true, + "orphan chunk bytes must exist in the backend" + ); + + cleanup(&pool, &file_hash, file_id, &[fresh_hash]).await; + } + + // ── Verification read ──────────────────────────────────────── + #[tokio::test] + async fn hash_chunk_sequence_recomputes_and_validates_sizes() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + let user = seed_user(&pool).await; + + let data = content(2 * 1024 * 1024 + 137, 24); + let (file_hash, _chunks, file_id) = + seed_owned_content(&svc, &pool, user, &data, "verify").await; + + let manifest: (Vec, Vec) = sqlx::query_as( + "SELECT chunk_hashes, chunk_sizes FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(&file_hash) + .fetch_one(pool.as_ref()) + .await + .expect("manifest"); + let sequence: Vec<(String, u64)> = manifest + .0 + .iter() + .cloned() + .zip(manifest.1.iter().map(|s| *s as u64)) + .collect(); + + let (computed, head) = svc + .hash_chunk_sequence(&sequence, 16) + .await + .expect("verification read"); + assert_eq!(computed, file_hash, "recomputed hash must match"); + assert_eq!( + &head[..], + &data[..16], + "sniff head must be the file's first bytes" + ); + + // A wrong declared size must be rejected — Range arithmetic + // depends on manifest sizes being true. + let mut lying = sequence.clone(); + lying[0].1 += 1; + assert!( + svc.hash_chunk_sequence(&lying, 0).await.is_err(), + "size lie must fail verification" + ); + + cleanup(&pool, &file_hash, file_id, &[]).await; + } +} diff --git a/src/interfaces/api/handlers/delta_upload_handler.rs b/src/interfaces/api/handlers/delta_upload_handler.rs new file mode 100644 index 00000000..e45765c0 --- /dev/null +++ b/src/interfaces/api/handlers/delta_upload_handler.rs @@ -0,0 +1,322 @@ +//! Delta-upload protocol endpoints — "upload only what changed". +//! +//! Wire surface of [`DeltaUploadService`]; see that module (and +//! `docs/delta-upload-protocol.md`) for the protocol and its security +//! model. These handlers only authenticate, rate-limit and translate +//! the wire formats — every decision lives in the application service. +//! +//! Chunk-frame wire format (`PUT /api/files/delta/chunks`): the body is a +//! sequence of `[u32 big-endian length][length bytes]` frames, one per +//! chunk, with `Content-Type: application/octet-stream`. Frames are capped +//! at the CDC maximum chunk size (1 MiB) and the whole request at the same +//! per-request ceiling as resumable chunk PUTs (`chunk_max_bytes`). + +use axum::{ + Json, + body::Body, + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use bytes::{Buf, Bytes, BytesMut}; +use futures::Stream; +use std::sync::Arc; +use tokio_stream::StreamExt; + +use crate::application::services::delta_upload_service::{ + DeltaChunksResponse, DeltaCommitOutcome, DeltaCommitRequest, DeltaNegotiateRequest, + DeltaNegotiateResponse, +}; +use crate::common::di::AppState; +use crate::common::errors::DomainError; +use crate::infrastructure::services::dedup_service::CDC_MAX_CHUNK; +use crate::interfaces::errors::AppError; +use crate::interfaces::middleware::auth::AuthUser; +use http_body_util::BodyStream; +use serde::Serialize; +use utoipa::ToSchema; + +/// 409 body of `POST /api/files/delta/commit` when chunks vanished between +/// negotiate and commit (GC race) or were never claimable: the client +/// uploads exactly these hashes and retries the same commit. +#[derive(Debug, Serialize, ToSchema)] +pub struct DeltaStillMissingResponse { + pub still_missing: Vec, +} + +/// Per-caller flood guard shared by the three delta endpoints. +fn check_rate_limit(state: &Arc, auth_user: &AuthUser) -> Result<(), AppError> { + if state + .delta_upload_rate_limiter + .check_and_increment(&auth_user.id.to_string()) + .is_err() + { + tracing::info!( + target: "audit", + event = "delta_upload.rejected", + reason = "rate_limited", + caller_id = %auth_user.id, + "👮🏻‍♂️ Delta upload rejected: per-caller rate limit exceeded", + ); + return Err(AppError::new( + StatusCode::TOO_MANY_REQUESTS, + "Too many delta-upload requests; please retry shortly", + "RateLimited", + )); + } + Ok(()) +} + +/// Parse a `[u32 BE length][bytes]` frame sequence from the request body. +/// +/// Streaming: peak RAM is one frame (≤ 1 MiB) plus one HTTP frame, +/// regardless of how many chunks the request carries. `max_total` bounds +/// the whole request body. +fn parse_chunk_frames( + body: Body, + max_total: usize, +) -> impl Stream> + Send { + async_stream::try_stream! { + let mut body_stream = BodyStream::new(body); + let mut buf = BytesMut::new(); + let mut expecting: Option = None; + let mut total: usize = 0; + + loop { + // Drain every complete frame already buffered. + loop { + match expecting { + None => { + if buf.len() < 4 { + break; + } + let len = + u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; + buf.advance(4); + if len == 0 || len > CDC_MAX_CHUNK { + Err(DomainError::validation_error(format!( + "Chunk frame of {len} bytes out of bounds (1 ..= {CDC_MAX_CHUNK})" + )))?; + } + expecting = Some(len); + } + Some(len) => { + if buf.len() < len { + break; + } + let frame = buf.split_to(len).freeze(); + expecting = None; + yield frame; + } + } + } + + match body_stream.next().await { + Some(Ok(http_frame)) => { + if let Some(data) = http_frame.data_ref() { + total += data.len(); + if total > max_total { + Err(DomainError::validation_error(format!( + "Request body exceeds the {max_total}-byte per-request cap; \ + split the chunk upload into several requests" + )))?; + } + buf.extend_from_slice(data); + } + } + Some(Err(e)) => { + Err(DomainError::validation_error(format!( + "Failed to read request body: {e}" + )))?; + } + None => { + if expecting.is_some() || !buf.is_empty() { + Err(DomainError::validation_error( + "Truncated chunk frame at end of body", + ))?; + } + break; + } + } + } + } +} + +#[utoipa::path( + post, + path = "/api/files/delta/negotiate", + request_body = DeltaNegotiateRequest, + responses( + (status = 200, description = "Chunks the caller must upload (the rest are claimable without bytes)", body = DeltaNegotiateResponse), + (status = 400, description = "Malformed chunk list (hash format, size bounds, count ceiling)"), + (status = 429, description = "Rate limited"), + ), + security(("bearerAuth" = [])), + tag = "delta-upload" +)] +pub async fn delta_negotiate( + State(state): State>, + auth_user: AuthUser, + Json(request): Json, +) -> Result { + check_rate_limit(&state, &auth_user)?; + let response = state + .applications + .delta_upload_service + .negotiate_with_perms(auth_user.id, &request) + .await + .map_err(AppError::from)?; + Ok(Json(response)) +} + +#[utoipa::path( + put, + path = "/api/files/delta/chunks", + request_body(content_type = "application/octet-stream", + description = "Sequence of [u32 BE length][bytes] frames, one per chunk (each ≤ 1 MiB)"), + responses( + (status = 200, description = "Server-computed identity of every received frame, in order", body = DeltaChunksResponse), + (status = 400, description = "Malformed framing, oversized frame, or oversized request"), + (status = 429, description = "Rate limited"), + ), + security(("bearerAuth" = [])), + tag = "delta-upload" +)] +pub async fn delta_upload_chunks( + State(state): State>, + auth_user: AuthUser, + body: Body, +) -> Result { + check_rate_limit(&state, &auth_user)?; + let frames = parse_chunk_frames(body, state.core.config.storage.chunk_max_bytes); + let response = state + .applications + .delta_upload_service + .receive_chunks(frames) + .await + .map_err(AppError::from)?; + Ok(Json(response)) +} + +#[utoipa::path( + post, + path = "/api/files/delta/commit", + request_body = DeltaCommitRequest, + responses( + (status = 201, description = "File created from the committed chunk sequence", body = crate::application::dtos::file_dto::FileDto), + (status = 200, description = "Existing file's content replaced", body = crate::application::dtos::file_dto::FileDto), + (status = 400, description = "Malformed request, or the declared file_hash does not match the chunk sequence"), + (status = 404, description = "Target folder/file not found or not accessible"), + (status = 409, description = "Chunks not claimable — upload them and retry", body = DeltaStillMissingResponse), + (status = 429, description = "Rate limited"), + (status = 507, description = "Storage quota exceeded"), + ), + security(("bearerAuth" = [])), + tag = "delta-upload" +)] +pub async fn delta_commit( + State(state): State>, + auth_user: AuthUser, + Json(request): Json, +) -> Result { + check_rate_limit(&state, &auth_user)?; + let outcome = state + .applications + .delta_upload_service + .commit_with_perms(auth_user.id, request) + .await + .map_err(AppError::from)?; + Ok(match outcome { + DeltaCommitOutcome::Done { file, created } => { + let status = if created { + StatusCode::CREATED + } else { + StatusCode::OK + }; + (status, Json(file)).into_response() + } + DeltaCommitOutcome::StillMissing(still_missing) => ( + StatusCode::CONFLICT, + Json(DeltaStillMissingResponse { still_missing }), + ) + .into_response(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Encode frames the way a client would. + fn encode(frames: &[&[u8]]) -> Vec { + let mut out = Vec::new(); + for f in frames { + out.extend_from_slice(&(f.len() as u32).to_be_bytes()); + out.extend_from_slice(f); + } + out + } + + async fn collect(body: Body, max_total: usize) -> Result, DomainError> { + let stream = parse_chunk_frames(body, max_total); + futures::pin_mut!(stream); + let mut out = Vec::new(); + while let Some(item) = stream.next().await { + out.push(item?); + } + Ok(out) + } + + #[tokio::test] + async fn roundtrips_frames_in_order() { + let wire = encode(&[b"first", b"second chunk", &[0xAB; 1000]]); + let frames = collect(Body::from(wire), usize::MAX).await.unwrap(); + assert_eq!(frames.len(), 3); + assert_eq!(&frames[0][..], b"first"); + assert_eq!(&frames[1][..], b"second chunk"); + assert_eq!(frames[2].len(), 1000); + } + + #[tokio::test] + async fn empty_body_yields_no_frames() { + let frames = collect(Body::empty(), usize::MAX).await.unwrap(); + assert!(frames.is_empty()); + } + + #[tokio::test] + async fn rejects_zero_length_frame() { + let wire = encode(&[b""]); + assert!(collect(Body::from(wire), usize::MAX).await.is_err()); + } + + #[tokio::test] + async fn rejects_frame_above_cdc_max() { + let mut wire = Vec::new(); + wire.extend_from_slice(&((CDC_MAX_CHUNK as u32) + 1).to_be_bytes()); + // Header alone is enough — the length is rejected before any data. + assert!(collect(Body::from(wire), usize::MAX).await.is_err()); + } + + #[tokio::test] + async fn rejects_truncated_frame() { + let mut wire = encode(&[b"complete"]); + wire.extend_from_slice(&10u32.to_be_bytes()); + wire.extend_from_slice(b"only5"); // promises 10, delivers 5 + assert!(collect(Body::from(wire), usize::MAX).await.is_err()); + } + + #[tokio::test] + async fn rejects_request_above_total_cap() { + let wire = encode(&[&[0u8; 600], &[1u8; 600]]); + assert!(collect(Body::from(wire), 1000).await.is_err()); + } + + #[tokio::test] + async fn accepts_frame_exactly_at_cdc_max() { + let big = vec![7u8; CDC_MAX_CHUNK]; + let wire = encode(&[&big]); + let frames = collect(Body::from(wire), usize::MAX).await.unwrap(); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].len(), CDC_MAX_CHUNK); + } +} diff --git a/src/interfaces/api/handlers/mod.rs b/src/interfaces/api/handlers/mod.rs index 2033ed2c..aebeb707 100644 --- a/src/interfaces/api/handlers/mod.rs +++ b/src/interfaces/api/handlers/mod.rs @@ -7,6 +7,7 @@ pub mod carddav_handler; pub mod chunked_upload_handler; pub mod contacts_handler; pub mod dedup_handler; +pub mod delta_upload_handler; pub mod device_auth_handler; pub mod favorites_handler; pub mod file_handler; diff --git a/src/interfaces/api/mod.rs b/src/interfaces/api/mod.rs index 51eb8e6d..a655c3f1 100644 --- a/src/interfaces/api/mod.rs +++ b/src/interfaces/api/mod.rs @@ -81,6 +81,9 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; handlers::file_handler::list_files_query, handlers::file_handler::upload_file_with_thumbnails, handlers::file_handler::create_file_by_hash, + handlers::delta_upload_handler::delta_negotiate, + handlers::delta_upload_handler::delta_upload_chunks, + handlers::delta_upload_handler::delta_commit, handlers::file_handler::download_file, handlers::file_handler::get_thumbnail, handlers::file_handler::upload_thumbnail, @@ -262,6 +265,13 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; ResourceContentDto, // File schemas FileDto, + // Delta-upload schemas + crate::application::services::delta_upload_service::ChunkRef, + crate::application::services::delta_upload_service::DeltaNegotiateRequest, + crate::application::services::delta_upload_service::DeltaNegotiateResponse, + crate::application::services::delta_upload_service::DeltaChunksResponse, + crate::application::services::delta_upload_service::DeltaCommitRequest, + handlers::delta_upload_handler::DeltaStillMissingResponse, MoveFilePayload, PaginationDto, PaginationRequestDto, diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 262dfa9c..4b0255b1 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -54,6 +54,9 @@ use crate::interfaces::api::handlers::batch_handler::{self, BatchHandlerState}; use crate::interfaces::api::handlers::chunked_upload_handler::{ cancel_upload, complete_upload, create_upload, get_upload_status, upload_chunk, }; +use crate::interfaces::api::handlers::delta_upload_handler::{ + delta_commit, delta_negotiate, delta_upload_chunks, +}; use crate::interfaces::api::handlers::file_handler::{ create_file_by_hash, delete_file, download_file, get_file_metadata, get_thumbnail, list_files_query, move_file_simple, rename_file, upload_file_with_thumbnails, upload_thumbnail, @@ -230,6 +233,9 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { .route("/", get(list_files_query)) .route("/upload", post(upload_file_with_thumbnails)) .route("/by-hash", post(create_file_by_hash)) + .route("/delta/negotiate", post(delta_negotiate)) + .route("/delta/chunks", put(delta_upload_chunks)) + .route("/delta/commit", post(delta_commit)) .route("/{id}", get(download_file)) .route( "/{id}/thumbnail/{size}",