fix(upload): bound delta-worker connections instead of disabling delta
Follow-up to the connection-exhaustion fix. Rather than routing large files to plain uploads (which kept STORAGE dedup but gave up delta's re-upload bandwidth savings), keep delta for every file >= 8 MB and instead cap each worker's concurrent connections so a few large files uploading at once can't blow past the browser's ~6-per-host budget and starve the small-file plain uploads. - deltaWorker.js: serialize negotiate (at most one in flight per worker) and drop chunk-PUT concurrency 2 -> 1, so each worker holds ~2 connections max. - deltaUpload.ts: revert the 64 MB threshold back to 8 MB — every large file gets sub-file dedup again. (Storage dedup was never affected: BLAKE3 + CDC + ref-counting run server-side for plain and delta uploads alike.) With main upload concurrency at 2, total in-flight upload connections stay <= ~4. npm run check: clean, 58 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -30,8 +30,12 @@ const SLICE_BYTES = 8 * 1024 * 1024;
|
||||
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;
|
||||
/** 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;
|
||||
|
||||
@@ -175,37 +179,44 @@ workerScope.onmessage = async (event) => {
|
||||
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;
|
||||
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)}`;
|
||||
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})`;
|
||||
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)}`;
|
||||
}
|
||||
});
|
||||
negotiateTail = run.catch(() => {});
|
||||
negotiations.push(run);
|
||||
};
|
||||
|
||||
// ── Chunking stage (drives the other two) ────────────────────
|
||||
|
||||
Reference in New Issue
Block a user