feat(ui:delta): add heartbeat on worker
This commit is contained in:
@@ -84,7 +84,14 @@ interface LogMsg {
|
|||||||
msg: string;
|
msg: string;
|
||||||
extra?: Record<string, unknown>;
|
extra?: Record<string, unknown>;
|
||||||
}
|
}
|
||||||
type WorkerMsg = ProgressMsg | FallbackMsg | DoneMsg | LogMsg;
|
/** Kept-alive signal the worker emits every 5 s while awaiting a long
|
||||||
|
* fetch (commit, slow chunk PUT). The orchestrator's on-every-message
|
||||||
|
* `armStall()` resets the watchdog just from receiving this — the type
|
||||||
|
* handler below intentionally no-ops so nothing else fires. */
|
||||||
|
interface HeartbeatMsg {
|
||||||
|
type: 'heartbeat';
|
||||||
|
}
|
||||||
|
type WorkerMsg = ProgressMsg | FallbackMsg | DoneMsg | LogMsg | HeartbeatMsg;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Try to upload `file` through the delta protocol. Resolves `null` whenever
|
* Try to upload `file` through the delta protocol. Resolves `null` whenever
|
||||||
@@ -137,9 +144,22 @@ export function tryDeltaUpload(
|
|||||||
let savedBytes = 0;
|
let savedBytes = 0;
|
||||||
|
|
||||||
let stallTimer: ReturnType<typeof setTimeout>;
|
let stallTimer: ReturnType<typeof setTimeout>;
|
||||||
|
// Page Visibility listener — pause the stall watchdog while the
|
||||||
|
// tab is hidden. Background tabs get main-thread timer throttling
|
||||||
|
// (Chrome/Firefox: 1 s min tick, ~5 min hidden → timers may pause
|
||||||
|
// entirely) but Web Workers keep running at full speed. Without
|
||||||
|
// this pause, a tab-switch during a large upload would let messages
|
||||||
|
// queue up on the throttled main thread while the watchdog fires
|
||||||
|
// spuriously — poisoning `usable = false` for the rest of the
|
||||||
|
// session even though the worker was healthy the whole time.
|
||||||
|
let visibilityListener: (() => void) | null = null;
|
||||||
const settle = (answer: DeltaUploadAnswer | null) => {
|
const settle = (answer: DeltaUploadAnswer | null) => {
|
||||||
clearTimeout(timer);
|
clearTimeout(timer);
|
||||||
clearTimeout(stallTimer);
|
clearTimeout(stallTimer);
|
||||||
|
if (visibilityListener) {
|
||||||
|
document.removeEventListener('visibilitychange', visibilityListener);
|
||||||
|
visibilityListener = null;
|
||||||
|
}
|
||||||
worker.terminate();
|
worker.terminate();
|
||||||
resolve(answer);
|
resolve(answer);
|
||||||
};
|
};
|
||||||
@@ -157,9 +177,18 @@ export function tryDeltaUpload(
|
|||||||
// — exactly what freezes a folder upload ~2 min per large file. Disable
|
// — exactly what freezes a folder upload ~2 min per large file. Disable
|
||||||
// delta for this file AND every later one so they fall straight to a plain
|
// delta for this file AND every later one so they fall straight to a plain
|
||||||
// upload instead of each burning the full size-scaled delta timeout.
|
// upload instead of each burning the full size-scaled delta timeout.
|
||||||
|
//
|
||||||
|
// Long single fetches (commit, slow chunk PUTs) don't emit progress
|
||||||
|
// on their own — the worker sends `{ type: 'heartbeat' }` every 5 s
|
||||||
|
// while awaiting a network request so the watchdog stays fresh.
|
||||||
const STALL_MS = 20_000;
|
const STALL_MS = 20_000;
|
||||||
const armStall = () => {
|
const armStall = () => {
|
||||||
clearTimeout(stallTimer);
|
clearTimeout(stallTimer);
|
||||||
|
// Don't count time while the tab is hidden — background throttling
|
||||||
|
// on the main thread breaks the "no message in 20 s = wedged"
|
||||||
|
// premise. When the user comes back, the visibility listener
|
||||||
|
// re-arms.
|
||||||
|
if (typeof document !== 'undefined' && document.hidden) return;
|
||||||
stallTimer = setTimeout(() => {
|
stallTimer = setTimeout(() => {
|
||||||
usable = false;
|
usable = false;
|
||||||
uploadLog.error(
|
uploadLog.error(
|
||||||
@@ -169,11 +198,22 @@ export function tryDeltaUpload(
|
|||||||
settle(null);
|
settle(null);
|
||||||
}, STALL_MS);
|
}, STALL_MS);
|
||||||
};
|
};
|
||||||
|
if (typeof document !== 'undefined') {
|
||||||
|
visibilityListener = () => {
|
||||||
|
if (document.hidden) clearTimeout(stallTimer);
|
||||||
|
else armStall();
|
||||||
|
};
|
||||||
|
document.addEventListener('visibilitychange', visibilityListener);
|
||||||
|
}
|
||||||
armStall();
|
armStall();
|
||||||
|
|
||||||
worker.onmessage = (event: MessageEvent<WorkerMsg>) => {
|
worker.onmessage = (event: MessageEvent<WorkerMsg>) => {
|
||||||
armStall(); // worker is alive — reset the liveness watchdog
|
armStall(); // worker is alive — reset the liveness watchdog
|
||||||
const msg = event.data;
|
const msg = event.data;
|
||||||
|
if (msg.type === 'heartbeat') {
|
||||||
|
// armStall() above already served its purpose — no other work.
|
||||||
|
return;
|
||||||
|
}
|
||||||
if (msg.type === 'log') {
|
if (msg.type === 'log') {
|
||||||
// Relay worker log through the shared logger so runtime-set
|
// Relay worker log through the shared logger so runtime-set
|
||||||
// level (via `log.getLogger('oxi:upload').setLevel(...)`)
|
// level (via `log.getLogger('oxi:upload').setLevel(...)`)
|
||||||
|
|||||||
@@ -89,6 +89,26 @@ workerScope.onmessage = async (event) => {
|
|||||||
const log = (level, msg, extra) =>
|
const log = (level, msg, extra) =>
|
||||||
workerScope.postMessage({ type: 'log', level, msg, extra });
|
workerScope.postMessage({ type: 'log', level, msg, extra });
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Wrap a long-running fetch so the main-thread stall watchdog stays
|
||||||
|
* fresh while the worker is legitimately awaiting the network. The
|
||||||
|
* watchdog resets on every worker → main-thread message; heartbeats
|
||||||
|
* every 5 s keep it from firing during a slow commit or chunk PUT.
|
||||||
|
* Cleanup in `finally` runs regardless of resolve / reject.
|
||||||
|
*
|
||||||
|
* @template T
|
||||||
|
* @param {() => Promise<T>} fn
|
||||||
|
* @returns {Promise<T>}
|
||||||
|
*/
|
||||||
|
const withHeartbeat = async (fn) => {
|
||||||
|
const hb = setInterval(() => workerScope.postMessage({ type: 'heartbeat' }), 5000);
|
||||||
|
try {
|
||||||
|
return await fn();
|
||||||
|
} finally {
|
||||||
|
clearInterval(hb);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
/** @param {string} reason */
|
/** @param {string} reason */
|
||||||
const fallback = (reason) => {
|
const fallback = (reason) => {
|
||||||
log('warn', `worker fallback: ${reason}`);
|
log('warn', `worker fallback: ${reason}`);
|
||||||
@@ -204,14 +224,16 @@ workerScope.onmessage = async (event) => {
|
|||||||
const wire = await encodeFrames(batch);
|
const wire = await encodeFrames(batch);
|
||||||
log('debug', `chunk PUT: ${batch.length} chunks, ${wire.length} bytes`);
|
log('debug', `chunk PUT: ${batch.length} chunks, ${wire.length} bytes`);
|
||||||
// eslint-disable-next-line no-await-in-loop -- bounded by pool size
|
// eslint-disable-next-line no-await-in-loop -- bounded by pool size
|
||||||
const response = await fetch('/api/files/delta/chunks', {
|
const response = await withHeartbeat(() =>
|
||||||
method: 'PUT',
|
fetch('/api/files/delta/chunks', {
|
||||||
headers: {
|
method: 'PUT',
|
||||||
'Content-Type': 'application/octet-stream',
|
headers: {
|
||||||
...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {})
|
'Content-Type': 'application/octet-stream',
|
||||||
},
|
...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {})
|
||||||
body: wire
|
},
|
||||||
});
|
body: wire
|
||||||
|
})
|
||||||
|
);
|
||||||
if (!response.ok) {
|
if (!response.ok) {
|
||||||
failed = `chunk PUT failed (HTTP ${response.status})`;
|
failed = `chunk PUT failed (HTTP ${response.status})`;
|
||||||
log('error', failed);
|
log('error', failed);
|
||||||
@@ -242,11 +264,13 @@ workerScope.onmessage = async (event) => {
|
|||||||
const run = negotiateTail.then(async () => {
|
const run = negotiateTail.then(async () => {
|
||||||
if (failed) return;
|
if (failed) return;
|
||||||
try {
|
try {
|
||||||
const response = await fetch('/api/files/delta/negotiate', {
|
const response = await withHeartbeat(() =>
|
||||||
method: 'POST',
|
fetch('/api/files/delta/negotiate', {
|
||||||
headers: mutHeaders,
|
method: 'POST',
|
||||||
body: JSON.stringify({ chunks: fresh.map(({ h, s }) => ({ h, s })) })
|
headers: mutHeaders,
|
||||||
});
|
body: JSON.stringify({ chunks: fresh.map(({ h, s }) => ({ h, s })) })
|
||||||
|
})
|
||||||
|
);
|
||||||
if (!response.ok) {
|
if (!response.ok) {
|
||||||
failed = failed || `negotiate failed (HTTP ${response.status})`;
|
failed = failed || `negotiate failed (HTTP ${response.status})`;
|
||||||
log('error', `negotiate failed (HTTP ${response.status})`);
|
log('error', `negotiate failed (HTTP ${response.status})`);
|
||||||
@@ -343,11 +367,13 @@ workerScope.onmessage = async (event) => {
|
|||||||
};
|
};
|
||||||
for (let attempt = 0; ; attempt++) {
|
for (let attempt = 0; ; attempt++) {
|
||||||
// eslint-disable-next-line no-await-in-loop -- retry loop
|
// eslint-disable-next-line no-await-in-loop -- retry loop
|
||||||
const response = await fetch('/api/files/delta/commit', {
|
const response = await withHeartbeat(() =>
|
||||||
method: 'POST',
|
fetch('/api/files/delta/commit', {
|
||||||
headers: mutHeaders,
|
method: 'POST',
|
||||||
body: JSON.stringify(commitBody)
|
headers: mutHeaders,
|
||||||
});
|
body: JSON.stringify(commitBody)
|
||||||
|
})
|
||||||
|
);
|
||||||
/** @type {any} */
|
/** @type {any} */
|
||||||
let body = null;
|
let body = null;
|
||||||
try {
|
try {
|
||||||
@@ -372,14 +398,16 @@ workerScope.onmessage = async (event) => {
|
|||||||
}
|
}
|
||||||
const wire = await encodeFrames(retry);
|
const wire = await encodeFrames(retry);
|
||||||
// eslint-disable-next-line no-await-in-loop -- retry loop
|
// eslint-disable-next-line no-await-in-loop -- retry loop
|
||||||
const put = await fetch('/api/files/delta/chunks', {
|
const put = await withHeartbeat(() =>
|
||||||
method: 'PUT',
|
fetch('/api/files/delta/chunks', {
|
||||||
headers: {
|
method: 'PUT',
|
||||||
'Content-Type': 'application/octet-stream',
|
headers: {
|
||||||
...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {})
|
'Content-Type': 'application/octet-stream',
|
||||||
},
|
...(csrfToken ? { 'X-CSRF-Token': csrfToken } : {})
|
||||||
body: wire
|
},
|
||||||
});
|
body: wire
|
||||||
|
})
|
||||||
|
);
|
||||||
if (!put.ok) {
|
if (!put.ok) {
|
||||||
fallback(`retry chunk PUT failed (HTTP ${put.status})`);
|
fallback(`retry chunk PUT failed (HTTP ${put.status})`);
|
||||||
return;
|
return;
|
||||||
|
|||||||
Reference in New Issue
Block a user