diff --git a/docs/delta-upload-protocol.md b/docs/delta-upload-protocol.md index 171e1a1d..686198d5 100644 --- a/docs/delta-upload-protocol.md +++ b/docs/delta-upload-protocol.md @@ -108,6 +108,33 @@ 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`). +## Delta download (sync clients) + +The inverse direction, for a client app that already holds an older +version locally and wants the server's current one: + +1. `GET /api/files/{id}/manifest` → `{ file_hash, total_size, chunks }` + — the file's chunk recipe. **Owner-scoped** like the rest of the + delta surface (shared files use the regular download endpoints). + Served with `ETag: ""`; a manifest is immutable for a + given hash, so `If-None-Match` revalidation answers 304 — polling + sync clients pay one header round-trip per unchanged file. +2. Diff the manifest against the local chunk inventory (chunk the local + copy with the same WASM module the upload direction ships). +3. `POST /api/files/delta/download` with `{ "hashes": […] }` → the + requested chunks as `[u32 BE length][bytes]` frames in request order + (the same wire format as the upload direction). Entitlement is the + same possession rule as negotiate/commit: chunks must be reachable + through the caller's own files; anything else → 404 + `{ "not_available": […] }` — deliberately indistinguishable from + "never existed". Batches are bounded by `OXICLOUD_CHUNK_MAX_BYTES`; + split large deltas across requests. +4. Reassemble locally per the manifest order and verify the whole-file + BLAKE3 against `file_hash`. + +Editing 3 bytes of a 24 MB file on one device costs a second device one +manifest GET plus ~1 chunk (~256 KB) instead of 24 MB. + ## Security model - **No content oracle.** Possession is proven per chunk: without bytes @@ -122,8 +149,9 @@ short-circuits to a pure reference bump — chunks aren't even looked 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`. + `file_hash_mismatch` — and `delta_download.rejected` with + `manifest_not_owner` / `chunks_not_owned`. AuthZ denials surface as + the engine's standard `authz.denied`. ## Error summary diff --git a/src/application/services/delta_upload_service.rs b/src/application/services/delta_upload_service.rs index 68a463e0..07e4312a 100644 --- a/src/application/services/delta_upload_service.rs +++ b/src/application/services/delta_upload_service.rs @@ -36,12 +36,13 @@ 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::ports::storage_ports::{FileReadPort, 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::repositories::pg::FileBlobReadRepository; use crate::infrastructure::services::dedup_service::{CDC_MAX_CHUNK, DedupService}; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; @@ -97,6 +98,36 @@ pub struct DeltaCommitRequest { pub file_id: Option, } +/// Response of `GET /api/files/{id}/manifest` — the recipe to rebuild the +/// file from chunks. Immutable for a given `file_hash`, so clients cache +/// it keyed by hash (the endpoint also serves it with `ETag: file_hash`). +#[derive(Debug, Serialize, ToSchema)] +pub struct DeltaManifestResponse { + /// BLAKE3 of the whole file (verify the local reassembly against it). + pub file_hash: String, + /// Total size in bytes. + pub total_size: u64, + /// Full chunk sequence, in file order (per occurrence). + pub chunks: Vec, +} + +/// Request body of `POST /api/files/delta/download` — distinct chunk +/// hashes to fetch, served back as `[u32 BE length][bytes]` frames in +/// request order (the same wire format the upload direction uses). +#[derive(Debug, Deserialize, ToSchema)] +pub struct DeltaDownloadRequest { + pub hashes: Vec, +} + +/// Outcome of a chunk-download authorization. +pub enum DeltaDownloadOutcome { + /// Every requested chunk is servable: `(hash, size)` in request order. + Ready(Vec<(String, u64)>), + /// Some chunks are not available to this caller (not reachable through + /// their files, or unknown — deliberately indistinguishable). + NotAvailable(Vec), +} + /// Resolved commit mode after request validation. enum CommitMode { Create { name: String, folder_id: String }, @@ -121,26 +152,35 @@ pub enum DeltaCommitOutcome { pub struct DeltaUploadService { dedup: Arc, uploads: Arc, + file_read: Arc, quota: Arc, authz: Arc, /// Whole-file ceiling — same `max_upload_size` that bounds byte uploads. max_total_size: u64, + /// Per-request ceiling for batched chunk downloads — same budget as + /// the chunk-upload requests (`chunk_max_bytes`). + max_download_batch: u64, } impl DeltaUploadService { + #[allow(clippy::too_many_arguments)] pub fn new( dedup: Arc, uploads: Arc, + file_read: Arc, quota: Arc, authz: Arc, max_total_size: u64, + max_download_batch: u64, ) -> Self { Self { dedup, uploads, + file_read, quota, authz, max_total_size, + max_download_batch, } } @@ -414,6 +454,151 @@ impl DeltaUploadService { }) } + // ── Delta download ("download only what changed") ──────────── + + /// The chunk recipe of a file the caller owns — step 1 of a delta + /// download. Owner-scoped like the rest of the delta protocol (shared + /// files use the regular download endpoints); a non-owned or unknown + /// file is `NotFound`, and the denial is audited. + pub async fn file_manifest_with_perms( + &self, + caller_id: Uuid, + file_id: &str, + ) -> Result { + let file_uuid = Uuid::parse_str(file_id) + .map_err(|_| DomainError::not_found("File", file_id.to_string()))?; + // Engine check first (Read on the file): owners always pass and + // the engine audits denials; the ownership constraint below is the + // chunk layer's own entitlement standard. + self.authz + .require( + Subject::User(caller_id), + Permission::Read, + Resource::File(file_uuid), + ) + .await?; + + let file = self.file_read.get_file(file_id).await?; + let file_hash = file.content_hash().to_string(); + if !self + .dedup + .user_owns_blob_reference(&file_hash, &caller_id.to_string()) + .await + { + // Read permission without ownership (e.g. a grant): the delta + // surface is owner-scoped — the regular download endpoints + // serve shared content. + tracing::info!( + target: "audit", + event = "delta_download.rejected", + reason = "manifest_not_owner", + caller_id = %caller_id, + file_id = %file_id, + "👮🏻‍♂️ Delta download rejected: manifest requested by a non-owner", + ); + return Err(DomainError::not_found("File", file_id.to_string())); + } + + let Some((chunks, total_size)) = self.dedup.manifest_chunk_list(&file_hash).await? else { + return Err(DomainError::not_found("File", file_id.to_string())); + }; + Ok(DeltaManifestResponse { + file_hash, + total_size, + chunks: chunks.into_iter().map(|(h, s)| ChunkRef { h, s }).collect(), + }) + } + + /// Authorize a batched chunk download — step 2. Every hash must be + /// reachable through the caller's own files; otherwise the full list of + /// unavailable hashes is returned (the same information N individual + /// requests would reveal, in one round-trip). The total payload is + /// bounded by the per-request budget so clients split large deltas. + pub async fn authorize_chunk_download_with_perms( + &self, + caller_id: Uuid, + request: &DeltaDownloadRequest, + ) -> Result { + if request.hashes.is_empty() { + return Err(DomainError::validation_error("hashes must not be empty")); + } + if request.hashes.len() > self.max_chunk_count() { + return Err(DomainError::validation_error(format!( + "Too many chunks: {} (maximum {})", + request.hashes.len(), + self.max_chunk_count() + ))); + } + let mut distinct_seen = HashSet::new(); + for hash in &request.hashes { + if !is_valid_hash(hash) { + return Err(DomainError::validation_error( + "Invalid chunk hash format. Expected BLAKE3 (64 hex characters)", + )); + } + if !distinct_seen.insert(hash.as_str()) { + return Err(DomainError::validation_error( + "Duplicate hashes in download request", + )); + } + } + + let entitled = self + .dedup + .claimable_chunks(caller_id, &request.hashes) + .await?; + let not_available: Vec = request + .hashes + .iter() + .filter(|h| !entitled.contains(*h)) + .cloned() + .collect(); + if !not_available.is_empty() { + tracing::info!( + target: "audit", + event = "delta_download.rejected", + reason = "chunks_not_owned", + caller_id = %caller_id, + requested = request.hashes.len(), + denied = not_available.len(), + "👮🏻‍♂️ Delta download rejected: caller requested chunks outside their files", + ); + return Ok(DeltaDownloadOutcome::NotAvailable(not_available)); + } + + let sizes = self.dedup.chunk_sizes(&request.hashes).await?; + let mut ordered = Vec::with_capacity(request.hashes.len()); + let mut total: u64 = 0; + for hash in &request.hashes { + // Entitled implies indexed; a vanished row between the two + // queries surfaces as unavailable rather than a 500. + let Some(size) = sizes.get(hash) else { + return Ok(DeltaDownloadOutcome::NotAvailable(vec![hash.clone()])); + }; + total = total.saturating_add(*size); + ordered.push((hash.clone(), *size)); + } + if total > self.max_download_batch { + return Err(DomainError::validation_error(format!( + "Requested chunks total {total} bytes; the per-request ceiling is {} — split the download into smaller batches", + self.max_download_batch + ))); + } + Ok(DeltaDownloadOutcome::Ready(ordered)) + } + + /// Stream one authorized chunk's bytes (entitlement was established by + /// [`authorize_chunk_download_with_perms`]). + pub async fn chunk_stream( + &self, + hash: &str, + ) -> Result< + std::pin::Pin> + Send>>, + DomainError, + > { + self.dedup.chunk_stream(hash).await + } + /// 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( diff --git a/src/common/di.rs b/src/common/di.rs index 4e9a94ca..07227509 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -467,9 +467,11 @@ impl AppServiceFactory { crate::application::services::delta_upload_service::DeltaUploadService::new( core.dedup_service.clone(), file_upload_service.clone(), + repos.file_read_repository.clone(), storage_usage.clone(), authz.clone(), self.config.storage.max_upload_size as u64, + self.config.storage.chunk_max_bytes as u64, ), ); diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 5b6105d9..867c0767 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -516,6 +516,76 @@ impl DedupService { // 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. + // + // The download direction reuses invariant 1: a chunk's bytes are only + // served to callers whose own files already reference it. + + /// The ordered chunk list composing `file_hash`, for the delta-download + /// manifest: `(chunks[(hash, size)], total_size)`. + /// + /// Legacy whole-file blobs (pre-CDC, not yet re-chunked) are presented + /// as a single-chunk manifest of themselves — the chunk download path + /// can serve them directly, so sync clients need no special case. + pub async fn manifest_chunk_list( + &self, + file_hash: &str, + ) -> Result, u64)>, DomainError> { + let manifest = sqlx::query_as::<_, (Vec, Vec, i64)>( + "SELECT chunk_hashes, chunk_sizes, total_size + FROM storage.chunk_manifests WHERE file_hash = $1", + ) + .bind(file_hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {e}")))?; + + if let Some((hashes, sizes, total)) = manifest { + let chunks = hashes + .into_iter() + .zip(sizes.into_iter().map(|s| s as u64)) + .collect(); + return Ok(Some((chunks, total as u64))); + } + + // Legacy fallback: the blob is its own single chunk. + let legacy = sqlx::query_scalar::<_, i64>("SELECT size FROM storage.blobs WHERE hash = $1") + .bind(file_hash) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Legacy blob lookup: {e}")) + })?; + Ok(legacy.map(|size| (vec![(file_hash.to_string(), size as u64)], size as u64))) + } + + /// Sizes of the given chunk hashes from the dedup index, keyed by hash. + /// Hashes without a row are simply absent from the result. + pub async fn chunk_sizes( + &self, + hashes: &[String], + ) -> Result, DomainError> { + if hashes.is_empty() { + return Ok(std::collections::HashMap::new()); + } + sqlx::query_as::<_, (String, i64)>( + "SELECT hash, size FROM storage.blobs WHERE hash = ANY($1)", + ) + .bind(hashes) + .fetch_all(self.pool.as_ref()) + .await + .map(|rows| rows.into_iter().map(|(h, s)| (h, s as u64)).collect()) + .map_err(|e| DomainError::internal_error("Dedup", format!("chunk_sizes query: {e}"))) + } + + /// Stream one chunk's raw bytes from the backend. The caller is + /// responsible for entitlement (see [`claimable_chunks`]). + pub async fn chunk_stream( + &self, + hash: &str, + ) -> Result> + Send>>, DomainError> + { + self.backend.get_blob_stream(hash).await + } /// Of `hashes` (distinct), the subset `caller_id` may claim without /// uploading bytes: chunks referenced by manifests of the caller's diff --git a/src/interfaces/api/handlers/delta_upload_handler.rs b/src/interfaces/api/handlers/delta_upload_handler.rs index e45765c0..6912d37f 100644 --- a/src/interfaces/api/handlers/delta_upload_handler.rs +++ b/src/interfaces/api/handlers/delta_upload_handler.rs @@ -14,8 +14,8 @@ use axum::{ Json, body::Body, - extract::State, - http::StatusCode, + extract::{Path, State}, + http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, }; use bytes::{Buf, Bytes, BytesMut}; @@ -24,8 +24,8 @@ use std::sync::Arc; use tokio_stream::StreamExt; use crate::application::services::delta_upload_service::{ - DeltaChunksResponse, DeltaCommitOutcome, DeltaCommitRequest, DeltaNegotiateRequest, - DeltaNegotiateResponse, + DeltaChunksResponse, DeltaCommitOutcome, DeltaCommitRequest, DeltaDownloadOutcome, + DeltaDownloadRequest, DeltaManifestResponse, DeltaNegotiateRequest, DeltaNegotiateResponse, }; use crate::common::di::AppState; use crate::common::errors::DomainError; @@ -44,7 +44,15 @@ pub struct DeltaStillMissingResponse { pub still_missing: Vec, } -/// Per-caller flood guard shared by the three delta endpoints. +/// 404 body of `POST /api/files/delta/download` when some requested chunks +/// are not reachable through the caller's files (or don't exist — the two +/// are deliberately indistinguishable). +#[derive(Debug, Serialize, ToSchema)] +pub struct DeltaNotAvailableResponse { + pub not_available: Vec, +} + +/// Per-caller flood guard shared by the delta endpoints. fn check_rate_limit(state: &Arc, auth_user: &AuthUser) -> Result<(), AppError> { if state .delta_upload_rate_limiter @@ -243,6 +251,117 @@ pub async fn delta_commit( }) } +#[utoipa::path( + get, + path = "/api/files/{id}/manifest", + params(("id" = String, Path, description = "File ID")), + responses( + (status = 200, description = "Chunk recipe of the file (immutable per file_hash; served with ETag = file_hash)", body = DeltaManifestResponse), + (status = 304, description = "Not modified (If-None-Match matched the current file_hash)"), + (status = 404, description = "File not found, not accessible, or not owned by the caller"), + (status = 429, description = "Rate limited"), + ), + security(("bearerAuth" = [])), + tag = "delta-upload" +)] +pub async fn delta_file_manifest( + State(state): State>, + auth_user: AuthUser, + Path(file_id): Path, + headers: HeaderMap, +) -> Result { + check_rate_limit(&state, &auth_user)?; + let manifest = state + .applications + .delta_upload_service + .file_manifest_with_perms(auth_user.id, &file_id) + .await + .map_err(AppError::from)?; + + // A manifest is immutable for a given file_hash, so the hash IS the + // strong validator: sync clients polling a file revalidate for free. + let etag = format!("\"{}\"", manifest.file_hash); + if headers + .get(header::IF_NONE_MATCH) + .and_then(|v| v.to_str().ok()) + .is_some_and(|inm| inm == etag) + { + return Ok(Response::builder() + .status(StatusCode::NOT_MODIFIED) + .header(header::ETAG, etag) + .body(Body::empty()) + .unwrap()); + } + Ok(( + StatusCode::OK, + [ + (header::ETAG, etag), + (header::CACHE_CONTROL, "private, no-cache".to_string()), + ], + Json(manifest), + ) + .into_response()) +} + +#[utoipa::path( + post, + path = "/api/files/delta/download", + request_body = DeltaDownloadRequest, + responses( + (status = 200, description = "Requested chunks as [u32 BE length][bytes] frames, in request order", + content_type = "application/octet-stream"), + (status = 400, description = "Malformed hashes, duplicates, or batch above the per-request ceiling"), + (status = 404, description = "Some chunks are not available to this caller", body = DeltaNotAvailableResponse), + (status = 429, description = "Rate limited"), + ), + security(("bearerAuth" = [])), + tag = "delta-upload" +)] +pub async fn delta_download_chunks( + State(state): State>, + auth_user: AuthUser, + Json(request): Json, +) -> Result { + check_rate_limit(&state, &auth_user)?; + let service = state.applications.delta_upload_service.clone(); + let outcome = service + .authorize_chunk_download_with_perms(auth_user.id, &request) + .await + .map_err(AppError::from)?; + + let ordered = match outcome { + DeltaDownloadOutcome::NotAvailable(not_available) => { + return Ok(( + StatusCode::NOT_FOUND, + Json(DeltaNotAvailableResponse { not_available }), + ) + .into_response()); + } + DeltaDownloadOutcome::Ready(ordered) => ordered, + }; + let total: u64 = ordered.iter().map(|(_, s)| 4 + s).sum(); + + // Stream the frames: 4-byte length headers come from the (entitled) + // index sizes; bytes stream straight from the blob backend. Peak RAM + // is one backend read frame, independent of batch size. + let body_stream: std::pin::Pin> + Send>> = + Box::pin(async_stream::try_stream! { + for (hash, size) in ordered { + yield Bytes::copy_from_slice(&(size as u32).to_be_bytes()); + let mut chunk = service.chunk_stream(&hash).await.map_err(std::io::Error::other)?; + while let Some(part) = chunk.next().await { + yield part?; + } + } + }); + Ok(Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/octet-stream") + .header(header::CONTENT_LENGTH, total.to_string()) + .body(Body::from_stream(body_stream)) + .unwrap()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/interfaces/api/mod.rs b/src/interfaces/api/mod.rs index a655c3f1..e30113f3 100644 --- a/src/interfaces/api/mod.rs +++ b/src/interfaces/api/mod.rs @@ -84,6 +84,8 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; handlers::delta_upload_handler::delta_negotiate, handlers::delta_upload_handler::delta_upload_chunks, handlers::delta_upload_handler::delta_commit, + handlers::delta_upload_handler::delta_file_manifest, + handlers::delta_upload_handler::delta_download_chunks, handlers::file_handler::download_file, handlers::file_handler::get_thumbnail, handlers::file_handler::upload_thumbnail, @@ -272,6 +274,9 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; crate::application::services::delta_upload_service::DeltaChunksResponse, crate::application::services::delta_upload_service::DeltaCommitRequest, handlers::delta_upload_handler::DeltaStillMissingResponse, + handlers::delta_upload_handler::DeltaNotAvailableResponse, + crate::application::services::delta_upload_service::DeltaManifestResponse, + crate::application::services::delta_upload_service::DeltaDownloadRequest, MoveFilePayload, PaginationDto, PaginationRequestDto, diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 4b0255b1..14c9d017 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -55,7 +55,7 @@ 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, + delta_commit, delta_download_chunks, delta_file_manifest, 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, @@ -236,6 +236,8 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { .route("/delta/negotiate", post(delta_negotiate)) .route("/delta/chunks", put(delta_upload_chunks)) .route("/delta/commit", post(delta_commit)) + .route("/delta/download", post(delta_download_chunks)) + .route("/{id}/manifest", get(delta_file_manifest)) .route("/{id}", get(download_file)) .route( "/{id}/thumbnail/{size}", diff --git a/static/js/core/types.js b/static/js/core/types.js index f0b49ae1..aee0f564 100644 --- a/static/js/core/types.js +++ b/static/js/core/types.js @@ -541,3 +541,54 @@ * @property {string} hash BLAKE3 of the owned content (64 hex chars) */ +// ------------------- Delta sync (chunk negotiation) + +/** + * One chunk reference on the delta wire: terse on purpose (a 10 GB file + * is ~40 000 of these). Mirrors `ChunkRef` on the server + * (`delta_upload_service.rs`). + * @typedef {Object} DeltaChunkRef + * @property {string} h BLAKE3 of the chunk (64 hex chars) + * @property {number} s chunk size in bytes (1 ..= 1 MiB) + */ + +/** + * Response of `POST /api/files/delta/negotiate` — the distinct chunk + * hashes the caller must upload (user-scoped, advisory). + * @typedef {Object} DeltaNegotiateAnswer + * @property {string[]} missing + */ + +/** + * Request body of `POST /api/files/delta/commit`. Exactly one of + * (`name` + `folder_id`) or `file_id` selects create vs update mode. + * 201/200 responses carry a {@link FileItem}; 409 carries + * `{still_missing: string[]}` (upload those chunks and retry). + * @typedef {Object} DeltaCommitRequest + * @property {string} file_hash BLAKE3 of the whole file (verified server-side) + * @property {DeltaChunkRef[]} chunks full sequence, in file order + * @property {string} [name] create mode: file name + * @property {string} [folder_id] create mode: target folder + * @property {string} [file_id] update mode: file whose content is replaced + */ + +/** + * Response of `GET /api/files/{id}/manifest` — the recipe to rebuild a + * file from chunks (delta download, step 1). Immutable per `file_hash`; + * the endpoint serves it with `ETag: file_hash` so polling sync clients + * revalidate with a 304 for free. + * @typedef {Object} DeltaManifestAnswer + * @property {string} file_hash BLAKE3 of the whole file + * @property {number} total_size bytes + * @property {DeltaChunkRef[]} chunks full sequence, in file order + */ + +/** + * Request body of `POST /api/files/delta/download` (delta download, + * step 2). Responds with `[u32 BE length][bytes]` frames in request + * order, or 404 `{not_available: string[]}` for chunks outside the + * caller's files. + * @typedef {Object} DeltaDownloadRequest + * @property {string[]} hashes distinct chunk hashes to fetch + */ +