diff --git a/docs/delta-upload-protocol.md b/docs/delta-upload-protocol.md index 31fd58ff..171e1a1d 100644 --- a/docs/delta-upload-protocol.md +++ b/docs/delta-upload-protocol.md @@ -12,15 +12,20 @@ Editing a few bytes of a 500 MB file re-uploads ~1 MiB instead of ## Who can use it -Any authenticated API client. The OxiCloud web frontend adopts it in a -later phase; generic WebDAV/NextCloud clients cannot (their protocols -have no delta concept) — they keep uploading full bytes, and the server -keeps deduplicating those on write. +Any authenticated API client. The OxiCloud web frontend uses it +automatically for files ≥ 8 MiB (`features/files/deltaUpload.js` + +`workers/deltaWorker.js`, chunking with the vendored WASM build of the +server's own FastCDC+BLAKE3 crates, falling back to a plain byte upload +on any failure). Generic WebDAV/NextCloud clients cannot (their +protocols have no delta concept) — they keep uploading full bytes, and +the server keeps deduplicating those on write. Chunk boundaries are the **client's choice**: matching the server's -FastCDC parameters maximizes cross-version sharing, but any split with -chunks of 1 byte … 1 MiB is valid — correctness is guaranteed by -server-side verification, not by the chunking scheme. +FastCDC parameters (64 KB / 256 KB / 1 MiB, as the bundled WASM module +does) maximizes cross-version sharing — including against versions that +entered through plain byte uploads — but any split with chunks of +1 byte … 1 MiB is valid; correctness is guaranteed by server-side +verification, not by the chunking scheme. ## The three steps diff --git a/static/js/features/files/deltaUpload.js b/static/js/features/files/deltaUpload.js new file mode 100644 index 00000000..f6a17d14 --- /dev/null +++ b/static/js/features/files/deltaUpload.js @@ -0,0 +1,160 @@ +/** + * OxiCloud - Delta upload ("upload only what changed"). + * + * Main-thread orchestrator for `workers/deltaWorker.js`, which runs the + * whole client side of the delta protocol off the UI thread: FastCDC + * chunking + BLAKE3 (the same WASM crate and parameters as the server, + * so boundaries match bit for bit), per-batch negotiation, upload of + * only the missing chunks, and the commit. + * + * This SUBSUMES the previous whole-file instant upload: a fully known + * file negotiates to "nothing missing" and the commit short-circuits on + * possession of the file hash — same zero-byte outcome, one pipeline. + * + * Performance posture: + * - Stages overlap inside the worker (hash ‖ negotiate ‖ upload), so + * wall-clock approaches max(hash, upload) instead of their sum. + * - RAM stays flat: 8 MiB read slices; chunk bytes are re-sliced from + * the File at upload time, never hoarded. + * - Files below {@link DELTA_UPLOAD_MIN_SIZE} skip the pipeline: the + * round-trips cost more than the bytes. + * - Any failure falls back silently to the normal byte upload — delta + * is an optimization, never a gate. + */ + +import { getCsrfToken } from '../../core/csrf.js'; + +/** + * Files smaller than this upload normally: hashing + negotiation + * round-trips outweigh the transfer. + */ +export const DELTA_UPLOAD_MIN_SIZE = 8 * 1024 * 1024; + +// Absolute URL on purpose — works in dev and in the release IIFE bundle +// (same pattern as the pdf.js loader in thumbnail.js). +const DELTA_WORKER_URL = '/js/workers/deltaWorker.js'; + +/** Budget: 120 s base + 90 s per GB (hashing + uploading the delta). */ +const DELTA_TIMEOUT_BASE_MS = 120000; +const DELTA_TIMEOUT_PER_GB_MS = 90000; + +/** + * `false` once the environment proved unable to run the worker/WASM — + * later files skip straight to the byte upload. `null` = not yet known. + * @type {boolean | null} + */ +let _deltaUploadUsable = null; + +/** + * Result contract shared with the uploaders' `UploadAnswer`, plus the + * bandwidth accounting the UI surfaces. + * @typedef {Object} DeltaUploadAnswer + * @property {boolean} ok + * @property {any} [data] FileDto on success + * @property {string} [errorMsg] + * @property {boolean} [isQuotaError] + * @property {number} [savedBytes] bytes NOT transferred thanks to dedup + */ + +/** + * Try to upload `file` through the delta protocol. + * + * Resolves `null` whenever the plain byte upload should proceed (file too + * small, environment unusable, any transport/protocol failure). Resolves + * a {@link DeltaUploadAnswer} when the outcome is conclusive — success, + * quota exceeded, or a name conflict a byte upload would also hit. + * + * @param {File} file + * @param {string | null | undefined} folderId + * @param {(pct: number) => void} [onProgress] 0-99 while transferring + * @returns {Promise} + */ +export function tryDeltaUpload(file, folderId, onProgress) { + if (!folderId || file.size < DELTA_UPLOAD_MIN_SIZE || _deltaUploadUsable === false || typeof Worker === 'undefined') { + return Promise.resolve(null); + } + + return new Promise((resolve) => { + /** @type {Worker} */ + let worker; + try { + worker = new Worker(DELTA_WORKER_URL, { type: 'module' }); + } catch (_) { + _deltaUploadUsable = false; + resolve(null); + return; + } + + const sizeGB = file.size / (1024 * 1024 * 1024); + const timeoutMs = DELTA_TIMEOUT_BASE_MS + Math.ceil(sizeGB) * DELTA_TIMEOUT_PER_GB_MS; + + let savedBytes = 0; + + /** @param {DeltaUploadAnswer | null} answer */ + const settle = (answer) => { + clearTimeout(timer); + worker.terminate(); + resolve(answer); + }; + const timer = setTimeout(() => settle(null), timeoutMs); + + worker.onmessage = (event) => { + const msg = /** @type {any} */ (event.data); + if (msg.type === 'progress') { + savedBytes = msg.reusedBytes; + if (onProgress && msg.totalBytes > 0) { + const pct = Math.min(99, Math.round((100 * (msg.reusedBytes + msg.uploadedBytes)) / msg.totalBytes)); + onProgress(pct); + } + return; + } + if (msg.type === 'fallback') { + settle(null); + return; + } + if (msg.type === 'done') { + if (msg.status === 201 || msg.status === 200) { + settle({ ok: true, data: msg.body, savedBytes }); + return; + } + /** @type {string} */ + const errorMsg = msg.body?.message || msg.body?.error || `Delta upload failed (HTTP ${msg.status})`; + if (msg.status === 507) { + settle({ ok: false, isQuotaError: true, errorMsg }); + return; + } + if (msg.status === 409 && !msg.body?.still_missing) { + // Duplicate name — a byte upload would hit the same wall. + settle({ ok: false, errorMsg }); + return; + } + // still_missing exhausted, 4xx/5xx oddities: byte upload is + // the safe road (the server dedups it on write anyway). + settle(null); + } + }; + worker.onerror = () => { + // Worker script failed to load/parse — permanent environment trait. + _deltaUploadUsable = false; + settle(null); + }; + + worker.postMessage({ + file, + folderId, + name: file.name, + csrfToken: getCsrfToken() || '' + }); + }); +} + +/** + * Bilingual one-line summary for the bandwidth saved by a batch. + * @param {number} savedBytes + * @param {string} locale + * @returns {string} + */ +export function formatSavedSummary(savedBytes, locale) { + const mb = (savedBytes / (1024 * 1024)).toFixed(1); + return locale.startsWith('es') ? `Deduplicación: ${mb} MB no necesitaron subirse` : `Deduplication: ${mb} MB didn't need uploading`; +} diff --git a/static/js/features/files/fileOperations.js b/static/js/features/files/fileOperations.js index a4bc092d..f62b52d3 100644 --- a/static/js/features/files/fileOperations.js +++ b/static/js/features/files/fileOperations.js @@ -12,7 +12,7 @@ import { i18n } from '../../core/i18n.js'; import { notifications } from '../../core/notifications.js'; import { invalidateFolderMeta } from '../../model/filesModel.js'; import { triggerBrowserDownload } from '../../utils/download.js'; -import { tryInstantUpload } from './instantUpload.js'; +import { formatSavedSummary, tryDeltaUpload } from './deltaUpload.js'; /** * @typedef {Object} BatchResult @@ -392,6 +392,7 @@ const fileOps = { let uploadedCount = 0; let successCount = 0; let quotaStop = false; + let savedBytesTotal = 0; const targetFolderId = app.currentPath || app.userHomeFolderId; @@ -405,12 +406,19 @@ const fileOps = { if (quotaStop) return; const file = readableFiles[idx]; - // ── Instant upload: when the server already has this exact - // content for this user, register it by hash — zero bytes - // on the wire. Any miss/failure falls back to a byte upload. - /** @type {UploadAnswer | null} */ - let result = await tryInstantUpload(file, targetFolderId); + // ── Delta upload: chunk + hash locally (worker/WASM) and + // transfer only what the server doesn't already have for + // this user. Any miss/failure falls back to a byte upload. + /** @type {UploadAnswer & { savedBytes?: number } | null} */ + let result = await tryDeltaUpload(file, targetFolderId, (pct) => { + if (batchId) { + try { + notifications.updateFile(batchId, file.name, pct, 'uploading'); + } catch (_) {} + } + }); if (result) { + savedBytesTotal += result.savedBytes || 0; if (batchId) { try { notifications.updateFile(batchId, file.name, 100, result.ok ? 'done' : 'error'); @@ -492,6 +500,14 @@ const fileOps = { // All done this._finishUploadToast(successCount, totalFiles); + if (savedBytesTotal > 0 && notifications) { + notifications.addNotification({ + icon: 'fa-bolt', + iconClass: 'upload', + title: i18n?.getCurrentLocale?.()?.startsWith('es') ? 'Subida delta' : 'Delta upload', + text: formatSavedSummary(savedBytesTotal, i18n?.getCurrentLocale?.() || 'en') + }); + } // Refresh storage usage display try { @@ -643,6 +659,7 @@ const fileOps = { let uploadedCount = 0; let successCount = 0; let quotaStop = false; + let savedBytesTotal = 0; // ── Concurrent upload with limited parallelism ────────── // FIFOs are pre-caught by the 0-byte arrayBuffer guard, @@ -672,13 +689,14 @@ const fileOps = { const parentPath = parts.slice(0, -1).join('/'); const targetFolderId = folderMap.get(parentPath) || currentFolderId; - // ── Instant upload (zero bytes on the wire) ── + // ── Delta upload (only changed bytes on the wire) ── // Same fallback contract as uploadFiles: a null result // means "do the byte upload". The shared accounting // after this try block handles both outcomes. - const instant = await tryInstantUpload(file, targetFolderId); - if (instant) { - result = instant; + const delta = await tryDeltaUpload(file, targetFolderId); + if (delta) { + result = delta; + savedBytesTotal += delta.savedBytes || 0; } else { // ── FIFO/pipe guard (0-byte files only) ── // Named pipes (runit supervise/control) report size=0 @@ -773,6 +791,14 @@ const fileOps = { await Promise.all(workers); this._finishUploadToast(successCount, totalFiles); + if (savedBytesTotal > 0 && notifications) { + notifications.addNotification({ + icon: 'fa-bolt', + iconClass: 'upload', + title: i18n?.getCurrentLocale?.()?.startsWith('es') ? 'Subida delta' : 'Delta upload', + text: formatSavedSummary(savedBytesTotal, i18n?.getCurrentLocale?.() || 'en') + }); + } try { await refreshUserData(); diff --git a/static/js/features/files/instantUpload.js b/static/js/features/files/instantUpload.js deleted file mode 100644 index bda924e1..00000000 --- a/static/js/features/files/instantUpload.js +++ /dev/null @@ -1,177 +0,0 @@ -/** - * OxiCloud - Instant upload (zero-byte dedup upload) - * - * Before transferring a file's bytes, compute its BLAKE3 locally (in a - * worker, off the main thread) and ask the server whether the caller - * already owns that exact content (`GET /api/dedup/check/{hash}` — the - * check is user-scoped, never a global content oracle). On a hit, a - * single metadata call (`POST /api/files/by-hash`) registers the file - * with ZERO content bytes on the wire. - * - * Performance posture: - * - Hashing runs in a dedicated worker with WASM SIMD128 — the UI thread - * never blocks, RAM stays constant (8 MiB slices). - * - Files below {@link INSTANT_UPLOAD_MIN_SIZE} skip the whole dance: - * two extra round-trips cost more than just uploading them. - * - Any failure (no WASM support, worker error, server miss, races) - * falls back silently to the normal byte upload — instant upload is - * an optimization, never a gate. - */ - -import { getCsrfHeaders } from '../../core/csrf.js'; - -/** - * Files smaller than this upload normally: hashing + two round-trips - * outweigh the transfer. 8 MiB matches the chunked-upload threshold's - * order of magnitude. - */ -export const INSTANT_UPLOAD_MIN_SIZE = 8 * 1024 * 1024; - -// Absolute URL on purpose — works in dev and in the release IIFE bundle -// (same pattern as the pdf.js loader in thumbnail.js). -const HASH_WORKER_URL = '/js/workers/hashWorker.js'; - -/** Hashing budget: 60 s base + 30 s per GB (WASM SIMD does ~0.5-1 GB/s). */ -const HASH_TIMEOUT_BASE_MS = 60000; -const HASH_TIMEOUT_PER_GB_MS = 30000; - -/** - * `false` once the environment proved unable to run the worker/WASM - * (old browser, blocked worker) — later files skip straight to the byte - * upload instead of failing the same way again. `null` = not yet known. - * @type {boolean | null} - */ -let _instantUploadUsable = null; - -/** - * Hash a file in a one-shot worker. Resolves `null` on any failure — - * the caller falls back to a normal upload. - * @param {File} file - * @returns {Promise} - */ -function hashFileInWorker(file) { - return new Promise((resolve) => { - /** @type {Worker} */ - let worker; - try { - worker = new Worker(HASH_WORKER_URL, { type: 'module' }); - } catch (_) { - _instantUploadUsable = false; - resolve(null); - return; - } - - const sizeGB = file.size / (1024 * 1024 * 1024); - const timeoutMs = HASH_TIMEOUT_BASE_MS + Math.ceil(sizeGB) * HASH_TIMEOUT_PER_GB_MS; - - /** @param {string | null} hash */ - const settle = (hash) => { - clearTimeout(timer); - worker.terminate(); - resolve(hash); - }; - const timer = setTimeout(() => settle(null), timeoutMs); - - worker.onmessage = (event) => { - const data = /** @type {{ ok: boolean, hash?: string, error?: string }} */ (event.data); - if (!data.ok) { - // The worker ran but WASM failed (e.g. no SIMD128 support): - // a permanent environment property, don't retry per file. - _instantUploadUsable = false; - } - settle(data.ok && data.hash ? data.hash : null); - }; - worker.onerror = () => { - // Worker script failed to load/parse — permanent. - _instantUploadUsable = false; - settle(null); - }; - - worker.postMessage({ file }); - }); -} - -/** - * Ask the server whether the caller already owns content with this hash. - * @param {string} hash - * @returns {Promise} - */ -async function callerOwnsHash(hash) { - try { - const response = await fetch(`/api/dedup/check/${hash}`, { - headers: { 'Cache-Control': 'no-cache, no-store, must-revalidate' } - }); - if (!response.ok) return false; - const body = /** @type {import('../../core/types.js').HashCheckAnswer} */ (await response.json()); - return body.exists === true; - } catch (_) { - return false; - } -} - -/** - * Try to register `file` as a zero-byte instant upload. - * - * Returns `null` whenever the byte upload should proceed (file too - * small, environment unusable, hash miss, lost race, transient errors). - * Returns an upload-result object compatible with the uploaders' - * `UploadAnswer` shape when the attempt is conclusive — success, quota - * exceeded, or name conflict (a byte upload would fail identically). - * - * @param {File} file - * @param {string | null | undefined} folderId - * @returns {Promise<{ ok: boolean, data?: any, errorMsg?: string, isQuotaError?: boolean } | null>} - */ -export async function tryInstantUpload(file, folderId) { - if (!folderId || file.size < INSTANT_UPLOAD_MIN_SIZE || _instantUploadUsable === false || typeof Worker === 'undefined') { - return null; - } - - const hash = await hashFileInWorker(file); - if (!hash) return null; - - if (!(await callerOwnsHash(hash))) return null; - - try { - const response = await fetch('/api/files/by-hash', { - method: 'POST', - headers: { - ...getCsrfHeaders(), - 'Content-Type': 'application/json', - 'Cache-Control': 'no-cache, no-store, must-revalidate' - }, - body: JSON.stringify( - /** @type {import('../../core/types.js').CreateFileByHash} */ ({ - name: file.name, - folder_id: folderId, - hash - }) - ) - }); - - if (response.status === 201) { - return { ok: true, data: await response.json() }; - } - - /** @type {string} */ - let errorMsg = `Instant upload failed (HTTP ${response.status})`; - try { - const body = await response.json(); - errorMsg = body.message || body.error || errorMsg; - } catch (_) {} - - if (response.status === 507) { - return { ok: false, isQuotaError: true, errorMsg }; - } - if (response.status === 409) { - // Duplicate name in the folder — a byte upload would hit the - // exact same conflict; surface it without transferring. - return { ok: false, errorMsg }; - } - // 404 (ownership race with a delete+GC), 4xx/5xx: fall back to the - // byte upload — the server dedups it on write anyway. - return null; - } catch (_) { - return null; - } -} diff --git a/static/js/vendors/hash-wasm/oxicloud_hash_wasm.js b/static/js/vendors/hash-wasm/oxicloud_hash_wasm.js index 2a341978..b706d36b 100644 --- a/static/js/vendors/hash-wasm/oxicloud_hash_wasm.js +++ b/static/js/vendors/hash-wasm/oxicloud_hash_wasm.js @@ -70,6 +70,99 @@ export class Blake3Hasher { } if (Symbol.dispose) Blake3Hasher.prototype[Symbol.dispose] = Blake3Hasher.prototype.free; +/** + * Incremental FastCDC chunker + whole-file BLAKE3, for the delta-upload + * worker. Feed the file in slices; every call returns the chunks that + * became FINAL; `finish()` flushes the tail and returns the file hash. + * + * ```js + * const c = new DeltaChunker(); + * for (const slice of slices) { + * for (const [h, s] of JSON.parse(c.update(bytes))) { … } + * } + * const { chunks, file_hash } = JSON.parse(c.finish()); + * ``` + * + * Correctness of the incremental split: FastCDC decides each cut by + * scanning at most `CDC_MAX_CHUNK` bytes from the chunk's start. When + * the chunker runs over the buffered prefix of a longer file, every + * produced chunk except the LAST ended on a content/max-size condition + * — its decision window was fully available, so the full-file chunker + * makes the same cut. Only the last chunk (cut by "end of buffer") is + * provisional: it stays buffered and is re-examined when more bytes + * arrive. By induction the emitted boundaries equal a single FastCDC + * pass over the whole file — the mirror test below proves it. + */ +export class DeltaChunker { + __destroy_into_raw() { + const ptr = this.__wbg_ptr; + this.__wbg_ptr = 0; + DeltaChunkerFinalization.unregister(this); + return ptr; + } + free() { + const ptr = this.__destroy_into_raw(); + wasm.__wbg_deltachunker_free(ptr, 0); + } + /** + * Flush the provisional tail and return + * `{"chunks":[["",size]…],"file_hash":"","total":N}`. + * `chunks` holds at most one entry (the tail); an empty file has none + * and its `file_hash` is BLAKE3 of the empty input. + * @returns {string} + */ + finish() { + let deferred1_0; + let deferred1_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + wasm.deltachunker_finish(retptr, this.__wbg_ptr); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + deferred1_0 = r0; + deferred1_1 = r1; + return getStringFromWasm0(r0, r1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_export2(deferred1_0, deferred1_1, 1); + } + } + /** + * Create a chunker with the server's CDC parameters. + */ + constructor() { + const ret = wasm.deltachunker_new(); + this.__wbg_ptr = ret; + DeltaChunkerFinalization.register(this, this.__wbg_ptr, this); + return this; + } + /** + * Feed one slice. Returns a JSON array of the chunks that became + * final: `[["", size], …]` (possibly empty). + * @param {Uint8Array} data + * @returns {string} + */ + update(data) { + let deferred2_0; + let deferred2_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + const ptr0 = passArray8ToWasm0(data, wasm.__wbindgen_export); + const len0 = WASM_VECTOR_LEN; + wasm.deltachunker_update(retptr, this.__wbg_ptr, ptr0, len0); + var r0 = getDataViewMemory0().getInt32(retptr + 4 * 0, true); + var r1 = getDataViewMemory0().getInt32(retptr + 4 * 1, true); + deferred2_0 = r0; + deferred2_1 = r1; + return getStringFromWasm0(r0, r1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_export2(deferred2_0, deferred2_1, 1); + } + } +} +if (Symbol.dispose) DeltaChunker.prototype[Symbol.dispose] = DeltaChunker.prototype.free; + /** * One-shot convenience for small buffers. * @param {Uint8Array} data @@ -96,28 +189,26 @@ export function blake3Hex(data) { function __wbg_get_imports() { const import0 = { __proto__: null, - __wbg___wbindgen_throw_bbadd78c1bac3a77: (arg0, arg1) => { + __wbg___wbindgen_throw_bbadd78c1bac3a77: function(arg0, arg1) { throw new Error(getStringFromWasm0(arg0, arg1)); - } + }, }; return { __proto__: null, - './oxicloud_hash_wasm_bg.js': import0 + "./oxicloud_hash_wasm_bg.js": import0, }; } -const Blake3HasherFinalization = - typeof FinalizationRegistry === 'undefined' - ? { register: () => {}, unregister: () => {} } - : new FinalizationRegistry((ptr) => wasm.__wbg_blake3hasher_free(ptr, 1)); +const Blake3HasherFinalization = (typeof FinalizationRegistry === 'undefined') + ? { register: () => {}, unregister: () => {} } + : new FinalizationRegistry(ptr => wasm.__wbg_blake3hasher_free(ptr, 1)); +const DeltaChunkerFinalization = (typeof FinalizationRegistry === 'undefined') + ? { register: () => {}, unregister: () => {} } + : new FinalizationRegistry(ptr => wasm.__wbg_deltachunker_free(ptr, 1)); let cachedDataViewMemory0 = null; function getDataViewMemory0() { - if ( - cachedDataViewMemory0 === null || - cachedDataViewMemory0.buffer.detached === true || - (cachedDataViewMemory0.buffer.detached === undefined && cachedDataViewMemory0.buffer !== wasm.memory.buffer) - ) { + if (cachedDataViewMemory0 === null || cachedDataViewMemory0.buffer.detached === true || (cachedDataViewMemory0.buffer.detached === undefined && cachedDataViewMemory0.buffer !== wasm.memory.buffer)) { cachedDataViewMemory0 = new DataView(wasm.memory.buffer); } return cachedDataViewMemory0; @@ -177,13 +268,9 @@ async function __wbg_load(module, imports) { const validResponse = module.ok && expectedResponseType(module.type); if (validResponse && module.headers.get('Content-Type') !== 'application/wasm') { - console.warn( - '`WebAssembly.instantiateStreaming` failed because your server does not serve Wasm with `application/wasm` MIME type. Falling back to `WebAssembly.instantiate` which is slower. Original error:\n', - e - ); - } else { - throw e; - } + console.warn("`WebAssembly.instantiateStreaming` failed because your server does not serve Wasm with `application/wasm` MIME type. Falling back to `WebAssembly.instantiate` which is slower. Original error:\n", e); + + } else { throw e; } } } @@ -201,10 +288,7 @@ async function __wbg_load(module, imports) { function expectedResponseType(type) { switch (type) { - case 'basic': - case 'cors': - case 'default': - return true; + case 'basic': case 'cors': case 'default': return true; } return false; } @@ -213,11 +297,12 @@ async function __wbg_load(module, imports) { function initSync(module) { if (wasm !== undefined) return wasm; + if (module !== undefined) { if (Object.getPrototypeOf(module) === Object.prototype) { - ({ module } = module); + ({module} = module) } else { - console.warn('using deprecated parameters for `initSync()`; pass a single object instead'); + console.warn('using deprecated parameters for `initSync()`; pass a single object instead') } } @@ -232,11 +317,12 @@ function initSync(module) { async function __wbg_init(module_or_path) { if (wasm !== undefined) return wasm; + if (module_or_path !== undefined) { if (Object.getPrototypeOf(module_or_path) === Object.prototype) { - ({ module_or_path } = module_or_path); + ({module_or_path} = module_or_path) } else { - console.warn('using deprecated parameters for the initialization function; pass a single object instead'); + console.warn('using deprecated parameters for the initialization function; pass a single object instead') } } @@ -245,11 +331,7 @@ async function __wbg_init(module_or_path) { } const imports = __wbg_get_imports(); - if ( - typeof module_or_path === 'string' || - (typeof Request === 'function' && module_or_path instanceof Request) || - (typeof URL === 'function' && module_or_path instanceof URL) - ) { + if (typeof module_or_path === 'string' || (typeof Request === 'function' && module_or_path instanceof Request) || (typeof URL === 'function' && module_or_path instanceof URL)) { module_or_path = fetch(module_or_path); } @@ -258,4 +340,4 @@ async function __wbg_init(module_or_path) { return __wbg_finalize_init(instance, module); } -export { __wbg_init as default, initSync }; +export { initSync, __wbg_init as default }; diff --git a/static/js/vendors/hash-wasm/oxicloud_hash_wasm_bg.wasm b/static/js/vendors/hash-wasm/oxicloud_hash_wasm_bg.wasm index 7ab00d46..0f09d2a6 100644 Binary files a/static/js/vendors/hash-wasm/oxicloud_hash_wasm_bg.wasm and b/static/js/vendors/hash-wasm/oxicloud_hash_wasm_bg.wasm differ diff --git a/static/js/workers/deltaWorker.js b/static/js/workers/deltaWorker.js new file mode 100644 index 00000000..3a86f650 --- /dev/null +++ b/static/js/workers/deltaWorker.js @@ -0,0 +1,330 @@ +/** + * OxiCloud — delta-upload worker ("upload only what changed"). + * + * Runs the whole client side of the delta protocol off the main thread: + * + * read 8 MiB slices ─► FastCDC chunk + BLAKE3 (WASM, same crate and + * parameters as the server) ─► negotiate hash batches ─► upload only + * the missing chunks (framed, bounded concurrency) ─► commit. + * + * The stages OVERLAP: negotiation of batch N and uploads of its missing + * chunks run while batch N+1 is still being hashed, so wall-clock time + * approaches max(hash time, upload time) instead of their sum. RAM stays + * flat: chunk bytes are re-sliced from the File at upload time, never + * hoarded. + * + * Protocol with the spawner: + * in : { file: File, folderId: string, name: string, csrfToken: string } + * out : { type: 'progress', hashedBytes, reusedBytes, uploadedBytes, totalBytes } + * { type: 'done', status, body } — conclusive HTTP outcome + * { type: 'fallback', reason } — do a plain byte upload + */ + +// Absolute URLs on purpose: vendors/workers are served verbatim in both +// dev and the release IIFE bundle (same pattern as the pdf.js loader). +const WASM_GLUE_URL = '/js/vendors/hash-wasm/oxicloud_hash_wasm.js'; + +/** File read granularity — large enough to amortize Blob→ArrayBuffer. */ +const SLICE_BYTES = 8 * 1024 * 1024; +/** Negotiate after this many freshly hashed chunks (~64 MiB of content). */ +const NEGOTIATE_BATCH = 256; +/** Group missing chunks into PUT bodies of at most this many bytes. */ +const UPLOAD_BATCH_BYTES = 8 * 1024 * 1024; +/** Concurrent chunk-PUT requests. */ +const UPLOAD_CONCURRENCY = 2; +/** Re-commit attempts when the server answers 409 still_missing. */ +const COMMIT_RETRIES = 2; + +/** + * Typed view of the dedicated-worker global scope (jsconfig targets the + * DOM lib, where `self` is a Window — cast to what this worker uses). + * @type {{ onmessage: ((event: MessageEvent) => void) | null, + * postMessage: (message: unknown) => void }} + */ +const workerScope = /** @type {any} */ (self); + +/** + * One chunk occurrence, in file order. + * @typedef {{ h: string, s: number, offset: number }} WorkerChunk + */ + +/** @returns {Promise} the initialized WASM module */ +async function loadWasm() { + const mod = await import(WASM_GLUE_URL); + await mod.default(); + return mod; +} + +workerScope.onmessage = async (event) => { + const { file, folderId, name, csrfToken } = /** @type {{ file: File, folderId: string, name: string, csrfToken: string }} */ (event.data); + + /** @param {string} reason */ + const fallback = (reason) => workerScope.postMessage({ type: 'fallback', reason }); + + /** @type {Record} */ + const mutHeaders = { 'Content-Type': 'application/json' }; + if (csrfToken) mutHeaders['X-CSRF-Token'] = csrfToken; + + let wasm; + try { + wasm = await loadWasm(); + } catch (err) { + fallback(`wasm unavailable: ${err instanceof Error ? err.message : String(err)}`); + return; + } + + // ── Shared pipeline state ───────────────────────────────────── + /** @type {WorkerChunk[]} */ + const chunks = []; // every occurrence, in file order + /** @type {Set} */ + const seenForNegotiate = new Set(); // distinct hashes already sent to negotiate + let reusedBytes = 0; + let uploadedBytes = 0; + let hashedBytes = 0; + let failed = /** @type {string | null} */ (null); + + let lastProgress = 0; + const progress = (force = false) => { + const now = Date.now(); + if (!force && now - lastProgress < 150) return; + lastProgress = now; + workerScope.postMessage({ + type: 'progress', + hashedBytes, + reusedBytes, + uploadedBytes, + totalBytes: file.size + }); + }; + + // ── Upload stage: bounded-concurrency drain of uploadByHash ── + /** @type {WorkerChunk[]} */ + const uploadQueue = []; + /** @type {Promise[]} */ + const uploadWorkers = []; + let uploadsClosed = false; + /** @type {(() => void) | null} */ + let wakeUploader = null; + const signalUploaders = () => { + if (wakeUploader) { + const w = wakeUploader; + wakeUploader = null; + w(); + } + }; + + /** Encode a batch of chunks as [u32 BE len][bytes] frames. */ + const encodeFrames = async (/** @type {WorkerChunk[]} */ batch) => { + const total = batch.reduce((n, c) => n + 4 + c.s, 0); + const wire = new Uint8Array(total); + const view = new DataView(wire.buffer); + let at = 0; + for (const c of batch) { + // eslint-disable-next-line no-await-in-loop -- sequential by design: constant RAM + const bytes = new Uint8Array(await file.slice(c.offset, c.offset + c.s).arrayBuffer()); + view.setUint32(at, c.s, false); + wire.set(bytes, at + 4); + at += 4 + c.s; + } + return wire; + }; + + const uploadLoop = async () => { + while (!failed) { + // Take up to UPLOAD_BATCH_BYTES from the queue. + /** @type {WorkerChunk[]} */ + const batch = []; + let bytes = 0; + while (uploadQueue.length > 0 && bytes < UPLOAD_BATCH_BYTES) { + const c = /** @type {WorkerChunk} */ (uploadQueue.shift()); + batch.push(c); + bytes += c.s; + } + if (batch.length === 0) { + if (uploadsClosed) return; + // eslint-disable-next-line no-await-in-loop -- queue wait + await new Promise((resolve) => { + wakeUploader = /** @type {() => void} */ (resolve); + }); + continue; + } + try { + // eslint-disable-next-line no-await-in-loop -- bounded by pool size + const wire = await encodeFrames(batch); + // eslint-disable-next-line no-await-in-loop -- bounded by pool size + const response = await fetch('/api/files/delta/chunks', { + method: 'PUT', + headers: { + 'Content-Type': 'application/octet-stream', + ...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {}) + }, + body: wire + }); + if (!response.ok) { + failed = `chunk PUT failed (HTTP ${response.status})`; + return; + } + for (const c of batch) uploadedBytes += c.s; + progress(); + } catch (err) { + failed = `chunk PUT failed: ${err instanceof Error ? err.message : String(err)}`; + return; + } + } + }; + for (let i = 0; i < UPLOAD_CONCURRENCY; i++) uploadWorkers.push(uploadLoop()); + + // ── Negotiate stage ─────────────────────────────────────────── + /** @type {Promise[]} */ + const negotiations = []; + const negotiate = (/** @type {WorkerChunk[]} */ fresh) => { + if (fresh.length === 0 || failed) return; + negotiations.push( + (async () => { + try { + const response = await fetch('/api/files/delta/negotiate', { + method: 'POST', + headers: mutHeaders, + body: JSON.stringify({ chunks: fresh.map(({ h, s }) => ({ h, s })) }) + }); + if (!response.ok) { + failed = failed || `negotiate failed (HTTP ${response.status})`; + return; + } + const missing = new Set(/** @type {{missing: string[]}} */ (await response.json()).missing); + for (const c of fresh) { + if (missing.has(c.h)) { + uploadQueue.push(c); + } else { + reusedBytes += c.s; + } + } + signalUploaders(); + progress(); + } catch (err) { + failed = failed || `negotiate failed: ${err instanceof Error ? err.message : String(err)}`; + } + })() + ); + }; + + // ── Chunking stage (drives the other two) ──────────────────── + try { + const chunker = new wasm.DeltaChunker(); + /** @type {WorkerChunk[]} */ + let freshBatch = []; + let offset = 0; + + /** @param {[string, number][]} emitted */ + const onChunks = (emitted) => { + for (const [h, s] of emitted) { + /** @type {WorkerChunk} */ + const chunk = { h, s, offset }; + offset += s; + chunks.push(chunk); + if (seenForNegotiate.has(h)) { + // Repeated content inside the same file: the first + // occurrence decides upload vs reuse; later ones are + // pure reuse for accounting. + reusedBytes += s; + } else { + seenForNegotiate.add(h); + freshBatch.push(chunk); + if (freshBatch.length >= NEGOTIATE_BATCH) { + negotiate(freshBatch); + freshBatch = []; + } + } + } + }; + + for (let read = 0; read < file.size && !failed; read += SLICE_BYTES) { + const end = Math.min(read + SLICE_BYTES, file.size); + // eslint-disable-next-line no-await-in-loop -- sequential by design: constant RAM + const slice = new Uint8Array(await file.slice(read, end).arrayBuffer()); + onChunks(JSON.parse(chunker.update(slice))); + hashedBytes = end; + progress(); + } + const fin = JSON.parse(chunker.finish()); + chunker.free(); + onChunks(fin.chunks); + negotiate(freshBatch); + const fileHash = /** @type {string} */ (fin.file_hash); + hashedBytes = file.size; + progress(true); + + // ── Drain: negotiations → uploads → commit ─────────────── + await Promise.all(negotiations); + uploadsClosed = true; + signalUploaders(); + await Promise.all(uploadWorkers); + if (failed) { + fallback(failed); + return; + } + + const commitBody = { + file_hash: fileHash, + chunks: chunks.map(({ h, s }) => ({ h, s })), + name, + folder_id: folderId + }; + for (let attempt = 0; ; attempt++) { + // eslint-disable-next-line no-await-in-loop -- retry loop + const response = await fetch('/api/files/delta/commit', { + method: 'POST', + headers: mutHeaders, + body: JSON.stringify(commitBody) + }); + /** @type {any} */ + let body = null; + try { + // eslint-disable-next-line no-await-in-loop -- retry loop + body = await response.json(); + } catch (_) {} + + const stillMissing = response.status === 409 && Array.isArray(body?.still_missing); + if (stillMissing && attempt < COMMIT_RETRIES) { + // GC race or a chunk we wrongly assumed claimable: upload + // exactly what the server names and try again. + const byHash = new Map(chunks.map((c) => [c.h, c])); + /** @type {WorkerChunk[]} */ + const retry = []; + for (const h of body.still_missing) { + const c = byHash.get(h); + if (!c) { + fallback('server requested an unknown chunk'); + return; + } + retry.push(c); + } + const wire = await encodeFrames(retry); + // eslint-disable-next-line no-await-in-loop -- retry loop + const put = await fetch('/api/files/delta/chunks', { + method: 'PUT', + headers: { + 'Content-Type': 'application/octet-stream', + ...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {}) + }, + body: wire + }); + if (!put.ok) { + fallback(`retry chunk PUT failed (HTTP ${put.status})`); + return; + } + for (const c of retry) uploadedBytes += c.s; + progress(true); + continue; + } + + // Conclusive: 201 created, or a real error (quota, name + // conflict, validation). The spawner maps it to the uploaders' + // UploadAnswer contract. + workerScope.postMessage({ type: 'done', status: response.status, body }); + return; + } + } catch (err) { + fallback(err instanceof Error ? err.message : String(err)); + } +}; diff --git a/static/js/workers/hashWorker.js b/static/js/workers/hashWorker.js deleted file mode 100644 index d2061ecd..00000000 --- a/static/js/workers/hashWorker.js +++ /dev/null @@ -1,79 +0,0 @@ -/** - * OxiCloud — BLAKE3 hashing worker (instant-upload support). - * - * Hashes a File off the main thread, reading it in fixed-size slices so - * RAM stays constant regardless of file size. The WASM module is compiled - * from the exact same `blake3` crate the server uses, so the digest - * computed here equals the server's content address bit for bit. - * - * Protocol: receives `{ file: File }`, answers - * `{ ok: true, hash: string }` or `{ ok: false, error: string }`. - * The spawner terminates the worker after one file. - */ - -// Absolute URL on purpose: vendors are served verbatim at /js/vendors/ in -// both dev and release mode (the release IIFE bundle would break a -// relative import) — same pattern as the pdf.js loader in thumbnail.js. -const WASM_GLUE_URL = '/js/vendors/hash-wasm/oxicloud_hash_wasm.js'; - -/** - * 8 MiB slices — large enough to amortize the per-slice Blob→ArrayBuffer - * round-trip, small enough that peak worker RAM stays flat for any size. - */ -const SLICE_BYTES = 8 * 1024 * 1024; - -/** - * Typed view of the dedicated-worker global scope. The project's - * jsconfig targets the DOM lib, where `self` is a Window — cast to the - * two members this worker actually uses. - * @type {{ onmessage: ((event: MessageEvent) => void) | null, - * postMessage: (message: unknown) => void }} - */ -const workerScope = /** @type {any} */ (self); - -/** - * Memoized WASM module (in-flight or settled), `default()` already run. - * Reset on failure so a later message can retry a transient load error. - * @type {Promise | null} - */ -let _wasmPromise = null; - -/** @returns {Promise} */ -function getWasm() { - if (!_wasmPromise) { - _wasmPromise = import(WASM_GLUE_URL) - .then(async (mod) => { - await mod.default(); - return mod; - }) - .catch((err) => { - _wasmPromise = null; - throw err; - }); - } - return _wasmPromise; -} - -workerScope.onmessage = async (event) => { - const file = /** @type {{ file: File }} */ (event.data).file; - try { - const wasm = await getWasm(); - const hasher = new wasm.Blake3Hasher(); - try { - for (let offset = 0; offset < file.size; offset += SLICE_BYTES) { - const end = Math.min(offset + SLICE_BYTES, file.size); - // eslint-disable-next-line no-await-in-loop -- sequential by design: constant RAM - const buffer = await file.slice(offset, end).arrayBuffer(); - hasher.update(new Uint8Array(buffer)); - } - workerScope.postMessage({ ok: true, hash: hasher.finalizeHex() }); - } finally { - hasher.free(); - } - } catch (err) { - workerScope.postMessage({ - ok: false, - error: err instanceof Error ? err.message : String(err) - }); - } -}; diff --git a/wasm/oxicloud-hash/Cargo.lock b/wasm/oxicloud-hash/Cargo.lock index b053178b..5675d509 100644 --- a/wasm/oxicloud-hash/Cargo.lock +++ b/wasm/oxicloud-hash/Cargo.lock @@ -65,6 +65,12 @@ dependencies = [ "libc", ] +[[package]] +name = "fastcdc" +version = "4.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77af40d8a8dadb92dc178569a5f5edb5f3056e98255c2de48ab5d59a52892e0c" + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -88,6 +94,7 @@ name = "oxicloud-hash-wasm" version = "0.1.0" dependencies = [ "blake3", + "fastcdc", "wasm-bindgen", ] diff --git a/wasm/oxicloud-hash/Cargo.toml b/wasm/oxicloud-hash/Cargo.toml index 0c54d4d1..b95fd253 100644 --- a/wasm/oxicloud-hash/Cargo.toml +++ b/wasm/oxicloud-hash/Cargo.toml @@ -20,6 +20,10 @@ crate-type = ["cdylib"] # evergreen browser since 2023 supports it, and the frontend falls back # to a plain byte upload when instantiation fails. blake3 = { version = "1.8.4", default-features = false, features = ["wasm32_simd"] } +# Same crate AND parameters as the server's CDC dedup engine — chunk +# boundaries computed in the browser must equal the server's bit for bit, +# or cross-version dedup between byte uploads and delta uploads collapses. +fastcdc = "4.0.0" wasm-bindgen = "0.2" [profile.release] diff --git a/wasm/oxicloud-hash/src/lib.rs b/wasm/oxicloud-hash/src/lib.rs index 91cb8d55..627c379b 100644 --- a/wasm/oxicloud-hash/src/lib.rs +++ b/wasm/oxicloud-hash/src/lib.rs @@ -64,6 +64,137 @@ pub fn blake3_hex(data: &[u8]) -> String { blake3::hash(data).to_hex().to_string() } +// ── Delta-upload chunker ───────────────────────────────────────────────────── + +/// CDC parameters — MUST mirror `dedup_service.rs` on the server +/// (`CDC_MIN_CHUNK` / `CDC_AVG_CHUNK` / `CDC_MAX_CHUNK`). Identical +/// parameters + identical crate ⇒ identical boundaries, which is what +/// makes a chunk hashed in the browser deduplicate against a chunk the +/// server cut from a byte upload. +const CDC_MIN_CHUNK: usize = 65_536; +const CDC_AVG_CHUNK: usize = 262_144; +const CDC_MAX_CHUNK: usize = 1_048_576; + +/// Incremental FastCDC chunker + whole-file BLAKE3, for the delta-upload +/// worker. Feed the file in slices; every call returns the chunks that +/// became FINAL; `finish()` flushes the tail and returns the file hash. +/// +/// ```js +/// const c = new DeltaChunker(); +/// for (const slice of slices) { +/// for (const [h, s] of JSON.parse(c.update(bytes))) { … } +/// } +/// const { chunks, file_hash } = JSON.parse(c.finish()); +/// ``` +/// +/// Correctness of the incremental split: FastCDC decides each cut by +/// scanning at most `CDC_MAX_CHUNK` bytes from the chunk's start. When +/// the chunker runs over the buffered prefix of a longer file, every +/// produced chunk except the LAST ended on a content/max-size condition +/// — its decision window was fully available, so the full-file chunker +/// makes the same cut. Only the last chunk (cut by "end of buffer") is +/// provisional: it stays buffered and is re-examined when more bytes +/// arrive. By induction the emitted boundaries equal a single FastCDC +/// pass over the whole file — the mirror test below proves it. +#[wasm_bindgen] +pub struct DeltaChunker { + /// Provisional tail: bytes after the last FINAL cut. + buf: Vec, + file_hasher: blake3::Hasher, + total: u64, +} + +/// Append one `["",len]` item to a hand-rolled JSON array — hashes +/// are hex and sizes are integers, so manual JSON is unambiguous and +/// keeps a serde dependency out of the wasm binary. +fn push_chunk_json(out: &mut String, hash: &str, len: usize) { + if !out.ends_with('[') { + out.push(','); + } + out.push_str("[\""); + out.push_str(hash); + out.push_str("\","); + out.push_str(&len.to_string()); + out.push(']'); +} + +#[wasm_bindgen] +impl DeltaChunker { + /// Create a chunker with the server's CDC parameters. + #[wasm_bindgen(constructor)] + pub fn new() -> DeltaChunker { + DeltaChunker { + buf: Vec::with_capacity(2 * CDC_MAX_CHUNK), + file_hasher: blake3::Hasher::new(), + total: 0, + } + } + + /// Feed one slice. Returns a JSON array of the chunks that became + /// final: `[["", size], …]` (possibly empty). + pub fn update(&mut self, data: &[u8]) -> String { + self.file_hasher.update(data); + self.total += data.len() as u64; + self.buf.extend_from_slice(data); + + let mut out = String::from("["); + let mut consumed = 0usize; + { + let chunks: Vec = fastcdc::v2020::FastCDC::new( + &self.buf, + CDC_MIN_CHUNK, + CDC_AVG_CHUNK, + CDC_MAX_CHUNK, + ) + .collect(); + // Every chunk but the last ended on a content/max condition → + // final. The last one ended because the buffer did → keep it. + for chunk in chunks.iter().take(chunks.len().saturating_sub(1)) { + let bytes = &self.buf[chunk.offset..chunk.offset + chunk.length]; + push_chunk_json( + &mut out, + &blake3::hash(bytes).to_hex().to_string(), + chunk.length, + ); + consumed = chunk.offset + chunk.length; + } + } + if consumed > 0 { + self.buf.drain(..consumed); + } + out.push(']'); + out + } + + /// Flush the provisional tail and return + /// `{"chunks":[["",size]…],"file_hash":"","total":N}`. + /// `chunks` holds at most one entry (the tail); an empty file has none + /// and its `file_hash` is BLAKE3 of the empty input. + pub fn finish(&mut self) -> String { + let mut out = String::from("{\"chunks\":["); + if !self.buf.is_empty() { + push_chunk_json( + &mut out, + &blake3::hash(&self.buf).to_hex().to_string(), + self.buf.len(), + ); + self.buf.clear(); + } + out.push_str("],\"file_hash\":\""); + out.push_str(&self.file_hasher.finalize().to_hex().to_string()); + out.push_str("\",\"total\":"); + out.push_str(&self.total.to_string()); + out.push('}'); + out + } +} + +impl Default for DeltaChunker { + fn default() -> Self { + Self::new() + } +} + #[cfg(test)] mod tests { use super::*; @@ -95,4 +226,94 @@ mod tests { "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262" ); } + + // ── DeltaChunker mirror test ───────────────────────────────── + // + // The client-side twin of the server's + // `test_stream_chunking_matches_slice_chunking`: incremental chunking + // with adversarial slice sizes must produce exactly the boundaries of + // one FastCDC pass over the whole buffer — the property cross-version + // dedup between byte uploads and delta uploads hangs on. + + fn run_chunker(data: &[u8], slice: usize) -> (Vec<(String, usize)>, String) { + let mut chunker = DeltaChunker::new(); + let mut chunks: Vec<(String, usize)> = Vec::new(); + let mut parse = |json: &str, into: &mut Vec<(String, usize)>| { + // items look like ["",N] — split on '[' groups. + for item in json.split("[\"").skip(1) { + let hash = &item[..64]; + let size: usize = item[66..item.find(']').unwrap()].parse().unwrap(); + into.push((hash.to_string(), size)); + } + }; + for piece in data.chunks(slice.max(1)) { + let emitted = chunker.update(piece); + parse(&emitted, &mut chunks); + } + let fin = chunker.finish(); + let tail_json = &fin[fin.find('[').unwrap()..=fin.find(']').unwrap()]; + parse(tail_json, &mut chunks); + let file_hash = fin.split("\"file_hash\":\"").nth(1).unwrap()[..64].to_string(); + (chunks, file_hash) + } + + #[test] + fn incremental_chunking_matches_single_pass() { + // 4 MiB of xorshift noise — genuinely content-defined cut points + // (a byte-periodic generator would only ever hit max-size cuts). + let mut state: u64 = 0x243F_6A88_85A3_08D3; + let mut data = Vec::with_capacity(4 * 1024 * 1024); + while data.len() < 4 * 1024 * 1024 { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + data.extend_from_slice(&state.to_le_bytes()); + } + + let reference: Vec<(String, usize)> = + fastcdc::v2020::FastCDC::new(&data, CDC_MIN_CHUNK, CDC_AVG_CHUNK, CDC_MAX_CHUNK) + .map(|c| { + ( + blake3::hash(&data[c.offset..c.offset + c.length]) + .to_hex() + .to_string(), + c.length, + ) + }) + .collect(); + assert!(reference.len() > 4, "test data must span several chunks"); + + // Slice sizes chosen to stress every refill path: tiny (7 B), + // typical worker slice (8 MiB > file), page-ish, and exactly the + // CDC max so provisional tails land on boundaries. + for slice in [7usize, 4096, CDC_MAX_CHUNK, 8 * 1024 * 1024] { + let (chunks, file_hash) = run_chunker(&data, slice); + assert_eq!( + chunks, reference, + "boundaries must not depend on slicing (slice={slice})" + ); + assert_eq!( + file_hash, + blake3_hex(&data), + "file hash must match one-shot BLAKE3 (slice={slice})" + ); + } + } + + #[test] + fn delta_chunker_empty_and_tiny_inputs() { + let (chunks, file_hash) = run_chunker(b"", 1024); + assert!(chunks.is_empty()); + assert_eq!( + file_hash, + "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262" + ); + + let tiny = b"below the CDC minimum"; + let (chunks, file_hash) = run_chunker(tiny, 4); + assert_eq!(chunks.len(), 1, "tiny input is one (tail) chunk"); + assert_eq!(chunks[0].1, tiny.len()); + assert_eq!(chunks[0].0, blake3_hex(tiny)); + assert_eq!(file_hash, blake3_hex(tiny)); + } }