Edouard Vanbelle
e357caf43b
in this case logs are added in the delta upload worker
you can increase verbosity from console via:
```javascript
// Usage:
oxi.setLogLevel('oxi:upload', 'debug') // deep dive
oxi.setLogLevel('oxi:upload', 'warn') // quiet
oxi.log.setLevel('debug') // everything to debug
```
values are stored in localstorage
example:
```
oxi.setLogLevel('oxi:upload', 'debug');
'oxi:upload → debug'
deltaUpload.ts:120 [f32ce1] delta start {file: 'Revue de presseg.odp', size: 23658927}
deltaUpload.ts:185 [f32ce1] worker: worker start {file: 'Revue de presseg.odp', size: 23658927}
deltaUpload.ts:186 [f32ce1] worker: wasm loaded
deltaUpload.ts:186 [f32ce1] worker: hashed — blake3=150edbc35f3475a0bdeecd15fe5278dda305ce3b27e53e9c40515a69f0b883f3 (76 chunks)
deltaUpload.ts:186 [f32ce1] worker: negotiate: 76 hashes → 76 missing, 0 dedup'd
deltaUpload.ts:186 [f32ce1] worker: chunk PUT: 28 chunks, 8825338 bytes
deltaUpload.ts:186 [f32ce1] worker: chunk PUT: 29 chunks, 8950230 bytes
deltaUpload.ts:186 [f32ce1] worker: chunk PUT: 19 chunks, 5883663 bytes
deltaUpload.ts:186 [f32ce1] worker: ✅ committed — uploaded 23 658 927 B (no dedup, blake3=150edbc35f3475a0bdeecd15fe5278dda305ce3b27e53e9c40515a69f0b883f3)
deltaUpload.ts:185 [f32ce1] worker: commit HTTP 201 {blake3: '150edbc35f3475a0bdeecd15fe5278dda305ce3b27e53e9c40515a69f0b883f3', uploadedBytes: 23658927, reusedBytes: 0, totalBytes: 23658927, attempt: 0}
deltaUpload.ts:213 [f32ce1] delta done {file: 'Revue de presseg.odp', blake3: '150edbc35f3475a0bdeecd15fe5278dda305ce3b27e53e9c40515a69f0b883f3', savedBytes: 0, uploadedBytes: 23658927}
```
423 lines
18 KiB
JavaScript
423 lines
18 KiB
JavaScript
/**
|
|
* 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 static build (served verbatim from /static).
|
|
const WASM_GLUE_URL = '/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;
|
|
/** Reclaim consumed queue slots periodically. A head cursor makes dequeue O(1);
|
|
* compaction bounds the backing array when hashing stays ahead of the network. */
|
|
const UPLOAD_QUEUE_COMPACT_AT = 4096;
|
|
/** Concurrent chunk-PUT requests. Kept at 1: several folder files upload through
|
|
* their own workers at once, and the browser only grants ~6 connections per
|
|
* host. Combined with serialized negotiate (below) each worker holds at most
|
|
* ~2 connections (one negotiate + one chunk PUT), so a couple of concurrent
|
|
* large files can't starve the plain uploads of the small ones. */
|
|
const UPLOAD_CONCURRENCY = 1;
|
|
/** 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<any>} 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);
|
|
|
|
/**
|
|
* Forward a log line to the main-thread orchestrator, which routes it
|
|
* through the shared `loglevel` logger (namespace `oxi:upload`). Worker
|
|
* can't `import 'loglevel'` — it's served from /static and isn't
|
|
* bundler-resolved — so postMessage is the transport.
|
|
*
|
|
* @param {'debug'|'info'|'warn'|'error'} level
|
|
* @param {string} msg
|
|
* @param {Record<string, unknown>=} extra
|
|
*/
|
|
const log = (level, msg, extra) =>
|
|
workerScope.postMessage({ type: 'log', level, msg, extra });
|
|
|
|
/** @param {string} reason */
|
|
const fallback = (reason) => {
|
|
log('warn', `worker fallback: ${reason}`);
|
|
workerScope.postMessage({ type: 'fallback', reason });
|
|
};
|
|
|
|
log('info', `worker start`, { file: name, size: file.size });
|
|
|
|
/** @type {Record<string, string>} */
|
|
const mutHeaders = { 'Content-Type': 'application/json' };
|
|
if (csrfToken) mutHeaders['X-CSRF-Token'] = csrfToken;
|
|
|
|
let wasm;
|
|
try {
|
|
wasm = await loadWasm();
|
|
log('debug', 'wasm loaded');
|
|
} 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<string>} */
|
|
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 | undefined)[]} */
|
|
const uploadQueue = [];
|
|
let uploadHead = 0;
|
|
/** @type {Promise<void>[]} */
|
|
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 (uploadHead < uploadQueue.length && bytes < UPLOAD_BATCH_BYTES) {
|
|
const c = /** @type {WorkerChunk} */ (uploadQueue[uploadHead]);
|
|
uploadQueue[uploadHead] = undefined;
|
|
uploadHead++;
|
|
batch.push(c);
|
|
bytes += c.s;
|
|
}
|
|
if (uploadHead === uploadQueue.length) {
|
|
uploadQueue.length = 0;
|
|
uploadHead = 0;
|
|
} else if (
|
|
uploadHead >= UPLOAD_QUEUE_COMPACT_AT &&
|
|
uploadHead * 2 >= uploadQueue.length
|
|
) {
|
|
uploadQueue.copyWithin(0, uploadHead);
|
|
uploadQueue.length -= uploadHead;
|
|
uploadHead = 0;
|
|
}
|
|
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);
|
|
log('debug', `chunk PUT: ${batch.length} chunks, ${wire.length} bytes`);
|
|
// 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})`;
|
|
log('error', failed);
|
|
return;
|
|
}
|
|
for (const c of batch) uploadedBytes += c.s;
|
|
progress();
|
|
} catch (err) {
|
|
failed = `chunk PUT failed: ${err instanceof Error ? err.message : String(err)}`;
|
|
log('error', failed);
|
|
return;
|
|
}
|
|
}
|
|
};
|
|
for (let i = 0; i < UPLOAD_CONCURRENCY; i++) uploadWorkers.push(uploadLoop());
|
|
|
|
// ── Negotiate stage ───────────────────────────────────────────
|
|
// Serialized: each negotiate awaits the previous one, so at most a single
|
|
// negotiate request is ever in flight per worker. Together with the single
|
|
// chunk-PUT lane (UPLOAD_CONCURRENCY = 1) this caps the worker at ~2
|
|
// concurrent connections, leaving room under the browser's ~6-per-host
|
|
// budget for the other folder files uploading in parallel.
|
|
/** @type {Promise<void>[]} */
|
|
const negotiations = [];
|
|
let negotiateTail = Promise.resolve();
|
|
const negotiate = (/** @type {WorkerChunk[]} */ fresh) => {
|
|
if (fresh.length === 0 || failed) return;
|
|
const run = negotiateTail.then(async () => {
|
|
if (failed) return;
|
|
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})`;
|
|
log('error', `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;
|
|
}
|
|
}
|
|
log(
|
|
'info',
|
|
`negotiate: ${fresh.length} hashes → ${missing.size} missing, ${fresh.length - missing.size} dedup'd`
|
|
);
|
|
signalUploaders();
|
|
progress();
|
|
} catch (err) {
|
|
failed = failed || `negotiate failed: ${err instanceof Error ? err.message : String(err)}`;
|
|
log('error', `negotiate failed: ${err instanceof Error ? err.message : String(err)}`);
|
|
}
|
|
});
|
|
negotiateTail = run.catch(() => {});
|
|
negotiations.push(run);
|
|
};
|
|
|
|
// ── 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);
|
|
// Log the whole-file BLAKE3 immediately so it's visible in the
|
|
// trace regardless of whether the commit succeeds. Correlates
|
|
// the client-side view with the server's `file_blobs.hash`.
|
|
log('info', `hashed — blake3=${fileHash} (${chunks.length} chunks)`);
|
|
|
|
// ── 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.
|
|
const ok = response.status >= 200 && response.status < 300;
|
|
if (ok) {
|
|
// Human-friendly outcome line — the raw commit line below
|
|
// still carries the byte counts for anyone who wants them.
|
|
if (uploadedBytes === 0 && reusedBytes > 0) {
|
|
log('info', `✅ file already on server — 100% dedup, no bytes transferred (${reusedBytes.toLocaleString()} B reused, blake3=${fileHash})`);
|
|
} else if (reusedBytes > 0) {
|
|
const pct = Math.round((100 * reusedBytes) / file.size);
|
|
log('info', `✅ committed — uploaded ${uploadedBytes.toLocaleString()} B, reused ${reusedBytes.toLocaleString()} B (${pct}% dedup, blake3=${fileHash})`);
|
|
} else {
|
|
log('info', `✅ committed — uploaded ${uploadedBytes.toLocaleString()} B (no dedup, blake3=${fileHash})`);
|
|
}
|
|
}
|
|
log(ok ? 'info' : 'warn', `commit HTTP ${response.status}`, {
|
|
blake3: fileHash,
|
|
uploadedBytes,
|
|
reusedBytes,
|
|
totalBytes: file.size,
|
|
attempt,
|
|
});
|
|
// Include the final counters + file hash on the done envelope
|
|
// so the orchestrator's summary is accurate even when the last
|
|
// throttled progress() got skipped (fast dedup-heavy paths
|
|
// complete under 150ms — progress' throttle window — so
|
|
// reusedBytes never surfaced via a progress message).
|
|
workerScope.postMessage({
|
|
type: 'done',
|
|
status: response.status,
|
|
body,
|
|
reusedBytes,
|
|
uploadedBytes,
|
|
fileHash,
|
|
});
|
|
return;
|
|
}
|
|
} catch (err) {
|
|
fallback(err instanceof Error ? err.message : String(err));
|
|
}
|
|
};
|