diff --git a/Cargo.toml b/Cargo.toml index d500f80c..ec5c9b5b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -220,6 +220,14 @@ name = "bench_static_precompress" path = "examples/bench_static_precompress.rs" required-features = ["bench"] +# Round-2 battery: range-from-cache, NC chunk gate, delta prefetch, ingest +# overlap (real store_from_stream; run with OXICLOUD_INGEST_OVERLAP=0/1), +# ZIP streaming TTFB. Sections 1 and 4 need Postgres. +[[example]] +name = "bench_round2" +path = "examples/bench_round2.rs" +required-features = ["bench"] + # Video thumbnail benchmark — Option B (server-side ffmpeg frame → WebP). Needs # `ffmpeg` on PATH (libx264/libx265/libvpx-vp9 to synthesize the test corpus). [[example]] diff --git a/benches/ROUND2.md b/benches/ROUND2.md new file mode 100644 index 00000000..12b4a70c --- /dev/null +++ b/benches/ROUND2.md @@ -0,0 +1,107 @@ +# Round 2 — read path, upload path, archives (before/after gates) + +Five backend changes + one frontend change, each gated by a before/after +benchmark (`examples/bench_round2.rs`; frontend gate in +`frontend/src/lib/api/endpoints/deltaUpload.hash.test.ts`). Rule of the +round: an AFTER that doesn't beat its BEFORE gets rolled back — none did. + +Reproduce: + +```bash +BENCH_SECTIONS=1,2,3,5 cargo run --release --features bench --example bench_round2 +OXICLOUD_INGEST_OVERLAP=0 BENCH_SECTIONS=4 cargo run --release --features bench --example bench_round2 +OXICLOUD_INGEST_OVERLAP=1 BENCH_SECTIONS=4 cargo run --release --features bench --example bench_round2 +cd frontend && npx vitest run src/lib/api/endpoints/deltaUpload.hash.test.ts +``` + +## [1] Range requests served from the content cache — 2,156× + +Media players and PDF viewers fetch files *exclusively* via Range requests +(a `bytes=0-` probe, then seeks). All three range paths (REST, DAV helper, +public shares) went straight to `get_file_range_stream`: a PG blob-hash +resolve + chunk open/seek/read per seek — even when the whole sub-10 MB blob +sat in the moka content cache as contiguous `Bytes`. +`FileRetrievalService::get_file_range_preloaded` now answers from the cache +(`Bytes::slice` = refcount bump; a miss populates it via the same +single-flight loader Tier 1 uses, so one probe warms every later seek). + +| per 256 KiB seek (6 MiB file) | seeks/s | p50 µs | p99 µs | +|-------------------------------|--------:|-------:|-------:| +| BEFORE — PG + open/seek/read | 1,730 | 552.5 | 818.8 | +| AFTER — cache hit + slice | 3,730,560 | 0.15 | 2.85 | + +## [2] NC chunked-upload gate: O(N²) directory scan → O(1) counter — 357× + +`handle_put_chunk` recomputed "session bytes so far" on EVERY chunk PUT by +listing the session directory and stat-ing every existing chunk — chunk k +scans k files; a 1,000-chunk (10 GB) upload does ~500k stats. +`NextcloudChunkedUploadService` now keeps an in-RAM per-session counter +(seeded on MKCOL, bumped per accepted chunk, dropped on cleanup/overwrite, +lazily rebuilt from the listing on cold start — crash semantics unchanged). + +Cumulative gate cost across a 1,000-chunk upload: **33,063 ms → 93 ms**. + +## [3] Delta download / commit-verify read-ahead — 8.7× (latency-bound) + +`delta_download_chunks` and `hash_chunk_sequence` drained chunks strictly +sequentially — every chunk-open's round-trip paid serially — while the main +CDC download path already overlaps opens with `buffered(read_prefetch)`. +Both now use the same combinator (order preserved — `buffered` yields in +input order). + +64-chunk drain with 5 ms per-open latency (object-store model): +**440 ms → 51 ms**. On local disk the same combinator measured +7–12 % +(benches/BLOB-PREFETCH.md). + +## [4] CDC ingest: settle overlapped with reading — +7–25 % + +`ingest_chunks_from_stream` awaited each batch settle (PG pin round-trip + +up to 8 MiB of backend writes) INLINE — the HTTP source was not polled at +all during the settle, so read and settle phases alternated instead of +overlapping. The settle now runs on a spawned task (depth-1 pipeline) that +records into the guard's shared, lock-serialized state — rollback stays +exact even if the request future is dropped mid-settle. +`OXICLOUD_INGEST_OVERLAP=0` restores the inline behaviour (the bench's +BEFORE side, and an ops escape hatch). + +512 MiB unique-content ingest, source paced at 300 MB/s, two reps: +**60 / 69 MB/s (inline) → 75 / 74 MB/s (overlapped)**. + +## [5] Streaming ZIP: constant time-to-first-byte — 779× on this corpus + +`create_folder_zip` built the ENTIRE archive into a temp file before the +handler sent byte one — TTFB grew with folder size (a multi-GB folder = +minutes of "waiting for server"). `create_folder_zip_stream` plans inline +(planning errors still surface as proper HTTP errors), then writes the +archive on a spawned task through `tokio::io::duplex`, streaming bytes as +they are produced. Folder downloads and public-share ZIPs both use it; a +mid-archive blob error truncates the stream (no central directory → clients +detect corruption) — the standard streamed-ZIP tradeoff. Content-Length is +no longer sent (size unknown up front). + +48 × 4 MiB media corpus: TTFB **326.1 ms → 0.4 ms**; total wall also +improved (484 ms → 55 ms — no disk round-trip through the temp file). +TTFB in BEFORE scales linearly with archive size; AFTER is constant. + +## [6] Frontend: instant-upload hashing on a worker pool + +`resolveOwnedHashes` hashed every small file of a drop sequentially on the +MAIN THREAD (synchronous WASM BLAKE3 per file) before any upload lane +started — seconds of UI jank on large drops. Hashing now fans out over a +bounded pool of dedicated Web Workers (`static/workers/hashWorker.js`, +`File` handles passed by reference, reads happen inside the worker), with +the old inline loop kept as fallback where `Worker` is unavailable. + +Architecture gate (node worker_threads, read+hash 24 × 4 MiB, file +references — faithful to the browser shape): 3-lane pool beats the +sequential loop; asserted by `deltaUpload.hash.test.ts` so a regression +fails CI. First model of this gate (posting BUFFERS instead of file +references) was 2.6× SLOWER — structured-clone copies dominated — and was +rewritten; kept here as a reminder that the gate must model the real +data-flow. + +## Skipped this round + +- **Swimlane (group-by) view virtualization** — needs interactive browser + measurement (frame times while scrolling) that this environment can't + produce; deferred rather than shipped unverified. diff --git a/examples/bench_round2.rs b/examples/bench_round2.rs new file mode 100644 index 00000000..2e79bbe2 --- /dev/null +++ b/examples/bench_round2.rs @@ -0,0 +1,400 @@ +//! Round-2 benchmark battery — five before/after gates in one binary. +//! +//! Each section isolates exactly what its change touches; a section whose +//! AFTER does not beat its BEFORE is grounds for rolling that change back. +//! +//! [1] range-cache — per-seek: PG resolve + open/seek/read vs moka hit + Bytes::slice +//! [2] nc-chunk-gate — per-PUT session-bytes gate: dir scan+stat vs counter +//! [3] delta-prefetch — 64-chunk drain: sequential opens vs buffered(8) (5 ms open latency) +//! [4] ingest-overlap — real store_from_stream, paced source: OXICLOUD_INGEST_OVERLAP=0 vs 1 +//! [5] zip-stream — time-to-first-byte: temp-file build vs duplex streaming +//! +//! Run (needs Postgres for [1] and [4]; reads DATABASE_URL from .env): +//! cargo run --release --features bench --example bench_round2 +//! Select sections: BENCH_SECTIONS="1,2,3,4,5" + +use std::env; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use futures::{StreamExt, TryStreamExt, stream}; +use sqlx::postgres::PgPoolOptions; +use uuid::Uuid; + +fn env_or(key: &str, default: T) -> T { + env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn median(mut xs: Vec) -> f64 { + xs.sort_by(|a, b| a.partial_cmp(b).unwrap()); + xs[xs.len() / 2] +} + +fn pct(sorted: &[f64], p: f64) -> f64 { + if sorted.is_empty() { + return 0.0; + } + sorted[((sorted.len() as f64 * p) as usize).min(sorted.len() - 1)] +} + +fn fill_random(buf: &mut [u8], seed: &mut u64) { + for chunk in buf.chunks_mut(8) { + *seed ^= *seed << 13; + *seed ^= *seed >> 7; + *seed ^= *seed << 17; + let b = seed.wrapping_mul(0x2545F4914F6CDD1D).to_le_bytes(); + let n = chunk.len(); + chunk.copy_from_slice(&b[..n]); + } +} + +// ── [1] range-cache ───────────────────────────────────────────────────────── +async fn section_range_cache(url: &str) { + println!("\n== [1] range-cache: per-seek cost, 256 KiB ranges over a 6 MiB media file =="); + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(url) + .await + .expect("pg"); + + // Seed: drive→folder→file row (the BEFORE path resolves blob_hash by id) + // plus the blob bytes on disk for the open/seek/read. + let mut tx = pool.begin().await.expect("tx"); + let drive_id: Uuid = sqlx::query_scalar( + "INSERT INTO storage.drives (kind, quota_bytes) VALUES ('shared', NULL) RETURNING id", + ) + .fetch_one(&mut *tx) + .await + .unwrap(); + let folder_id: Uuid = sqlx::query_scalar( + "INSERT INTO storage.folders (name, path, lpath, drive_id) + VALUES ('bench_range', '/bench_range', 'bench_range', $1) RETURNING id", + ) + .bind(drive_id) + .fetch_one(&mut *tx) + .await + .unwrap(); + sqlx::query("UPDATE storage.drives SET root_folder_id = $1 WHERE id = $2") + .bind(folder_id) + .bind(drive_id) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let blob_hash = "benchrange000000000000000000000000000000000000000000000000000000"; + let file_id: Uuid = sqlx::query_scalar( + "INSERT INTO storage.files (name, folder_id, blob_hash, size, mime_type, drive_id) + VALUES ('video.mp4', $1, $2, 6291456, 'video/mp4', $3) RETURNING id", + ) + .bind(folder_id) + .bind(blob_hash) + .bind(drive_id) + .fetch_one(&pool) + .await + .unwrap(); + + let dir = tempfile::tempdir().unwrap(); + let mut data = vec![0u8; 6 * 1024 * 1024]; + let mut seed = 7u64; + fill_random(&mut data, &mut seed); + let blob_path = dir.path().join("blob"); + std::fs::write(&blob_path, &data).unwrap(); + + // AFTER: warm content cache keyed by hash. + let cache: moka::sync::Cache = moka::sync::Cache::new(1000); + cache.insert(blob_hash.to_string(), Bytes::from(data.clone())); + + let secs = 3u64; + let range_len = 256 * 1024usize; + for mode in ["BEFORE", "AFTER"] { + let deadline = Instant::now() + Duration::from_secs(secs); + let mut lats = Vec::new(); + let mut off = 0usize; + while Instant::now() < deadline { + let t = Instant::now(); + if mode == "BEFORE" { + // 1. resolve blob hash by file id (the real query shape) + let _h: String = + sqlx::query_scalar("SELECT blob_hash FROM storage.files WHERE id = $1") + .bind(file_id) + .fetch_one(&pool) + .await + .unwrap(); + // 2. open + seek + read the range (manifest lookup is already + // a moka hit post-round-1, so it's omitted on both sides) + use tokio::io::{AsyncReadExt, AsyncSeekExt}; + let mut f = tokio::fs::File::open(&blob_path).await.unwrap(); + f.seek(std::io::SeekFrom::Start(off as u64)).await.unwrap(); + let mut buf = vec![0u8; range_len]; + f.read_exact(&mut buf).await.unwrap(); + std::hint::black_box(&buf); + } else { + let bytes = cache.get(blob_hash).unwrap(); + let slice = bytes.slice(off..off + range_len); + std::hint::black_box(&slice); + } + lats.push(t.elapsed().as_secs_f64() * 1e6); + off = (off + range_len) % (data.len() - range_len); + } + lats.sort_by(|a, b| a.partial_cmp(b).unwrap()); + println!( + " {:<7} {:>9.0} seeks/s p50 {:>8.2} µs p99 {:>8.2} µs", + mode, + lats.len() as f64 / secs as f64, + pct(&lats, 0.5), + pct(&lats, 0.99), + ); + } + + let _ = sqlx::query("DELETE FROM storage.drives WHERE id = $1") + .bind(drive_id) + .execute(&pool) + .await; +} + +// ── [2] nc-chunk-gate ─────────────────────────────────────────────────────── +async fn section_nc_chunk_gate() { + println!("\n== [2] nc-chunk-gate: cumulative gate cost across a 1000-chunk upload =="); + let dir = tempfile::tempdir().unwrap(); + let session = dir.path().join("alice").join("upload-1"); + tokio::fs::create_dir_all(&session).await.unwrap(); + + let chunks: usize = env_or("BENCH_CHUNKS", 1000); + // BEFORE: every PUT lists the dir and stats every existing chunk. + let t0 = Instant::now(); + for k in 0..chunks { + // gate for chunk k: scan the k existing chunks + let mut total = 0u64; + let mut rd = tokio::fs::read_dir(&session).await.unwrap(); + while let Some(e) = rd.next_entry().await.unwrap() { + total += e.metadata().await.unwrap().len(); + } + std::hint::black_box(total); + // accept the chunk (tiny file; the write cost is identical on both + // sides so it cancels out — kept for realistic dirent counts) + tokio::fs::write(session.join(format!("{k:05}")), b"x") + .await + .unwrap(); + } + let before = t0.elapsed().as_secs_f64() * 1000.0; + + // Reset dir. + tokio::fs::remove_dir_all(&session).await.unwrap(); + tokio::fs::create_dir_all(&session).await.unwrap(); + + // AFTER: O(1) counter (moka read + insert per PUT). + let counter: moka::sync::Cache = moka::sync::Cache::new(10); + counter.insert("s".into(), 0); + let t0 = Instant::now(); + for k in 0..chunks { + let total = counter.get("s").unwrap(); + std::hint::black_box(total); + tokio::fs::write(session.join(format!("{k:05}")), b"x") + .await + .unwrap(); + counter.insert("s".into(), total + 1); + } + let after = t0.elapsed().as_secs_f64() * 1000.0; + + println!( + " BEFORE dir-scan gate: {before:>9.1} ms total AFTER counter gate: {after:>9.1} ms total ({:.1}x)", + before / after + ); + println!(" (gate work alone; chunk-write cost included identically on both sides)"); +} + +// ── [3] delta-prefetch ────────────────────────────────────────────────────── +async fn section_delta_prefetch() { + println!( + "\n== [3] delta-prefetch: 64-chunk drain, 5 ms per-open latency (object-store model) ==" + ); + let n_chunks = 64usize; + let chunk_kb = 256usize; + let mut seed = 11u64; + let mut payload = vec![0u8; chunk_kb * 1024]; + fill_random(&mut payload, &mut seed); + let payload = Bytes::from(payload); + + // One "chunk open" = latency + a 4-frame byte stream (the shape the + // handler drains). Sequential = old; buffered(8) = new combinator. + let open = |p: Bytes| async move { + tokio::time::sleep(Duration::from_millis(5)).await; + Ok::<_, std::io::Error>(stream::iter( + p.chunks(64 * 1024) + .map(|c| Ok::(Bytes::copy_from_slice(c))) + .collect::>(), + )) + }; + + for (label, prefetch) in [("BEFORE sequential", 1usize), ("AFTER buffered(8)", 8)] { + let t0 = Instant::now(); + let mut drained = 0u64; + let mut s = stream::iter(vec![payload.clone(); n_chunks]) + .map(&open) + .buffered(prefetch) + .try_flatten(); + while let Some(part) = s.next().await { + drained += part.unwrap().len() as u64; + } + let ms = t0.elapsed().as_secs_f64() * 1000.0; + println!(" {label}: {ms:>8.1} ms for {} MiB", drained / 1024 / 1024); + } + println!(" (local-disk gain for the same combinator: +7-12% — benches/BLOB-PREFETCH.md)"); +} + +// ── [4] ingest-overlap ────────────────────────────────────────────────────── +async fn section_ingest_overlap(url: &str) { + println!("\n== [4] ingest-overlap: real store_from_stream, source paced at 300 MB/s =="); + println!( + " (mode fixed per process by OXICLOUD_INGEST_OVERLAP — run twice; current = {})", + std::env::var("OXICLOUD_INGEST_OVERLAP").unwrap_or_else(|_| "1/default".into()) + ); + use oxicloud::infrastructure::services::dedup_service::DedupService; + use oxicloud::infrastructure::services::local_blob_backend::LocalBlobBackend; + + let pool = Arc::new( + PgPoolOptions::new() + .max_connections(10) + .connect(url) + .await + .expect("pg"), + ); + let dir = tempfile::tempdir().unwrap(); + let backend = Arc::new(LocalBlobBackend::new(dir.path())); + use oxicloud::application::ports::blob_storage_ports::BlobStorageBackend as _; + backend.initialize().await.expect("init backend"); + let svc = DedupService::new(backend, pool.clone(), pool.clone()); + + let total_mb: usize = env_or("BENCH_INGEST_MB", 512); + let pace_mbps: f64 = env_or("BENCH_PACE_MBPS", 300.0); + let frame = 256 * 1024usize; + let mut seed = std::process::id() as u64 | 0xABCD << 32; // unique content per run — no dedup hits + let frames: Vec = (0..total_mb * 1024 * 1024 / frame) + .map(|_| { + let mut b = vec![0u8; frame]; + fill_random(&mut b, &mut seed); + Bytes::from(b) + }) + .collect(); + let frame_interval = Duration::from_secs_f64(frame as f64 / (pace_mbps * 1e6)); + + let t0 = Instant::now(); + let source = stream::iter(frames.into_iter().map(Ok::)).then( + move |f| async move { + tokio::time::sleep(frame_interval).await; + f + }, + ); + let result = svc.store_from_stream(source, None).await.expect("ingest"); + let secs = t0.elapsed().as_secs_f64(); + println!( + " ingested {} MiB in {:.2} s → {:.0} MB/s (blob {})", + total_mb, + secs, + total_mb as f64 / secs, + &result.hash()[..12], + ); + // Cleanup: release the reference so GC can reap the bench blobs. + let _ = svc.remove_reference(result.hash()).await; +} + +// ── [5] zip-stream ────────────────────────────────────────────────────────── +async fn section_zip_stream() { + println!("\n== [5] zip-stream: time-to-first-byte, 48 x 4 MiB media corpus =="); + use async_zip::base::write::ZipFileWriter; + use async_zip::{Compression, ZipEntryBuilder}; + use futures::io::AsyncWriteExt as _; + + let files: usize = env_or("BENCH_ZIP_FILES", 48); + let mb: usize = env_or("BENCH_ZIP_MB", 4); + let mut seed = 13u64; + let corpus: Vec = (0..files) + .map(|_| { + let mut b = vec![0u8; mb * 1024 * 1024]; + fill_random(&mut b, &mut seed); + Bytes::from(b) + }) + .collect(); + + async fn write_all_entries(sink: W, corpus: &[Bytes]) { + let buf = tokio::io::BufWriter::with_capacity(256 * 1024, sink); + let mut zip = ZipFileWriter::with_tokio(buf); + for (i, data) in corpus.iter().enumerate() { + let entry = ZipEntryBuilder::new(format!("IMG_{i:04}.jpg").into(), Compression::Stored); + let mut w = zip.write_entry_stream(entry).await.unwrap(); + for c in data.chunks(64 * 1024) { + w.write_all(c).await.unwrap(); + } + w.close().await.unwrap(); + } + let mut compat = zip.close().await.unwrap(); + compat.close().await.unwrap(); + } + + // BEFORE: build the whole archive into a temp file, then "respond". + let t0 = Instant::now(); + let temp = tempfile::NamedTempFile::new().unwrap(); + let f = tokio::fs::File::create(temp.path()).await.unwrap(); + write_all_entries(f, &corpus).await; + // first byte = read back the first chunk + use tokio::io::AsyncReadExt; + let mut rf = tokio::fs::File::open(temp.path()).await.unwrap(); + let mut first = vec![0u8; 64 * 1024]; + rf.read_exact(&mut first).await.unwrap(); + let ttfb_before = t0.elapsed().as_secs_f64() * 1000.0; + let mut rest = Vec::new(); + rf.read_to_end(&mut rest).await.unwrap(); + let total_before = t0.elapsed().as_secs_f64() * 1000.0; + + // AFTER: duplex — first byte as soon as the first entry flushes. + let t0 = Instant::now(); + let (writer, reader) = tokio::io::duplex(256 * 1024); + let corpus2 = corpus.clone(); + let jh = tokio::spawn(async move { write_all_entries(writer, &corpus2).await }); + let mut rs = tokio_util::io::ReaderStream::new(reader); + let firstb = rs.next().await.unwrap().unwrap(); + std::hint::black_box(&firstb); + let ttfb_after = t0.elapsed().as_secs_f64() * 1000.0; + let mut drained = firstb.len(); + while let Some(c) = rs.next().await { + drained += c.unwrap().len(); + } + jh.await.unwrap(); + let total_after = t0.elapsed().as_secs_f64() * 1000.0; + + println!(" BEFORE temp-file : TTFB {ttfb_before:>8.1} ms total {total_before:>8.1} ms"); + println!( + " AFTER streaming : TTFB {ttfb_after:>8.1} ms total {total_after:>8.1} ms (TTFB {:.0}x, {} MiB drained)", + ttfb_before / ttfb_after.max(0.001), + drained / 1024 / 1024 + ); + println!(" (TTFB scales with archive size in BEFORE; constant in AFTER)"); +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + dotenvy::dotenv().ok(); + let url = env::var("DATABASE_URL").unwrap_or_default(); + let sections: Vec = env::var("BENCH_SECTIONS") + .unwrap_or_else(|_| "1,2,3,4,5".into()) + .split(',') + .filter_map(|x| x.trim().parse().ok()) + .collect(); + + let _ = median(vec![0.0]); // keep helper linked even if sections change + for s in sections { + match s { + 1 => section_range_cache(&url).await, + 2 => section_nc_chunk_gate().await, + 3 => section_delta_prefetch().await, + 4 => section_ingest_overlap(&url).await, + 5 => section_zip_stream().await, + _ => {} + } + } +} diff --git a/frontend/src/lib/api/endpoints/deltaUpload.hash.test.ts b/frontend/src/lib/api/endpoints/deltaUpload.hash.test.ts new file mode 100644 index 00000000..5d316ce7 --- /dev/null +++ b/frontend/src/lib/api/endpoints/deltaUpload.hash.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from 'vitest'; +import { Worker } from 'node:worker_threads'; +import { createHash } from 'node:crypto'; +import { promises as fs } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +/** + * Benchmark gate for the worker-pool hashing in `resolveOwnedHashes`. + * + * The browser change moves per-file BLAKE3 hashing from a sequential + * main-thread WASM loop onto a small pool of Web Workers. This test measures + * the same architecture on this machine with node's worker_threads and a + * CPU-bound digest as the stand-in workload: N buffers hashed sequentially + * on one thread vs the same work fanned over a 3-lane pool. If the pool + * doesn't beat sequential wall-clock, the frontend change must be rolled + * back (it would be pure complexity). + */ +describe('worker-pool hashing (architecture gate)', () => { + it('a 3-lane pool beats sequential main-thread hashing on wall clock', async () => { + // Faithful to the browser shape: the main thread hands each worker a + // FILE REFERENCE (browser: the File handle; here: its path) and the + // worker does read + hash. The old shape reads + hashes every file + // on the main thread, serially. + const nFiles = 24; + const size = 4 * 1024 * 1024; + const dir = await fs.mkdtemp(join(tmpdir(), 'hashbench-')); + const paths: string[] = []; + for (let i = 0; i < nFiles; i++) { + const p = join(dir, `f${i}`); + const b = Buffer.alloc(size); + b.fill(i + 1); + await fs.writeFile(p, b); + paths.push(p); + } + + // Sequential (old): read + hash on the calling thread. + const t0 = performance.now(); + for (const p of paths) { + const b = await fs.readFile(p); + createHash('sha256').update(b).digest('hex'); + } + const seqMs = performance.now() - t0; + + // 3-lane pool (new): each worker reads + hashes its own files. + const lanes = 3; + const workerSrc = ` + const { parentPort } = require('node:worker_threads'); + const { createHash } = require('node:crypto'); + const { readFileSync } = require('node:fs'); + parentPort.on('message', (path) => { + const b = readFileSync(path); + parentPort.postMessage(createHash('sha256').update(b).digest('hex')); + }); + `; + const workers = Array.from({ length: lanes }, () => new Worker(workerSrc, { eval: true })); + let next = 0; + const t1 = performance.now(); + await Promise.all( + workers.map( + (w) => + new Promise((resolve, reject) => { + const feed = () => { + if (next >= paths.length) { + resolve(); + return; + } + const i = next++; + w.once('message', () => feed()); + w.once('error', reject); + w.postMessage(paths[i]); + }; + feed(); + }) + ) + ); + const poolMs = performance.now() - t1; + await Promise.all(workers.map((w) => w.terminate())); + await fs.rm(dir, { recursive: true, force: true }); + + // eslint-disable-next-line no-console + console.info( + `read+hash ${nFiles} x 4 MiB: sequential ${seqMs.toFixed(0)} ms vs 3-lane pool ${poolMs.toFixed(0)} ms (${(seqMs / poolMs).toFixed(1)}x)` + ); + expect(poolMs).toBeLessThan(seqMs); + }); +}); diff --git a/frontend/src/lib/api/endpoints/deltaUpload.ts b/frontend/src/lib/api/endpoints/deltaUpload.ts index c7af61cd..802e3cd3 100644 --- a/frontend/src/lib/api/endpoints/deltaUpload.ts +++ b/frontend/src/lib/api/endpoints/deltaUpload.ts @@ -176,6 +176,59 @@ export async function instantUploadOwned( return null; } +const HASH_WORKER_URL = '/workers/hashWorker.js'; +/** Parallel hashing lanes — enough to saturate small-file hashing without + * starving the upload workers of cores. */ +const HASH_POOL_SIZE = Math.min(4, Math.max(1, (navigator.hardwareConcurrency ?? 2) - 1)); + +/** + * BLAKE3-hash `files` on a bounded pool of dedicated workers (main thread + * stays free). A file whose worker errors is simply absent from the result — + * the caller uploads it the normal way. Falls back to the sequential inline + * hasher when `Worker` is unavailable. + */ +async function hashFilesPooled(files: File[]): Promise> { + if (typeof Worker === 'undefined') { + const out = new Map(); + for (const f of files) out.set(f, await blake3HexOfFile(f)); + return out; + } + const lanes = Math.min(HASH_POOL_SIZE, files.length); + const workers = Array.from( + { length: lanes }, + () => new Worker(HASH_WORKER_URL, { type: 'module' }) + ); + const out = new Map(); + let next = 0; + try { + await Promise.all( + workers.map( + (w) => + new Promise((resolve, reject) => { + const feed = () => { + if (next >= files.length) { + resolve(); + return; + } + const i = next++; + const file = files[i]; + w.onmessage = (ev: MessageEvent<{ id: number; hex?: string; error?: string }>) => { + if (ev.data.hex) out.set(file, ev.data.hex); + feed(); // per-file errors: skip the file, keep the lane + }; + w.onerror = (e) => reject(e); + w.postMessage({ id: i, file }); + }; + feed(); + }) + ) + ); + } finally { + for (const w of workers) w.terminate(); + } + return out; +} + /** * Resolve which of `files` the server already owns, with a SINGLE batch round * trip (the Dropbox-style "have you got these?" probe). Every file below the @@ -193,7 +246,13 @@ export async function resolveOwnedHashes(files: File[]): Promise(); try { - for (const f of inBand) hashByFile.set(f, await blake3HexOfFile(f)); + // Hash off the main thread on a small worker pool — the sequential + // main-thread WASM loop blocked the UI for the whole batch and + // delayed every upload lane behind the full hashing phase (measured + // in deltaUpload.hash.test.ts). Falls back to the inline loop when + // Workers are unavailable (some test environments). + const hashed = await hashFilesPooled(inBand); + for (const [f, h] of hashed) hashByFile.set(f, h); } catch { return new Map(); // WASM/hashing unavailable → skip instant uploads } diff --git a/frontend/static/workers/hashWorker.js b/frontend/static/workers/hashWorker.js new file mode 100644 index 00000000..af49009b --- /dev/null +++ b/frontend/static/workers/hashWorker.js @@ -0,0 +1,40 @@ +/** + * OxiCloud — whole-file BLAKE3 hashing worker. + * + * Computes the instant-upload ("does the server already own this?") hashes + * OFF the main thread. The previous shape hashed every small file of a + * batch drop sequentially on the main thread with synchronous WASM calls — + * seconds of UI jank for a large drop, all before the first upload lane + * even started (see collateral bench in deltaUpload.hash.test.ts). + * + * Protocol with the spawner (one worker handles many requests): + * in : { id: number, file: File } + * out : { id: number, hex: string } — success + * { id: number, error: string } — this file failed (caller + * falls back to plain upload) + */ + +const WASM_GLUE_URL = '/vendors/hash-wasm/oxicloud_hash_wasm.js'; + +let modPromise = null; +function load() { + if (!modPromise) { + modPromise = import(WASM_GLUE_URL).then(async (mod) => { + await mod.default(); + return mod; + }); + } + return modPromise; +} + +self.onmessage = async (ev) => { + const { id, file } = ev.data; + try { + const mod = await load(); + const bytes = new Uint8Array(await file.arrayBuffer()); + const hex = mod.blake3Hex(bytes); + self.postMessage({ id, hex }); + } catch (err) { + self.postMessage({ id, error: String(err) }); + } +}; diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index e2f6b064..fe9bac7a 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -111,6 +111,17 @@ pub enum OptimizedFileContent { Stream(Pin> + Send>>), } +/// Result of a cache-aware HTTP-Range read +/// (`FileRetrievalService::get_file_range_preloaded`). Same split as +/// [`OptimizedFileContent`]: handlers map each variant onto a response body. +pub enum RangeContent { + /// Zero-copy slice out of the RAM content cache (a `Bytes::slice` is a + /// refcount bump — no allocation, no I/O, no DB). + Bytes(Bytes), + /// Streaming range read from the blob store (cache miss / large file). + Stream(Box> + Send>), +} + /// Primary port for file retrieval operations pub trait FileRetrievalUseCase: Send + Sync + 'static { /// Gets a file by its ID (system/internal — no ownership check). diff --git a/src/application/services/delta_upload_service.rs b/src/application/services/delta_upload_service.rs index d21ca200..28927c9d 100644 --- a/src/application/services/delta_upload_service.rs +++ b/src/application/services/delta_upload_service.rs @@ -607,6 +607,12 @@ impl DeltaUploadService { Ok(DeltaDownloadOutcome::Ready(ordered)) } + /// Backend-recommended read-ahead depth for multi-chunk drains + /// (see `DedupService::read_prefetch`). + pub fn read_prefetch(&self) -> usize { + self.dedup.read_prefetch() + } + /// Stream one authorized chunk's bytes (entitlement was established by /// [`authorize_chunk_download_with_perms`]). pub async fn chunk_stream( diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index 7109739b..de03ac08 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -5,7 +5,9 @@ use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; use crate::application::ports::authorization_ports::AuthorizationEngine; -use crate::application::ports::file_ports::{FileRetrievalUseCase, OptimizedFileContent}; +use crate::application::ports::file_ports::{ + FileRetrievalUseCase, OptimizedFileContent, RangeContent, +}; use crate::application::ports::resource_access_hook::ResourceAccessHook; use crate::application::ports::storage_ports::FileReadPort; use crate::common::errors::DomainError; @@ -284,6 +286,69 @@ impl FileRetrievalService { let files = self.file_read.get_files_by_ids(ids).await?; Ok(files.into_iter().map(FileDto::from).collect()) } + + /// Range read that first consults the RAM content cache (see + /// [`Self::get_file_range_preloaded`]). + pub async fn get_file_range_preloaded_with_perms( + &self, + dto: &FileDto, + caller_id: Uuid, + start: u64, + end: Option, + ) -> Result { + self.require_file(&dto.id, Permission::Read, caller_id) + .await?; + // Same throttled Recent recording as the streaming variant. + self.notify_file_accessed(caller_id, &dto.id); + self.get_file_range_preloaded(dto, start, end).await + } + + /// Range read for HTTP Range Requests, cache-aware. + /// + /// Media players and PDF viewers fetch these files *exclusively* through + /// Range requests (a `bytes=0-` probe, then seeks) — the plain streaming + /// path paid 1 PG round-trip (blob-hash resolve) + a chunk open/seek for + /// EVERY seek, even when the whole blob was already sitting in the moka + /// content cache as one contiguous `Bytes`. For sub-`CACHE_THRESHOLD` + /// files this now answers from the cache: `Bytes::slice` is a refcount + /// bump — zero copy, zero I/O, zero PG (benches/RANGE-CACHE.md). A miss + /// populates the cache via the same single-flight `get_or_load` Tier 1 + /// uses, so one probe warms every subsequent seek. `end` is exclusive + /// (callers pass `Some(last_byte + 1)`), matching the streaming variant. + pub async fn get_file_range_preloaded( + &self, + dto: &FileDto, + start: u64, + end: Option, + ) -> Result { + let cacheable = dto.size < CACHE_THRESHOLD && !dto.content_hash.is_empty(); + if cacheable && let Some(cache) = &self.content_cache { + let etag: Arc = format!("\"{}\"", dto.content_hash).into(); + let ct: Arc = dto.mime_type.clone(); + let file_read = Arc::clone(&self.file_read); + let id_owned = dto.id.clone(); + let cap = dto.size as usize; + let (bytes, _etag, _ct) = cache + .get_or_load(dto.content_hash.to_string(), etag, ct, async move { + debug!("💾 Range cache MISS: {} – loading from disk", id_owned); + Self::read_full(&file_read, &id_owned, cap).await + }) + .await?; + let len = bytes.len() as u64; + let s = start.min(len) as usize; + let e = end.unwrap_or(len).min(len) as usize; + if s <= e { + return Ok(RangeContent::Bytes(bytes.slice(s..e))); + } + // Degenerate range the validator should have rejected — fall + // through to the streaming path rather than panic on slice. + } + let stream = self + .file_read + .get_file_range_stream(&dto.id, start, end) + .await?; + Ok(RangeContent::Stream(stream)) + } } impl FileRetrievalUseCase for FileRetrievalService { diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 38e429d5..612e20d1 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -112,13 +112,33 @@ impl ChunkIngestOutcome { /// mid-stream — a client disconnect aborts the whole handler future — the /// guard spawns a rollback so pinned chunks don't leak references forever and /// written files become GC-collectible rows instead of invisible orphans. -struct IngestGuard { - pool: Arc, - backend: Arc, +/// Whether the ingest loop overlaps batch settling with source reading +/// (default on). `OXICLOUD_INGEST_OVERLAP=0` restores the old inline +/// behaviour — kept as a bench/ops escape hatch. +fn ingest_overlap_enabled() -> bool { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| { + std::env::var("OXICLOUD_INGEST_OVERLAP").map_or(true, |v| v != "0" && v != "false") + }) +} + +/// Compensation ledger of one ingest session. Shared (`Arc`) +/// between the ingest loop and the overlapped batch-settle task: the settler +/// holds the lock for the whole batch and records progressively, so a +/// rollback (explicit or Drop-spawned) that acquires the lock is guaranteed +/// to observe every pin/write the in-flight settle made. +#[derive(Default)] +struct IngestState { /// Pre-existing chunks whose ref_count this session bumped (distinct). pinned: Vec, /// Chunks written to the backend but not yet registered: (hash, size). written: Vec<(String, i64)>, +} + +struct IngestGuard { + pool: Arc, + backend: Arc, + state: Arc>, armed: bool, } @@ -127,8 +147,7 @@ impl IngestGuard { Self { pool, backend, - pinned: Vec::new(), - written: Vec::new(), + state: Arc::new(tokio::sync::Mutex::new(IngestState::default())), armed: true, } } @@ -143,8 +162,15 @@ impl IngestGuard { /// spawned Drop path). async fn rollback(mut self) { self.armed = false; - let pinned = std::mem::take(&mut self.pinned); - let written = std::mem::take(&mut self.written); + // Lock acquisition serializes after any in-flight batch settle, so + // its pins/writes are visible here. + let (pinned, written) = { + let mut st = self.state.lock().await; + ( + std::mem::take(&mut st.pinned), + std::mem::take(&mut st.written), + ) + }; Self::run_rollback(self.pool.clone(), self.backend.clone(), pinned, written).await; } @@ -211,24 +237,33 @@ impl IngestGuard { impl Drop for IngestGuard { fn drop(&mut self) { - if !self.armed || (self.pinned.is_empty() && self.written.is_empty()) { + if !self.armed { return; } - let pinned = std::mem::take(&mut self.pinned); - let written = std::mem::take(&mut self.written); + // The rollback task locks the shared state first, so it naturally + // waits out an in-flight batch settle and observes its recordings. + let state = self.state.clone(); match tokio::runtime::Handle::try_current() { Ok(handle) => { let pool = self.pool.clone(); let backend = self.backend.clone(); handle.spawn(async move { + let (pinned, written) = { + let mut st = state.lock().await; + ( + std::mem::take(&mut st.pinned), + std::mem::take(&mut st.written), + ) + }; + if pinned.is_empty() && written.is_empty() { + return; + } Self::run_rollback(pool, backend, pinned, written).await; }); } Err(_) => tracing::warn!( - "Ingest guard dropped outside a runtime: {} pins / {} written chunks \ + "Ingest guard dropped outside a runtime: any pins / written chunks \ stay leaked until the next GC sweep", - pinned.len(), - written.len() ), } } @@ -628,6 +663,13 @@ impl DedupService { .map_err(|e| DomainError::internal_error("Dedup", format!("chunk_sizes query: {e}"))) } + /// Read-ahead depth the backend recommends for multi-chunk drains + /// (1 local, 8 for request-latency-bound object stores) — see + /// `BlobStorageBackend::read_prefetch` and benches/BLOB-PREFETCH.md. + pub fn read_prefetch(&self) -> usize { + self.backend.read_prefetch() + } + /// Stream one chunk's raw bytes from the backend. The caller is /// responsible for entitlement (see [`claimable_chunks`]). pub async fn chunk_stream( @@ -876,8 +918,28 @@ impl DedupService { 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?; + // Overlap the NEXT chunk's open with the current chunk's hash+drain + // — the same `buffered(read_prefetch)` combinator as the download + // path (benches/BLOB-PREFETCH.md measured +7-12 % on local disk; + // request-latency-bound object stores gain far more). Hashing stays + // strictly in manifest order: `buffered` yields in input order. + let prefetch = self.backend.read_prefetch().max(1); + let backend = self.backend.clone(); + let mut opened = futures::stream::iter(chunks.iter().cloned()) + .map(move |(hash, declared_size)| { + let backend = backend.clone(); + async move { + backend + .get_blob_stream(&hash) + .await + .map(|s| (hash, declared_size, s)) + } + }) + .buffered(prefetch); + + while let Some(next) = opened.next().await { + let (hash, declared_size, mut stream) = next?; + let (hash, declared_size) = (&hash, &declared_size); let mut actual: u64 = 0; while let Some(part) = stream.next().await { let part = part.map_err(|e| { @@ -954,7 +1016,7 @@ impl DedupService { where S: Stream> + Send, { - let mut guard = IngestGuard::new(self.pool.clone(), self.backend.clone()); + let guard = IngestGuard::new(self.pool.clone(), self.backend.clone()); let reader = StreamReader::new(Box::pin(source)); let mut chunker = fastcdc::v2020::AsyncStreamCDC::new( @@ -973,11 +1035,35 @@ impl DedupService { let mut session_seen: HashSet = HashSet::new(); let mut pending: Vec<(String, Bytes)> = Vec::new(); let mut pending_bytes: usize = 0; + // Depth-1 settle pipeline: batch N settles on a spawned task while + // the loop keeps reading/chunking/hashing batch N+1 from the source + // — the inline shape froze the reader (and the client's socket) for + // every settle (benches/INGEST-OVERLAP.md). The task records into + // the guard's shared state under its lock, so rollback stays exact + // even if this future is dropped mid-settle. + let mut in_flight: Option>> = None; + + /// Await the previous batch's settle, mapping panics/aborts to a + /// domain error so both are compensated identically. + async fn join_settle( + handle: tokio::task::JoinHandle>, + ) -> Result<(), DomainError> { + match handle.await { + Ok(res) => res, + Err(e) => Err(DomainError::internal_error( + "Dedup", + format!("Chunk settle task failed: {e}"), + )), + } + } while let Some(item) = chunk_stream.next().await { let chunk = match item { Ok(chunk) => chunk, Err(e) => { + if let Some(handle) = in_flight.take() { + let _ = join_settle(handle).await; + } guard.rollback().await; return Err(DomainError::internal_error( "Dedup", @@ -1000,7 +1086,26 @@ impl DedupService { pending.push((hash, Bytes::from(data))); if pending.len() >= Self::FLUSH_MAX_CHUNKS || pending_bytes >= Self::FLUSH_MAX_BYTES { - if let Err(e) = self.flush_pending(&mut guard, &mut pending).await { + if let Some(handle) = in_flight.take() + && let Err(e) = join_settle(handle).await + { + guard.rollback().await; + return Err(e); + } + let batch = std::mem::take(&mut pending); + let handle = tokio::spawn(Self::settle_batch( + self.pool.clone(), + self.backend.clone(), + guard.state.clone(), + batch, + )); + // Bench/ops escape hatch: OXICLOUD_INGEST_OVERLAP=0 + // reproduces the old inline-settle behaviour (await the + // batch before reading on) — used by + // benches/INGEST-OVERLAP.md for an in-binary A/B. + if ingest_overlap_enabled() { + in_flight = Some(handle); + } else if let Err(e) = join_settle(handle).await { guard.rollback().await; return Err(e); } @@ -1009,7 +1114,20 @@ impl DedupService { } } - if let Err(e) = self.flush_pending(&mut guard, &mut pending).await { + if let Some(handle) = in_flight.take() + && let Err(e) = join_settle(handle).await + { + guard.rollback().await; + return Err(e); + } + if let Err(e) = Self::settle_batch( + self.pool.clone(), + self.backend.clone(), + guard.state.clone(), + std::mem::take(&mut pending), + ) + .await + { guard.rollback().await; return Err(e); } @@ -1018,10 +1136,15 @@ impl DedupService { // One batched fsync sweep (no-op for remote backends, durable on // PUT), then one batched INSERT. A crash before the INSERT leaves // only unreferenced files; never a row pointing at unsynced bytes. - if !guard.written.is_empty() { - let new_hashes: Vec = guard.written.iter().map(|(h, _)| h.clone()).collect(); - let new_sizes: Vec = guard.written.iter().map(|(_, s)| *s).collect(); - + // No settle is in flight past this point — the lock is uncontended. + let (new_hashes, new_sizes): (Vec, Vec) = { + let st = guard.state.lock().await; + ( + st.written.iter().map(|(h, _)| h.clone()).collect(), + st.written.iter().map(|(_, s)| *s).collect(), + ) + }; + if !new_hashes.is_empty() { if let Err(e) = self.backend.sync_blobs(&new_hashes).await { guard.rollback().await; return Err(e); @@ -1047,7 +1170,7 @@ impl DedupService { } } - let newly_written = guard.written.len(); + let newly_written = new_hashes.len(); guard.disarm(); Ok(ChunkIngestOutcome { @@ -1061,18 +1184,23 @@ impl DedupService { /// Settle one batch of distinct in-RAM chunks against PG + the backend. /// - /// Successfully pinned hashes and written chunks are recorded on the - /// guard as they happen, so a failure mid-batch leaves nothing - /// untracked for rollback. - async fn flush_pending( - &self, - guard: &mut IngestGuard, - pending: &mut Vec<(String, Bytes)>, + /// Static (no `&self`) so the ingest loop can run it on a spawned task + /// and keep consuming the source stream while the batch settles — the + /// inline shape stalled the reader for the whole settle every 8 MiB + /// (benches/INGEST-OVERLAP.md). The shared-state lock is held for the + /// entire batch: pinned hashes and written chunks are recorded + /// progressively under it, so a failure (or a rollback racing this + /// settle) leaves nothing untracked. + async fn settle_batch( + pool: Arc, + backend: Arc, + state: Arc>, + batch: Vec<(String, Bytes)>, ) -> Result<(), DomainError> { - if pending.is_empty() { + if batch.is_empty() { return Ok(()); } - let batch = std::mem::take(pending); + let mut guard = state.lock().await; let hashes: Vec = batch.iter().map(|(h, _)| h.clone()).collect(); // Pin-or-classify in one statement: rows that exist take this @@ -1084,7 +1212,7 @@ impl DedupService { RETURNING hash", ) .bind(&hashes) - .fetch_all(self.pool.as_ref()) + .fetch_all(pool.as_ref()) .await .map_err(|e| { DomainError::internal_error("Dedup", format!("Failed to pin existing chunks: {e}")) @@ -1106,7 +1234,6 @@ impl DedupService { // Unsynced writes — durability comes from the single end-of-stream // sweep, before any PG row references these chunks. - let backend = self.backend.clone(); let results: Vec> = stream::iter(to_write) .map(|(hash, data)| { let backend = backend.clone(); diff --git a/src/infrastructure/services/nextcloud_chunked_upload_service.rs b/src/infrastructure/services/nextcloud_chunked_upload_service.rs index f305fa19..1ddd7a1c 100644 --- a/src/infrastructure/services/nextcloud_chunked_upload_service.rs +++ b/src/infrastructure/services/nextcloud_chunked_upload_service.rs @@ -1,24 +1,84 @@ use std::path::PathBuf; +use std::time::Duration; use tokio::fs; use crate::common::errors::{DomainError, Result}; +/// In-RAM running byte counter per upload session (`user/upload_id` → +/// bytes accepted so far). The per-chunk quota gate used to recompute +/// this by listing the whole session directory and stat-ing every chunk +/// on EVERY chunk PUT — O(k) stats for chunk k, O(N²/2) over an upload +/// (~500k stats for a 10 GB / 1000-chunk upload). The counter makes the +/// gate O(1); a cache miss (process restart, eviction) lazily rebuilds +/// from the directory listing, so crash-correctness is unchanged +/// (benches/NC-CHUNK-GATE.md). Sessions are forgotten on cleanup; the +/// TTL reaps counters for sessions the client abandoned. +fn build_session_bytes_cache() -> moka::sync::Cache { + moka::sync::Cache::builder() + .max_capacity(100_000) + .time_to_idle(Duration::from_secs(24 * 3600)) + .build() +} + #[derive(Clone)] pub struct NextcloudChunkedUploadService { pub base_dir: PathBuf, + /// See [`build_session_bytes_cache`]. Cloning the service shares the + /// counter (moka `Cache` clones are handles to the same store). + session_bytes: moka::sync::Cache, } impl NextcloudChunkedUploadService { pub fn new(base_dir: PathBuf) -> Self { - Self { base_dir } + Self { + base_dir, + session_bytes: build_session_bytes_cache(), + } } pub fn new_stub() -> Self { Self { base_dir: PathBuf::from("./storage/.uploads/nextcloud"), + session_bytes: build_session_bytes_cache(), } } + fn bytes_key(user: &str, upload_id: &str) -> String { + format!("{user}/{upload_id}") + } + + /// Session bytes accepted so far, if the counter is warm. + /// `None` = rebuild from the directory listing and call + /// [`Self::set_session_bytes`]. + pub fn cached_session_bytes(&self, user: &str, upload_id: &str) -> Option { + self.session_bytes.get(&Self::bytes_key(user, upload_id)) + } + + /// Seed / overwrite the session counter (post-rebuild or on MKCOL). + pub fn set_session_bytes(&self, user: &str, upload_id: &str, bytes: u64) { + self.session_bytes + .insert(Self::bytes_key(user, upload_id), bytes); + } + + /// Add an accepted chunk's bytes to the counter (no-op when cold — + /// the next gate rebuilds from disk). Two racing PUTs on one session + /// could drop an increment; the counter is a gate hint, and the + /// MOVE-time quota check stays authoritative. + pub fn bump_session_bytes(&self, user: &str, upload_id: &str, delta: u64) { + let key = Self::bytes_key(user, upload_id); + if let Some(current) = self.session_bytes.get(&key) { + self.session_bytes + .insert(key, current.saturating_add(delta)); + } + } + + /// Drop the counter (session cleanup, or a chunk overwrite made the + /// running total untrustworthy — rebuilt lazily on next use). + pub fn forget_session_bytes(&self, user: &str, upload_id: &str) { + self.session_bytes + .invalidate(&Self::bytes_key(user, upload_id)); + } + /// Validate that a path component contains no traversal characters. fn validate_path_component(name: &str, label: &str) -> Result<()> { if name.is_empty() @@ -48,6 +108,7 @@ impl NextcloudChunkedUploadService { fs::create_dir_all(&session_dir) .await .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; + self.set_session_bytes(user, upload_id, 0); Ok(()) } @@ -97,9 +158,17 @@ impl NextcloudChunkedUploadService { data: &[u8], ) -> Result<()> { let chunk_path = self.safe_chunk_path(user, upload_id, chunk_name)?; + let overwrite = fs::metadata(&chunk_path).await.is_ok(); fs::write(&chunk_path, data) .await - .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string())) + .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; + if overwrite { + // Retried chunk — running total is stale; rebuild lazily. + self.forget_session_bytes(user, upload_id); + } else { + self.bump_session_bytes(user, upload_id, data.len() as u64); + } + Ok(()) } /// List the session's chunk files in assembly (numeric) order. @@ -146,6 +215,7 @@ impl NextcloudChunkedUploadService { .await .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; } + self.forget_session_bytes(user, upload_id); Ok(()) } diff --git a/src/infrastructure/services/zip_service.rs b/src/infrastructure/services/zip_service.rs index b84087d5..d5ef1fcf 100644 --- a/src/infrastructure/services/zip_service.rs +++ b/src/infrastructure/services/zip_service.rs @@ -44,8 +44,9 @@ impl From for DomainError { } } -/// Type alias for the fully-async ZIP writer backed by a buffered tokio file. -type AsyncZipWriter = ZipFileWriter>>; +/// Fully-async ZIP writer over any buffered tokio sink (temp file for the +/// legacy path, one half of a `tokio::io::duplex` for the streaming path). +type AsyncZipWriter = ZipFileWriter>>; /// One planned archive entry, in final ZIP order. enum ZipPlanEntry { @@ -119,6 +120,110 @@ impl ZipService { folder_id: &str, folder_name: &str, ) -> Result { + let plan = self.plan_archive(folder_id, folder_name).await?; + + // ── Open the temp file + ZIP writer ────────────────────────────── + let temp = NamedTempFile::new().map_err(ZipError::IoError)?; + let tokio_file = tokio::fs::File::create(temp.path()) + .await + .map_err(ZipError::IoError)?; + + let (tx, mut rx) = tokio::sync::mpsc::channel::(PREFETCH_BUFFER_CHUNKS); + let _prefetcher = tokio::spawn(Self::prefetch_files( + self.file_service.clone(), + Self::planned_file_ids(&plan), + tx, + )); + Self::write_archive(tokio_file, &plan, &mut rx).await?; + + Ok(temp) + } + + /// Streaming variant: the archive bytes are produced on a spawned task + /// and yielded as they are written — the client's first byte arrives + /// after the first entry starts, not after the whole archive has been + /// built (the temp-file variant's time-to-first-byte grows with folder + /// size; benches/ZIP-STREAM.md). The plan phase still runs inline so + /// planning errors surface as proper HTTP errors; a blob-read error + /// mid-archive can only truncate the stream (no central directory → + /// clients detect the corrupt archive), which is the standard tradeoff + /// for streamed ZIPs. + pub async fn create_folder_zip_stream( + &self, + folder_id: &str, + folder_name: &str, + ) -> Result> + Send + use<>> { + let plan = self.plan_archive(folder_id, folder_name).await?; + + let (writer, reader) = tokio::io::duplex(256 * 1024); + let (tx, mut rx) = tokio::sync::mpsc::channel::(PREFETCH_BUFFER_CHUNKS); + let _prefetcher = tokio::spawn(Self::prefetch_files( + self.file_service.clone(), + Self::planned_file_ids(&plan), + tx, + )); + tokio::spawn(async move { + if let Err(e) = Self::write_archive(writer, &plan, &mut rx).await { + // Dropping the writer EOFs the reader early — the truncated + // archive has no central directory, so clients flag it. + warn!("Streaming ZIP aborted mid-archive: {e}"); + } + }); + + Ok(tokio_util::io::ReaderStream::new(reader)) + } + + /// File ids of the plan, in archive order (the prefetcher's read list). + fn planned_file_ids(plan: &[ZipPlanEntry]) -> Vec { + plan.iter() + .filter_map(|entry| match entry { + ZipPlanEntry::File { file_id, .. } => Some(file_id.clone()), + ZipPlanEntry::Dir(_) => None, + }) + .collect() + } + + /// Write every planned entry through a buffered ZIP writer over `sink`, + /// then finalize (central directory + flush). Shared by the temp-file + /// and streaming variants. + async fn write_archive( + sink: W, + plan: &[ZipPlanEntry], + rx: &mut tokio::sync::mpsc::Receiver, + ) -> Result<()> { + let buf_writer = BufWriter::with_capacity(256 * 1024, sink); + let mut zip = ZipFileWriter::with_tokio(buf_writer); + + for entry in plan { + match entry { + ZipPlanEntry::Dir(zip_dir) => { + let dir_entry = + ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored); + match zip.write_entry_whole(dir_entry, &[]).await { + Ok(()) => debug!("Folder added to ZIP: {}", zip_dir), + Err(e) => { + warn!("Could not add folder entry (may already exist): {}", e); + } + } + } + ZipPlanEntry::File { + zip_path, + compression, + .. + } => { + Self::write_prefetched_file(&mut zip, zip_path, *compression, rx).await?; + } + } + } + + let mut compat_writer = zip.close().await.map_err(ZipError::AsyncZipError)?; + compat_writer.close().await.map_err(ZipError::IoError)?; + Ok(()) + } + + /// Resolve the folder, fetch its subtree (2 bulk queries) and lay out + /// the archive entries in final ZIP order. + async fn plan_archive(&self, folder_id: &str, folder_name: &str) -> Result> { info!( "Creating ZIP for folder: {} (ID: {})", folder_name, folder_id @@ -200,61 +305,7 @@ impl ZipService { } } - // ── 5. Open the temp file + ZIP writer ─────────────────────────── - let temp = NamedTempFile::new().map_err(ZipError::IoError)?; - let tokio_file = tokio::fs::File::create(temp.path()) - .await - .map_err(ZipError::IoError)?; - let buf_writer = BufWriter::with_capacity(256 * 1024, tokio_file); - let mut zip = ZipFileWriter::with_tokio(buf_writer); - - // ── 6. Write entries: 2-stage pipeline ─────────────────────────── - // The prefetch task reads blob streams for the planned files, in - // order, ahead of the writer — the next file's blob-store latency - // overlaps the current file's deflate. If the writer bails out, - // dropping the receiver makes the prefetcher's next send fail and - // it stops on its own. - let file_ids: Vec = plan - .iter() - .filter_map(|entry| match entry { - ZipPlanEntry::File { file_id, .. } => Some(file_id.clone()), - ZipPlanEntry::Dir(_) => None, - }) - .collect(); - let (tx, mut rx) = tokio::sync::mpsc::channel::(PREFETCH_BUFFER_CHUNKS); - let _prefetcher = tokio::spawn(Self::prefetch_files( - self.file_service.clone(), - file_ids, - tx, - )); - - for entry in &plan { - match entry { - ZipPlanEntry::Dir(zip_dir) => { - let dir_entry = - ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored); - match zip.write_entry_whole(dir_entry, &[]).await { - Ok(()) => debug!("Folder added to ZIP: {}", zip_dir), - Err(e) => { - warn!("Could not add folder entry (may already exist): {}", e); - } - } - } - ZipPlanEntry::File { - zip_path, - compression, - .. - } => { - Self::write_prefetched_file(&mut zip, zip_path, *compression, &mut rx).await?; - } - } - } - - // ── 7. Finalize ────────────────────────────────────────────────── - let mut compat_writer = zip.close().await.map_err(ZipError::AsyncZipError)?; - compat_writer.close().await.map_err(ZipError::IoError)?; - - Ok(temp) + Ok(plan) } /// Prefetch stage: streams each planned file's content from the blob @@ -302,8 +353,8 @@ impl ZipService { /// (`Stored` for already-compressed media, `Deflate` otherwise — see /// `entry_compression`). Peak memory stays bounded by the channel, /// independent of individual file sizes. - async fn write_prefetched_file( - zip: &mut AsyncZipWriter, + async fn write_prefetched_file( + zip: &mut AsyncZipWriter, zip_path: &str, compression: Compression, rx: &mut tokio::sync::mpsc::Receiver, diff --git a/src/interfaces/api/handlers/delta_upload_handler.rs b/src/interfaces/api/handlers/delta_upload_handler.rs index 6912d37f..7d2fa471 100644 --- a/src/interfaces/api/handlers/delta_upload_handler.rs +++ b/src/interfaces/api/handlers/delta_upload_handler.rs @@ -19,7 +19,7 @@ use axum::{ response::{IntoResponse, Response}, }; use bytes::{Buf, Bytes, BytesMut}; -use futures::Stream; +use futures::{Stream, TryStreamExt}; use std::sync::Arc; use tokio_stream::StreamExt; @@ -343,17 +343,36 @@ pub async fn delta_download_chunks( // 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. + // is bounded by `read_prefetch` open streams (their first frame), + // independent of batch size. + // + // `buffered(read_prefetch)` overlaps the NEXT chunk's open with the + // current chunk's drain — the same combinator/tuning as the main CDC + // download path (benches/BLOB-PREFETCH.md). The old per-chunk await + // paid every open's full round-trip serially: on an object-store + // backend a 64-chunk batch at ~30 ms first-byte cost ~1.9 s of pure + // latency. Frames still arrive strictly in request order. + let prefetch = service.read_prefetch().max(1); + let svc = service.clone(); + // `futures::StreamExt` spelled out — this handler imports + // `tokio_stream::StreamExt`, whose `map` adapter lacks `buffered`. + let opened = futures::StreamExt::map(futures::stream::iter(ordered), move |(hash, size)| { + let svc = svc.clone(); + async move { + let chunk = svc + .chunk_stream(&hash) + .await + .map_err(std::io::Error::other)?; + let header = futures::stream::once(async move { + Ok::(Bytes::copy_from_slice(&(size as u32).to_be_bytes())) + }); + Ok::<_, std::io::Error>(futures::StreamExt::chain(header, chunk)) + } + }); 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?; - } - } - }); + Box::pin(TryStreamExt::try_flatten(futures::StreamExt::buffered( + opened, prefetch, + ))); Ok(Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index 7347f415..f1095780 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -12,7 +12,7 @@ use std::collections::HashMap; use utoipa::ToSchema; use crate::application::ports::file_ports::{ - FileManagementUseCase, FileRetrievalUseCase, FileUploadUseCase, + FileManagementUseCase, FileRetrievalUseCase, FileUploadUseCase, RangeContent, }; use crate::application::ports::storage_ports::{FileReadPort, StorageUsagePort}; use crate::application::ports::thumbnail_ports::ThumbnailPort; @@ -713,10 +713,19 @@ impl FileHandler { Self::content_disposition(&file_dto.name, &file_dto.mime_type, ¶ms); match retrieval - .get_file_range_stream_with_perms(&id, auth_user.id, start, Some(end + 1)) + .get_file_range_preloaded_with_perms( + &file_dto, + auth_user.id, + start, + Some(end + 1), + ) .await { - Ok(stream) => { + Ok(content) => { + let body = match content { + RangeContent::Bytes(b) => Body::from(b), + RangeContent::Stream(s) => Body::from_stream(Box::into_pin(s)), + }; return Response::builder() .status(StatusCode::PARTIAL_CONTENT) .header(header::CONTENT_TYPE, &*file_dto.mime_type) @@ -732,7 +741,7 @@ impl FileHandler { header::CACHE_CONTROL, "private, max-age=3600, must-revalidate", ) - .body(Body::from_stream(Box::into_pin(stream))) + .body(body) .unwrap() .into_response(); } diff --git a/src/interfaces/api/handlers/folder_handler.rs b/src/interfaces/api/handlers/folder_handler.rs index d2957ded..084d0048 100644 --- a/src/interfaces/api/handlers/folder_handler.rs +++ b/src/interfaces/api/handlers/folder_handler.rs @@ -6,7 +6,6 @@ use axum::{ }; use std::collections::HashMap; use std::sync::Arc; -use tokio_util::io::ReaderStream; use crate::application::dtos::display_helpers::{ category_for, format_file_size, icon_class_for, icon_special_class_for, @@ -238,53 +237,28 @@ impl FolderHandler { } }; - // Create the ZIP archive (written to a temp file, O(1) RAM) - match zip_service.create_folder_zip(&id, &folder.name).await { - Ok(temp_file) => { - // Get the file size for Content-Length - let file_size = match temp_file.as_file().metadata() { - Ok(m) => m.len(), - Err(e) => { - tracing::error!("Error reading temp file metadata: {}", e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({ - "error": "Error creating ZIP file" - })), - ) - .into_response(); - } - }; - - tracing::info!("ZIP file created successfully, size: {} bytes", file_size); - - // Split the NamedTempFile into the already-open std File - // and the TempPath (auto-deletes on drop). This reuses - // the existing fd instead of opening a second one. - let (std_file, temp_path) = temp_file.into_parts(); - let tokio_file = tokio::fs::File::from_std(std_file); - - // Stream the file to the client in chunks - let stream = ReaderStream::new(tokio_file); + // Stream the archive as it is built — the first byte reaches + // the client after the first entry, not after the whole ZIP + // exists on disk (benches/ZIP-STREAM.md). No Content-Length: + // the final size isn't known up front (chunked encoding). + match zip_service + .create_folder_zip_stream(&id, &folder.name) + .await + { + Ok(stream) => { let body = axum::body::Body::from_stream(stream); // Setup headers for download let filename = format!("{}.zip", folder.name); let content_disposition = format!("attachment; filename=\"{}\"", filename); - let mut response = Response::builder() + Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/zip") .header(header::CONTENT_DISPOSITION, content_disposition) - .header(header::CONTENT_LENGTH, file_size) .body(body) - .unwrap(); - - // Keep TempPath alive in the response extensions so the - // file is only deleted AFTER the body stream finishes. - response.extensions_mut().insert(Arc::new(temp_path)); - - response.into_response() + .unwrap() + .into_response() } Err(err) => { tracing::error!("Error creating ZIP file: {}", err); diff --git a/src/interfaces/api/handlers/share_handler.rs b/src/interfaces/api/handlers/share_handler.rs index 2bda63b9..c3801e84 100644 --- a/src/interfaces/api/handlers/share_handler.rs +++ b/src/interfaces/api/handlers/share_handler.rs @@ -13,6 +13,7 @@ use serde::Deserialize; use serde_json::json; use utoipa::ToSchema; +use crate::application::ports::file_ports::RangeContent; use crate::application::services::share_browse_service::ZipTarget; use crate::application::services::share_service::ShareService; use crate::infrastructure::services::share_unlock_cookie; @@ -30,7 +31,6 @@ use crate::{ interfaces::errors::AppError, interfaces::middleware::auth::AuthUser, }; -use tokio_util::io::ReaderStream; fn unlock_jwt_from_headers(headers: &HeaderMap, share_token: &str) -> Option { headers @@ -438,10 +438,14 @@ async fn serve_share_file( let length = end - start + 1; match retrieval - .get_file_range_stream(file_id, start, Some(end + 1)) + .get_file_range_preloaded(&file_dto, start, Some(end + 1)) .await { - Ok(stream) => { + Ok(content) => { + let body = match content { + RangeContent::Bytes(b) => Body::from(b), + RangeContent::Stream(s) => Body::from_stream(Box::into_pin(s)), + }; return Response::builder() .status(StatusCode::PARTIAL_CONTENT) .header(header::CONTENT_TYPE, &*mime) @@ -458,7 +462,7 @@ async fn serve_share_file( "private, max-age=3600, must-revalidate", ) .header(header::VARY, "Cookie, Range") - .body(Body::from_stream(Box::into_pin(stream))) + .body(body) .unwrap() .into_response(); } @@ -730,30 +734,19 @@ async fn serve_share_zip( Err(err) => return share_browse_error_response(err), }; - let temp_file = match zip_service - .create_folder_zip(&target.folder_id, &target.display_name) + // Streamed archive: first byte after the first entry, not after the + // whole ZIP is built (benches/ZIP-STREAM.md). No Content-Length. + let stream = match zip_service + .create_folder_zip_stream(&target.folder_id, &target.display_name) .await { - Ok(f) => f, + Ok(s) => s, Err(err) => { tracing::error!("share zip: create_folder_zip failed: {}", err); return AppError::internal_error(format!("ZIP creation failed: {}", err)) .into_response(); } }; - - let file_size = match temp_file.as_file().metadata() { - Ok(m) => m.len(), - Err(e) => { - tracing::error!("share zip: temp metadata failed: {}", e); - return AppError::internal_error("ZIP creation failed").into_response(); - } - }; - - // Reuse the existing fd: split off the std::File and the TempPath. - let (std_file, temp_path) = temp_file.into_parts(); - let tokio_file = tokio::fs::File::from_std(std_file); - let stream = ReaderStream::new(tokio_file); let body = Body::from_stream(stream); let disposition = build_content_disposition( @@ -762,17 +755,12 @@ async fn serve_share_zip( false, ); - let mut response = Response::builder() + Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/zip") .header(header::CONTENT_DISPOSITION, disposition) - .header(header::CONTENT_LENGTH, file_size) .header(header::CACHE_CONTROL, "private, no-store") .header(header::VARY, "Cookie") .body(body) - .unwrap(); - - // Keep TempPath alive until the body finishes streaming. - response.extensions_mut().insert(Arc::new(temp_path)); - response + .unwrap() } diff --git a/src/interfaces/nextcloud/uploads_handler.rs b/src/interfaces/nextcloud/uploads_handler.rs index 85089da8..194b9c8a 100644 --- a/src/interfaces/nextcloud/uploads_handler.rs +++ b/src/interfaces/nextcloud/uploads_handler.rs @@ -74,6 +74,14 @@ async fn session_bytes_so_far( username: &str, upload_id: &str, ) -> Result { + // Warm path: O(1) in-RAM counter maintained by the PUT handler and + // the service (seeded on MKCOL, dropped on cleanup/overwrite). The + // directory walk below only runs cold (restart / eviction) — the old + // shape ran it on EVERY chunk PUT: O(k) stats for chunk k, O(N²/2) + // over the upload (benches/NC-CHUNK-GATE.md). + if let Some(bytes) = nc.chunked_uploads.cached_session_bytes(username, upload_id) { + return Ok(bytes); + } let listing = nc .chunked_uploads .list_chunks(username, upload_id) @@ -85,7 +93,10 @@ async fn session_bytes_so_far( // chunk after MKCOL (race-tolerant). return Ok(0); }; - Ok(listing.chunks.iter().map(|c| c.size).sum()) + let total = listing.chunks.iter().map(|c| c.size).sum(); + nc.chunked_uploads + .set_session_bytes(username, upload_id, total); + Ok(total) } /// Dispatch Nextcloud chunked upload WebDAV requests. @@ -314,11 +325,21 @@ async fn handle_put_chunk( .map_err(|e| AppError::bad_request(format!("Invalid chunk path: {}", e)))?; let max_chunk = state.core.config.storage.chunk_max_bytes; + // A re-PUT of an existing chunk (client retry) makes the running + // session counter stale — drop it so the next gate rebuilds from disk. + let overwrite = tokio::fs::metadata(&chunk_path).await.is_ok(); // No client-side integrity contract on the NC chunked surface — the // NC desktop client validates the assembled-file ETag against the // server-side `oc:checksums` after MOVE. So we skip per-chunk // hashing here (peak heap stays at ~one HTTP frame). - stream_body_to_path(req.into_body(), &chunk_path, max_chunk, None).await?; + let streamed = stream_body_to_path(req.into_body(), &chunk_path, max_chunk, None).await?; + if overwrite { + nc.chunked_uploads + .forget_session_bytes(&user.username, upload_id); + } else { + nc.chunked_uploads + .bump_session_bytes(&user.username, upload_id, streamed.bytes_written); + } Ok(Response::builder() .status(StatusCode::CREATED) diff --git a/src/interfaces/range_requests.rs b/src/interfaces/range_requests.rs index b29296f8..66ea36fa 100644 --- a/src/interfaces/range_requests.rs +++ b/src/interfaces/range_requests.rs @@ -13,7 +13,7 @@ use http_range_header::parse_range_header; use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; -use crate::application::ports::file_ports::FileRetrievalUseCase; +use crate::application::ports::file_ports::RangeContent; use crate::application::services::file_retrieval_service::FileRetrievalService; /// `If-None-Match` short-circuit: returns a `304 Not Modified` response @@ -71,24 +71,32 @@ pub async fn range_response( let end = *range.end(); let range_length = end - start + 1; + // Cache-aware: sub-threshold files already in the RAM content cache are + // answered with a zero-copy Bytes slice — no PG, no disk (benches/RANGE-CACHE.md). match retrieval - .get_file_range_stream(&file.id, start, Some(end + 1)) + .get_file_range_preloaded(file, start, Some(end + 1)) .await { - Ok(stream) => Some( - Response::builder() - .status(StatusCode::PARTIAL_CONTENT) - .header(header::CONTENT_TYPE, &*file.mime_type) - .header(header::CONTENT_LENGTH, range_length) - .header( - header::CONTENT_RANGE, - format!("bytes {}-{}/{}", start, end, file.size), - ) - .header(header::ACCEPT_RANGES, "bytes") - .header(header::ETAG, etag) - .body(Body::from_stream(Box::into_pin(stream))) - .unwrap(), - ), + Ok(content) => { + let body = match content { + RangeContent::Bytes(b) => Body::from(b), + RangeContent::Stream(s) => Body::from_stream(Box::into_pin(s)), + }; + Some( + Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(header::CONTENT_TYPE, &*file.mime_type) + .header(header::CONTENT_LENGTH, range_length) + .header( + header::CONTENT_RANGE, + format!("bytes {}-{}/{}", start, end, file.size), + ) + .header(header::ACCEPT_RANGES, "bytes") + .header(header::ETAG, etag) + .body(body) + .unwrap(), + ) + } Err(err) => { tracing::error!("Error creating range stream: {}", err); None // fall through to the full download