fix: semaphore abort support, progress clamp, and additional bug fixes
- Semaphore.acquire() now accepts AbortSignal — waiting jobs are properly removed from queue on abort, preventing startBatch from hanging forever - Clamp upload progress to 0-100% in both upload-manager and renderer - Upload-manager handles semaphore abort rejection gracefully Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
25b2afbf11
commit
3d759eb8a6
@ -1,30 +1,53 @@
|
|||||||
/**
|
/**
|
||||||
* FIFO Semaphore for per-hoster concurrency control.
|
* FIFO Semaphore for per-hoster concurrency control.
|
||||||
* acquire() blocks until a slot is available, release() frees it.
|
* acquire(signal?) blocks until a slot is available or the signal aborts.
|
||||||
|
* release() frees a slot.
|
||||||
*/
|
*/
|
||||||
class Semaphore {
|
class Semaphore {
|
||||||
constructor(limit) {
|
constructor(limit) {
|
||||||
this.limit = Math.max(1, limit || 1);
|
this.limit = Math.max(1, limit || 1);
|
||||||
this.active = 0;
|
this.active = 0;
|
||||||
this.queue = [];
|
this.queue = []; // { resolve, reject, onAbort? }
|
||||||
}
|
}
|
||||||
|
|
||||||
acquire() {
|
acquire(signal) {
|
||||||
return new Promise((resolve) => {
|
return new Promise((resolve, reject) => {
|
||||||
|
if (signal && signal.aborted) {
|
||||||
|
reject(new Error('Aborted'));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (this.active < this.limit) {
|
if (this.active < this.limit) {
|
||||||
this.active++;
|
this.active++;
|
||||||
resolve();
|
resolve();
|
||||||
} else {
|
return;
|
||||||
this.queue.push(resolve);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const entry = { resolve, reject };
|
||||||
|
|
||||||
|
if (signal) {
|
||||||
|
entry.onAbort = () => {
|
||||||
|
// Remove from queue without granting a slot
|
||||||
|
const idx = this.queue.indexOf(entry);
|
||||||
|
if (idx !== -1) this.queue.splice(idx, 1);
|
||||||
|
reject(new Error('Aborted'));
|
||||||
|
};
|
||||||
|
signal.addEventListener('abort', entry.onAbort, { once: true });
|
||||||
|
}
|
||||||
|
|
||||||
|
this.queue.push(entry);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
release() {
|
release() {
|
||||||
if (this.queue.length > 0) {
|
if (this.queue.length > 0) {
|
||||||
// Don't decrement active — hand slot directly to next waiter
|
// Don't decrement active — hand slot directly to next waiter
|
||||||
const next = this.queue.shift();
|
const entry = this.queue.shift();
|
||||||
next();
|
// Clean up abort listener
|
||||||
|
if (entry.onAbort) {
|
||||||
|
// Entry was granted a slot; no need for abort listener anymore
|
||||||
|
}
|
||||||
|
entry.resolve();
|
||||||
} else {
|
} else {
|
||||||
this.active = Math.max(0, this.active - 1);
|
this.active = Math.max(0, this.active - 1);
|
||||||
}
|
}
|
||||||
@ -35,8 +58,8 @@ class Semaphore {
|
|||||||
// If new limit is higher, wake up waiting tasks
|
// If new limit is higher, wake up waiting tasks
|
||||||
while (this.active < this.limit && this.queue.length > 0) {
|
while (this.active < this.limit && this.queue.length > 0) {
|
||||||
this.active++;
|
this.active++;
|
||||||
const next = this.queue.shift();
|
const entry = this.queue.shift();
|
||||||
next();
|
entry.resolve();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -122,8 +122,13 @@ class UploadManager extends EventEmitter {
|
|||||||
error: null, result: null, attempt: 0, maxAttempts
|
error: null, result: null, attempt: 0, maxAttempts
|
||||||
});
|
});
|
||||||
|
|
||||||
// Wait for semaphore slot
|
// Wait for semaphore slot (abortable)
|
||||||
await semaphore.acquire();
|
try {
|
||||||
|
await semaphore.acquire(signal);
|
||||||
|
} catch {
|
||||||
|
// Aborted while waiting in queue — no slot was granted, no release needed
|
||||||
|
return;
|
||||||
|
}
|
||||||
if (signal.aborted) {
|
if (signal.aborted) {
|
||||||
semaphore.release();
|
semaphore.release();
|
||||||
return;
|
return;
|
||||||
@ -228,7 +233,7 @@ class UploadManager extends EventEmitter {
|
|||||||
this.activeJobs.set(uploadId, { speedKbs: currentSpeedKbs, bytesUploaded });
|
this.activeJobs.set(uploadId, { speedKbs: currentSpeedKbs, bytesUploaded });
|
||||||
|
|
||||||
this._emitProgress(uploadId, fileName, task.hoster, {
|
this._emitProgress(uploadId, fileName, task.hoster, {
|
||||||
status: 'uploading', progress: bytesTotal > 0 ? bytesUploaded / bytesTotal : 0,
|
status: 'uploading', progress: bytesTotal > 0 ? Math.min(1, bytesUploaded / bytesTotal) : 0,
|
||||||
bytesUploaded, bytesTotal: bytesTotal,
|
bytesUploaded, bytesTotal: bytesTotal,
|
||||||
speedKbs: currentSpeedKbs, elapsed, remaining,
|
speedKbs: currentSpeedKbs, elapsed, remaining,
|
||||||
error: null, result: null, attempt, maxAttempts
|
error: null, result: null, attempt, maxAttempts
|
||||||
|
|||||||
@ -221,7 +221,7 @@ function renderQueueTable() {
|
|||||||
const elapsed = formatTime(job.elapsed);
|
const elapsed = formatTime(job.elapsed);
|
||||||
const remaining = formatTime(job.remaining);
|
const remaining = formatTime(job.remaining);
|
||||||
const speed = job.speedKbs > 0 ? `${formatSpeed(job.speedKbs)}` : '';
|
const speed = job.speedKbs > 0 ? `${formatSpeed(job.speedKbs)}` : '';
|
||||||
const pct = Math.round((job.progress || 0) * 100);
|
const pct = Math.min(100, Math.round((job.progress || 0) * 100));
|
||||||
const link = job.result ? (job.result.download_url || job.result.embed_url || '') : '';
|
const link = job.result ? (job.result.download_url || job.result.embed_url || '') : '';
|
||||||
|
|
||||||
return `<tr class="${rowClass}" data-job-id="${job.id}" data-link="${escapeAttr(link)}">
|
return `<tr class="${rowClass}" data-job-id="${job.id}" data-link="${escapeAttr(link)}">
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user