Compare commits

..

3 Commits

Author SHA1 Message Date
Alex Shnitman abb9492d21 upgrade dependencies 2026-04-21 16:20:39 +03:00
Alex Shnitman 23de9824f0 cr fixes 2026-04-21 16:13:58 +03:00
rdiaz738 0ea934c08f Updated import and fixed race condition 2026-04-20 17:24:16 -07:00
6 changed files with 390 additions and 291 deletions
+43
View File
@@ -114,6 +114,49 @@ async def test_cancel_removes_from_pending(dq_env):
notifier.canceled.assert_awaited()
@pytest.mark.asyncio
async def test_cancel_before_start_marks_download_canceled(dq_env):
"""Regression test for the race condition where cancel() arrives after the
download has been placed in the queue and ``__start_download`` has been
scheduled via ``asyncio.create_task`` but has not yet executed. Without the
fix, the pending task would run ``download.start()`` despite the user
cancelling, because its ``download.canceled`` guard was never flipped."""
notifier = AsyncMock()
def fake_extract(self, url, ytdl_options_presets=None, ytdl_options_overrides=None):
return {
"_type": "video",
"id": "vid1",
"title": "Test Video",
"url": url,
"webpage_url": url,
}
dq = DownloadQueue(dq_env, notifier)
url = "https://example.com/race"
start_mock = AsyncMock()
with patch.object(DownloadQueue, "_DownloadQueue__extract_info", fake_extract), \
patch.object(DownloadQueue, "_DownloadQueue__start_download", start_mock):
await dq.add(
url,
"video",
"auto",
"any",
"best",
"",
"",
0,
auto_start=True,
)
assert dq.queue.exists(url)
download = dq.queue.get(url)
assert download.canceled is False
await dq.cancel([url])
assert not dq.queue.exists(url)
assert download.canceled is True
notifier.canceled.assert_awaited_with(url)
@pytest.mark.asyncio
async def test_start_pending_moves_to_queue(dq_env):
notifier = AsyncMock()
+4 -2
View File
@@ -1140,9 +1140,11 @@ class DownloadQueue:
if not self.queue.exists(id):
log.warning(f'requested cancel for non-existent download {id}')
continue
if self.queue.get(id).started():
self.queue.get(id).cancel()
dl = self.queue.get(id)
if dl.started():
dl.cancel()
else:
dl.canceled = True
self.queue.delete(id)
await self.notifier.canceled(id)
return {'status': 'ok'}
+1 -1
View File
@@ -58,6 +58,6 @@
"jsdom": "^27.4.0",
"typescript": "~5.9.3",
"typescript-eslint": "8.47.0",
"vitest": "^4.1.4"
"vitest": "^4.1.5"
}
}
+266 -249
View File
File diff suppressed because it is too large Load Diff
+13
View File
@@ -608,6 +608,19 @@
@if (batchImportStatus) {
<small>{{ batchImportStatus }}</small>
}
@if (batchImportTotal > 0) {
<div class="progress mt-2" style="height: 20px;">
<div class="progress-bar"
[class.bg-danger]="batchImportFailures > 0"
role="progressbar"
[style.width.%]="(batchImportCount / batchImportTotal) * 100"
[attr.aria-valuenow]="batchImportCount"
[attr.aria-valuemin]="0"
[attr.aria-valuemax]="batchImportTotal">
{{ batchImportCount }} / {{ batchImportTotal }}
</div>
</div>
}
</div>
</div>
<div class="modal-footer">
+63 -39
View File
@@ -1,7 +1,7 @@
import { AsyncPipe, DatePipe, KeyValuePipe, NgTemplateOutlet } from '@angular/common';
import { HttpClient } from '@angular/common/http';
import { AfterViewInit, ChangeDetectionStrategy, ChangeDetectorRef, Component, DestroyRef, ElementRef, viewChild, inject, OnDestroy, OnInit } from '@angular/core';
import { Observable, Subscription, map, distinctUntilChanged, finalize } from 'rxjs';
import { Observable, Subject, Subscription, from, map, distinctUntilChanged, finalize, mergeMap, takeUntil, tap } from 'rxjs';
import { FormsModule } from '@angular/forms';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
import { FontAwesomeModule } from '@fortawesome/angular-fontawesome';
@@ -104,8 +104,15 @@ export class App implements AfterViewInit, OnInit, OnDestroy {
batchImportModalOpen = false;
batchImportText = '';
batchImportStatus = '';
batchImportCount = 0;
batchImportTotal = 0;
batchImportFailures = 0;
importInProgress = false;
cancelImportFlag = false;
private batchImportCancel$ = new Subject<void>();
// Maximum number of /add requests to have in-flight at once during a batch
// import. Keeps the server from being hit with hundreds of simultaneous
// yt-dlp metadata extractions when a user pastes a huge URL list.
private static readonly BATCH_IMPORT_CONCURRENCY = 4;
ytDlpOptionsUpdateTime: string | null = null;
ytDlpVersion: string | null = null;
metubeVersion: string | null = null;
@@ -1173,8 +1180,10 @@ export class App implements AfterViewInit, OnInit, OnDestroy {
this.batchImportModalOpen = true;
this.batchImportText = '';
this.batchImportStatus = '';
this.batchImportCount = 0;
this.batchImportTotal = 0;
this.batchImportFailures = 0;
this.importInProgress = false;
this.cancelImportFlag = false;
setTimeout(() => {
const textarea = document.getElementById('batch-import-textarea');
if (textarea instanceof HTMLTextAreaElement) {
@@ -1200,48 +1209,63 @@ export class App implements AfterViewInit, OnInit, OnDestroy {
return;
}
this.importInProgress = true;
this.cancelImportFlag = false;
this.batchImportStatus = `Starting to import ${urls.length} URLs...`;
let index = 0;
const delayBetween = 1000;
const processNext = () => {
if (this.cancelImportFlag) {
this.batchImportStatus = `Import cancelled after ${index} of ${urls.length} URLs.`;
this.importInProgress = false;
return;
}
if (index >= urls.length) {
this.batchImportStatus = `Finished importing ${urls.length} URLs.`;
this.importInProgress = false;
return;
}
const url = urls[index];
this.batchImportStatus = `Importing URL ${index + 1} of ${urls.length}: ${url}`;
// Pass current selection options to backend
this.downloads.add(this.buildAddPayload({ url }))
.subscribe({
next: (status: Status) => {
this.batchImportCount = 0;
this.batchImportFailures = 0;
this.batchImportTotal = urls.length;
this.updateBatchImportStatus();
from(urls).pipe(
mergeMap(
url => this.downloads.add(this.buildAddPayload({ url })).pipe(
// downloads.add() already catches HTTP errors and emits a single
// Status value, so `tap` (not `finalize`) is the right place to
// count. This avoids incrementing the counter when an in-flight
// request is aborted by cancellation.
tap((status: Status) => {
if (status.status === 'error') {
alert(`Error adding URL ${url}: ${status.msg}`);
this.batchImportFailures++;
console.error(`Error adding URL ${url}: ${status.msg}`);
}
index++;
setTimeout(processNext, delayBetween);
},
error: (err) => {
console.error(`Error importing URL ${url}:`, err);
index++;
setTimeout(processNext, delayBetween);
}
});
};
processNext();
this.batchImportCount++;
this.updateBatchImportStatus();
this.cdr.markForCheck();
}),
),
App.BATCH_IMPORT_CONCURRENCY,
),
takeUntil(this.batchImportCancel$),
takeUntilDestroyed(this.destroyRef),
finalize(() => {
this.importInProgress = false;
this.updateBatchImportStatus(true);
this.cdr.markForCheck();
}),
).subscribe();
}
// Cancel the batch import process
private updateBatchImportStatus(done = false): void {
const parts: string[] = [];
if (done) {
const processed = this.batchImportCount;
if (processed < this.batchImportTotal) {
parts.push(`Import cancelled after ${processed} of ${this.batchImportTotal} URLs.`);
} else {
parts.push(`Finished importing ${this.batchImportTotal} URLs.`);
}
} else {
parts.push(`Importing ${this.batchImportCount} of ${this.batchImportTotal} URLs...`);
}
if (this.batchImportFailures > 0) {
parts.push(`${this.batchImportFailures} failed.`);
}
this.batchImportStatus = parts.join(' ');
}
// Cancel the batch import process: aborts in-flight and pending requests
// immediately via the cancellation Subject wired into the pipeline.
cancelBatchImport(): void {
if (this.importInProgress) {
this.cancelImportFlag = true;
this.batchImportStatus += ' Cancelling...';
this.batchImportCancel$.next();
}
}