From 5b9a751baddb25148342fe5a867ce3485e9738fa Mon Sep 17 00:00:00 2001 From: Roz Date: Sun, 26 Jul 2026 01:33:42 +0200 Subject: [PATCH 1/4] feat: add download management actions --- apps/backend/drizzle/0010_calm_cyclops.sql | 8 + apps/backend/drizzle/meta/0010_snapshot.json | 370 ++++++++++++++++++ apps/backend/drizzle/meta/_journal.json | 7 + apps/backend/src/__tests__/database.test.ts | 2 + .../src/__tests__/downloads-service.test.ts | 281 +++++++++++++ .../src/__tests__/import-watcher.test.ts | 65 +++ .../__tests__/management-downloads.test.ts | 91 +++++ .../src/__tests__/peer-download.test.ts | 52 +++ .../src/__tests__/qbittorrent-mapper.test.ts | 2 + apps/backend/src/__tests__/retry.test.ts | 21 + apps/backend/src/__tests__/semaphore.test.ts | 28 ++ apps/backend/src/database/schema.ts | 4 + apps/backend/src/index.ts | 19 +- apps/backend/src/lib/retry.ts | 36 +- apps/backend/src/lib/semaphore.ts | 27 +- apps/backend/src/lib/servers/peer.ts | 7 +- apps/backend/src/management-app.ts | 10 + .../download-operation-coordinator.ts | 55 +++ .../modules/downloads/downloads.controller.ts | 44 +++ .../modules/downloads/downloads.repository.ts | 39 +- .../src/modules/downloads/downloads.router.ts | 22 ++ .../modules/downloads/downloads.service.ts | 126 +++++- .../src/modules/downloads/import-watcher.ts | 218 +++++++---- apps/ui/app/components/DownloadActions.vue | 41 ++ apps/ui/app/pages/downloads.vue | 60 ++- apps/ui/app/types/management.ts | 4 + apps/ui/app/utils/download-actions.test.ts | 45 +++ apps/ui/app/utils/download-actions.ts | 14 + 28 files changed, 1589 insertions(+), 109 deletions(-) create mode 100644 apps/backend/drizzle/0010_calm_cyclops.sql create mode 100644 apps/backend/drizzle/meta/0010_snapshot.json create mode 100644 apps/backend/src/__tests__/management-downloads.test.ts create mode 100644 apps/backend/src/modules/downloads/download-operation-coordinator.ts create mode 100644 apps/backend/src/modules/downloads/downloads.controller.ts create mode 100644 apps/backend/src/modules/downloads/downloads.router.ts create mode 100644 apps/ui/app/components/DownloadActions.vue create mode 100644 apps/ui/app/utils/download-actions.test.ts create mode 100644 apps/ui/app/utils/download-actions.ts diff --git a/apps/backend/drizzle/0010_calm_cyclops.sql b/apps/backend/drizzle/0010_calm_cyclops.sql new file mode 100644 index 0000000..71ee707 --- /dev/null +++ b/apps/backend/drizzle/0010_calm_cyclops.sql @@ -0,0 +1,8 @@ +ALTER TABLE `downloads` ADD `last_operation` text DEFAULT 'transfer' NOT NULL;--> statement-breakpoint +ALTER TABLE `downloads` ADD `operation_failed` integer DEFAULT false NOT NULL;--> statement-breakpoint +UPDATE `downloads` +SET `last_operation` = CASE + WHEN `completed_at` IS NOT NULL OR `status` IN ('import_queued', 'imported') THEN 'import' + ELSE 'transfer' +END, +`operation_failed` = CASE WHEN `status` = 'failed' THEN true ELSE false END; diff --git a/apps/backend/drizzle/meta/0010_snapshot.json b/apps/backend/drizzle/meta/0010_snapshot.json new file mode 100644 index 0000000..1d70155 --- /dev/null +++ b/apps/backend/drizzle/meta/0010_snapshot.json @@ -0,0 +1,370 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "14803553-bed5-440d-b84a-a1c74743c831", + "prevId": "868c8329-4d5e-4af6-bbca-e9380502b247", + "tables": { + "api_keys": { + "name": "api_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "api_keys_key_hash_unique": { + "name": "api_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "downloads": { + "name": "downloads", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "torrent_filename": { + "name": "torrent_filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "item_id": { + "name": "item_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "dest_path": { + "name": "dest_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "part_path": { + "name": "part_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_size": { + "name": "release_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_json": { + "name": "release_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expected_bytes": { + "name": "expected_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_source": { + "name": "expected_bytes_source", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_mismatch": { + "name": "expected_bytes_mismatch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "downloaded_bytes": { + "name": "downloaded_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "attempts": { + "name": "attempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "completed_at": { + "name": "completed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_operation": { + "name": "last_operation", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'transfer'" + }, + "operation_failed": { + "name": "operation_failed", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "qb_category": { + "name": "qb_category", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "qb_source_server": { + "name": "qb_source_server", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "source_server_id": { + "name": "source_server_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_mode": { + "name": "import_mode", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_target": { + "name": "import_target", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "manual_import_command_id": { + "name": "manual_import_command_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "downloads_status_idx": { + "name": "downloads_status_idx", + "columns": [ + "status" + ], + "isUnique": false + }, + "downloads_updated_at_idx": { + "name": "downloads_updated_at_idx", + "columns": [ + "updated_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "downloads_status_check": { + "name": "downloads_status_check", + "value": "\"downloads\".\"status\" in ('downloading', 'import_queued', 'imported', 'failed')" + }, + "downloads_expected_bytes_source_check": { + "name": "downloads_expected_bytes_source_check", + "value": "\"downloads\".\"expected_bytes_source\" is null or \"downloads\".\"expected_bytes_source\" in ('content_length', 'content_range', 'release_size')" + }, + "downloads_last_operation_check": { + "name": "downloads_last_operation_check", + "value": "\"downloads\".\"last_operation\" in ('transfer', 'import')" + } + } + }, + "managed_keys": { + "name": "managed_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "server_id": { + "name": "server_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "managed_keys_key_hash_unique": { + "name": "managed_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} diff --git a/apps/backend/drizzle/meta/_journal.json b/apps/backend/drizzle/meta/_journal.json index 040233b..5c62aff 100644 --- a/apps/backend/drizzle/meta/_journal.json +++ b/apps/backend/drizzle/meta/_journal.json @@ -71,6 +71,13 @@ "when": 1782647842534, "tag": "0009_bumpy_ozymandias", "breakpoints": true + }, + { + "idx": 10, + "version": "6", + "when": 1785019399084, + "tag": "0010_calm_cyclops", + "breakpoints": true } ] } diff --git a/apps/backend/src/__tests__/database.test.ts b/apps/backend/src/__tests__/database.test.ts index 1366cb5..4bdfb8a 100644 --- a/apps/backend/src/__tests__/database.test.ts +++ b/apps/backend/src/__tests__/database.test.ts @@ -172,6 +172,8 @@ describe('DownloadsRepository', () => { expect(stale.status).toBe('failed') expect(stale.downloadedBytes).toBe(4) expect(stale.error).toContain('stale') + expect(stale.lastOperation).toBe('transfer') + expect(stale.operationFailed).toBe(true) handle.close() }) diff --git a/apps/backend/src/__tests__/downloads-service.test.ts b/apps/backend/src/__tests__/downloads-service.test.ts index cad39e4..a0d5928 100644 --- a/apps/backend/src/__tests__/downloads-service.test.ts +++ b/apps/backend/src/__tests__/downloads-service.test.ts @@ -7,6 +7,7 @@ import { openDatabase } from '../database/connection' import { FetchError } from '../lib/errors/FetchError' import { DownloadsRepository } from '../modules/downloads/downloads.repository' import { DownloadsService } from '../modules/downloads/downloads.service' +import { ImportWatcher } from '../modules/downloads/import-watcher' const release: Release = { id: 'remote:movie:1', @@ -66,6 +67,286 @@ async function waitForStatus(repository: DownloadsRepository, status: string) { } describe('DownloadsService download progress persistence', () => { + test('cancel settles a transfer queued behind the concurrency semaphore', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const started = Promise.withResolvers() + const peer = { + ...fakePeer(), + getRelease: async (itemId: string) => ({ ...release, filename: `${itemId}.mkv` }), + downloadFile: async (itemId: string, _destPath: string, options: any) => { + if (itemId === 'one') { + started.resolve() + await new Promise((_resolve, reject) => options.signal.addEventListener('abort', () => reject(options.signal.reason), { once: true })) + } + throw new Error('second transfer must not start') + }, + } + const service = new DownloadsService(downloadsConfig({ maxConcurrentDownloads: 1 }), { peers: [peer as any] }, repository) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'one', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + await started.promise + await service.startQbDownload({ peerId: 'peer-1', itemId: 'two', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + const two = repository.list().find(row => row.itemId === 'two')! + + await service.cancel(two.id) + + expect(repository.get(two.id)).toMatchObject({ status: 'failed', operationFailed: true, lastOperation: 'transfer' }) + await service.cancel(repository.list().find(row => row.itemId === 'one')!.id) + handle.close() + }, 1000) + + test('cancel settles immediately while a transient failure is in retry backoff', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const attempted = Promise.withResolvers() + const peer = fakePeer({ downloadFile: async () => { + attempted.resolve() + throw new FetchError('busy', new Response(null, { status: 503 })) + } }) + const service = new DownloadsService(downloadsConfig({ retryBaseDelayMs: 60_000, retryMaxDelayMs: 60_000 }), { peers: [peer as any] }, repository) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'one', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + await attempted.promise + const row = repository.list()[0]! + + await service.cancel(row.id) + + expect(repository.get(row.id)).toMatchObject({ status: 'failed', operationFailed: true, lastOperation: 'transfer' }) + handle.close() + }, 1000) + + test('cancel after downloadFile completes cannot queue the import', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const finalizing = Promise.withResolvers() + const finish = Promise.withResolvers() + const peer = fakePeer({ downloadFile: async (_itemId: string, _destPath: string, options: any) => { + await options.onProgress({ type: 'completed', downloadedBytes: 10, expectedBytes: 10 }) + finalizing.resolve() + await finish.promise + } }) + const service = new DownloadsService(downloadsConfig(), { peers: [peer as any] }, repository) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'one', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + await finalizing.promise + const row = repository.list()[0]! + const cancellation = service.cancel(row.id) + finish.resolve() + + await cancellation + + expect(repository.get(row.id)).toMatchObject({ status: 'failed', operationFailed: true, lastOperation: 'transfer' }) + handle.close() + }) + + test('cancel aborts only the numeric download id and preserves its partial file for retry', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const started = new Map void>() + const peer = { + ...fakePeer(), + getRelease: async (itemId: string) => ({ ...release, filename: `${itemId}.mkv` }), + downloadFile: async (itemId: string, _destPath: string, options: any) => { + await Bun.write(options.partPath, itemId) + await options.onProgress({ type: 'progress', downloadedBytes: itemId.length, expectedBytes: 10 }) + await new Promise((resolve, reject) => { + started.set(itemId, resolve) + options.signal.addEventListener('abort', () => reject(options.signal.reason), { once: true }) + }) + }, + } + const service = new DownloadsService(downloadsConfig(), { peers: [peer as any] }, repository) + + await service.startQbDownload({ peerId: 'peer-1', itemId: 'one', qbCategory: 'jack-x', qbSourceServer: 'Radarr', sourceServerId: 'radarr-1' }) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'two', qbCategory: 'jack-x', qbSourceServer: 'Radarr', sourceServerId: 'radarr-1' }) + for (let i = 0; i < 50 && started.size < 2; i++) + await Bun.sleep(10) + const rows = repository.list() + const one = rows.find(row => row.itemId === 'one')! + const two = rows.find(row => row.itemId === 'two')! + + await service.cancel(one.id) + + expect(repository.get(one.id)).toMatchObject({ status: 'failed', lastOperation: 'transfer' }) + expect(repository.get(one.id)?.error).toContain('cancelled') + expect(await Bun.file(one.partPath).exists()).toBe(true) + expect(repository.get(two.id)?.status).toBe('downloading') + started.get('two')?.() + await waitForStatus(repository, 'import_queued') + handle.close() + }) + + test('retry re-enqueues the failed transfer in place and resumes from its partial path', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const seenPartPaths: string[] = [] + let calls = 0 + const peer = fakePeer({ + downloadFile: async (_itemId: string, _destPath: string, options: any) => { + calls++ + seenPartPaths.push(options.partPath) + if (calls === 1) { + await Bun.write(options.partPath, 'part') + throw new FetchError('gone', new Response(null, { status: 404 })) + } + expect(await Bun.file(options.partPath).text()).toBe('part') + await options.onProgress({ type: 'completed', downloadedBytes: 10, expectedBytes: 10 }) + }, + }) + const service = new DownloadsService(downloadsConfig(), { peers: [peer as any] }, repository) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'movie:1', qbCategory: 'jack-x', qbSourceServer: 'Radarr', sourceServerId: 'radarr-1' }) + await waitForStatus(repository, 'failed') + const row = repository.list()[0]! + + await service.retry(row.id) + await waitForStatus(repository, 'import_queued') + + expect(repository.get(row.id)?.status).toBe('import_queued') + expect(repository.list()).toHaveLength(1) + expect(seenPartPaths).toEqual([row.partPath, row.partPath]) + handle.close() + }) + + test('delete cancels an active transfer and removes only that row and its artifacts', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const started = new Set() + const peer = { + ...fakePeer(), + getRelease: async (itemId: string) => ({ ...release, filename: `${itemId}.mkv` }), + downloadFile: async (itemId: string, destPath: string, options: any) => { + started.add(itemId) + await Bun.write(options.partPath, 'partial') + await Bun.write(destPath, 'completed') + await new Promise((_resolve, reject) => { + options.signal.addEventListener('abort', () => reject(options.signal.reason), { once: true }) + }) + }, + } + const service = new DownloadsService(downloadsConfig(), { peers: [peer as any] }, repository) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'one', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + await service.startQbDownload({ peerId: 'peer-1', itemId: 'two', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'r' }) + for (let i = 0; i < 50 && started.size < 2; i++) + await Bun.sleep(10) + const one = repository.list().find(row => row.itemId === 'one')! + const two = repository.list().find(row => row.itemId === 'two')! + + await service.delete(one.id) + + expect(repository.get(one.id)).toBeNull() + expect(await Bun.file(one.partPath).exists()).toBe(false) + expect(await Bun.file(one.destPath).exists()).toBe(false) + expect(repository.get(two.id)?.status).toBe('downloading') + expect(await Bun.file(two.partPath).exists()).toBe(true) + await service.cancel(two.id) + handle.close() + }) + + test('delete waits for import work and reserves its artifact paths', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const row = repository.create({ + torrentFilename: 'old.torrent', + peerId: 'peer-1', + peerName: 'Friend Jack', + itemId: 'old', + filename: release.filename, + destPath: join(completedPath, release.filename), + partPath: `${join(completedPath, release.filename)}.part`, + releaseSize: release.size, + release, + qbSourceServer: 'Radarr', + sourceServerId: 'radarr-1', + importMode: 'jack_manual', + importTarget: { kind: 'movie', movieId: 42 }, + }) + repository.markImportQueued(row.id) + repository.markFailed(row.id, 'import rejected', 'import') + await Bun.write(row.destPath, 'completed') + const started = Promise.withResolvers() + const finish = Promise.withResolvers() + const server = { + id: 'radarr-1', + name: 'Radarr', + isInitialized: true, + recentlyImportedDownloadIds: async () => new Set(), + manualImport: async () => { + started.resolve() + await finish.promise + return 42 + }, + manualImportCommandStatus: async () => ({ state: 'pending' as const }), + } + const watcher = new ImportWatcher(repository, { servers: [server as any] }, 1000) + const service = new DownloadsService(downloadsConfig(), { peers: [fakePeer() as any] }, repository) + + const retrying = watcher.retry(row.id) + await started.promise + const deleting = service.delete(row.id) + await Bun.sleep(10) + + expect(repository.get(row.id)).not.toBeNull() + expect(await Bun.file(row.destPath).exists()).toBe(true) + expect(await service.startQbDownload({ peerId: 'peer-1', itemId: 'replacement', qbCategory: 'x', qbSourceServer: 'Radarr', sourceServerId: 'radarr-1' })).toBe('duplicate') + + finish.resolve() + await retrying + await deleting + expect(repository.get(row.id)).toBeNull() + expect(await Bun.file(row.destPath).exists()).toBe(false) + expect(repository.list()).toHaveLength(0) + handle.close() + }) + + test('delete waits for a watcher tick before removing its artifact', async () => { + const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) + const repository = new DownloadsRepository(handle.db) + const row = repository.create({ + torrentFilename: 'old.torrent', + peerId: 'peer-1', + peerName: 'Friend Jack', + itemId: 'old', + filename: release.filename, + destPath: join(completedPath, release.filename), + partPath: `${join(completedPath, release.filename)}.part`, + releaseSize: release.size, + release, + qbSourceServer: 'Radarr', + sourceServerId: 'radarr-1', + importMode: 'jack_manual', + importTarget: { kind: 'movie', movieId: 42 }, + }) + repository.markImportQueued(row.id) + await Bun.write(row.destPath, 'completed') + const started = Promise.withResolvers() + const finish = Promise.withResolvers() + const server = { + id: 'radarr-1', + name: 'Radarr', + isInitialized: true, + recentlyImportedDownloadIds: async () => new Set(), + manualImport: async () => { + started.resolve() + await finish.promise + return 43 + }, + manualImportCommandStatus: async () => ({ state: 'pending' as const }), + } + const watcher = new ImportWatcher(repository, { servers: [server as any] }, 1000) + const service = new DownloadsService(downloadsConfig(), { peers: [fakePeer() as any] }, repository) + + const ticking = watcher.tick() + await started.promise + const deleting = service.delete(row.id) + await Bun.sleep(10) + expect(await Bun.file(row.destPath).exists()).toBe(true) + + finish.resolve() + await ticking + await deleting + expect(repository.get(row.id)).toBeNull() + expect(await Bun.file(row.destPath).exists()).toBe(false) + handle.close() + }) + test('creates a row only after release metadata resolves and moves it to import_queued', async () => { const handle = await openDatabase({ appConfigPath: join(tempDir, 'config.jsonc') }) const repository = new DownloadsRepository(handle.db) diff --git a/apps/backend/src/__tests__/import-watcher.test.ts b/apps/backend/src/__tests__/import-watcher.test.ts index aad7151..8bef8e9 100644 --- a/apps/backend/src/__tests__/import-watcher.test.ts +++ b/apps/backend/src/__tests__/import-watcher.test.ts @@ -129,6 +129,71 @@ function manualServer(name: string, importedHashes: string[], manualImport: (par } describe('ImportWatcher jack_manual trigger', () => { + test('coalesces overlapping ticks so they trigger one manual import', async () => { + const repo = makeRepo() + const row = manualRow(repo, 'My Radarr') + const started = Promise.withResolvers() + const finish = Promise.withResolvers() + const manualImport = mock(async () => { + started.resolve() + await finish.promise + return 106 + }) + const watcher = new ImportWatcher(repo, { servers: [manualServer('My Radarr', [], manualImport)] }, 1000) + + const first = watcher.tick() + await started.promise + const second = watcher.tick() + finish.resolve() + await Promise.all([first, second]) + + expect(manualImport).toHaveBeenCalledTimes(1) + expect(repo.get(row.id)?.manualImportCommandId).toBe(106) + }) + + test('serializes a manual retry against a watcher tick and revalidates state', async () => { + const repo = makeRepo() + const row = manualRow(repo, 'My Radarr') + repo.markFailed(row.id, 'manual import rejected', 'import') + const started = Promise.withResolvers() + const finish = Promise.withResolvers() + const manualImport = mock(async () => { + started.resolve() + await finish.promise + return 107 + }) + const watcher = new ImportWatcher(repo, { servers: [manualServer('My Radarr', [], manualImport)] }, 1000) + + const retrying = watcher.retry(row.id) + await started.promise + const ticking = watcher.tick() + finish.resolve() + await Promise.all([retrying, ticking]) + + expect(manualImport).toHaveBeenCalledTimes(1) + expect(repo.get(row.id)?.manualImportCommandId).toBe(107) + }) + + test('retry re-triggers a failed manual import without changing transfer bytes', async () => { + const repo = makeRepo() + const row = manualRow(repo, 'My Radarr') + repo.updateProgress(row.id, release.size) + repo.markFailed(row.id, 'manual import rejected', 'import') + const manualImport = mock(async () => 105) + const watcher = new ImportWatcher(repo, { servers: [manualServer('My Radarr', [], manualImport)] }, 1000) + + await watcher.retry(row.id) + + expect(manualImport).toHaveBeenCalledTimes(1) + expect(repo.get(row.id)).toMatchObject({ + status: 'import_queued', + downloadedBytes: release.size, + lastOperation: 'import', + operationFailed: false, + manualImportCommandId: 105, + }) + }) + test('pushes manualImport once across two ticks while the hash is absent from history', async () => { const repo = makeRepo() const row = manualRow(repo, 'My Radarr') diff --git a/apps/backend/src/__tests__/management-downloads.test.ts b/apps/backend/src/__tests__/management-downloads.test.ts new file mode 100644 index 0000000..3480c85 --- /dev/null +++ b/apps/backend/src/__tests__/management-downloads.test.ts @@ -0,0 +1,91 @@ +import { Database } from 'bun:sqlite' +import { describe, expect, mock, test } from 'bun:test' +import { drizzle } from 'drizzle-orm/bun-sqlite' +import { runMigrations } from '../database/connection' +import * as schema from '../database/schema' +import { getManagementApp } from '../management-app' +import { DownloadsRepository } from '../modules/downloads/downloads.repository' + +const HEADERS = { 'X-Management-Key': 'secret' } + +function setup() { + const sqlite = new Database(':memory:') + const db = drizzle({ client: sqlite, schema }) + runMigrations(db) + const repository = new DownloadsRepository(db) + const row = repository.create({ + torrentFilename: 'one.torrent', + peerId: 'peer-1', + peerName: 'Peer', + itemId: 'movie:1', + filename: 'one.mkv', + destPath: '/tmp/one.mkv', + partPath: '/tmp/one.mkv.part', + releaseSize: 10, + release: { id: 'r', title: 'one', filename: 'one.mkv', category: 2000, size: 10 } as any, + }) + const cancel = mock(async (id: number) => repository.get(id)!) + const retryTransfer = mock((id: number) => repository.get(id)!) + const deleteDownload = mock(async (id: number) => repository.delete(id)) + const retryImport = mock(async (id: number) => repository.get(id)!) + const downloadsService = { cancel, retry: retryTransfer, delete: deleteDownload } as any + const app = getManagementApp({ + environment: 'test', + managementKey: 'secret', + connectors: { peers: [], servers: [] }, + downloadsRepository: repository, + downloadsService, + importWatcher: { retry: retryImport } as any, + }) + return { app, cancel, retryTransfer, retryImport, deleteDownload, repository, row, sqlite } +} + +describe('management download actions', () => { + test('POST /downloads/:id/cancel targets the numeric record id', async () => { + const { app, cancel, row, sqlite } = setup() + + const response = await app.request(`/downloads/${row.id}/cancel`, { method: 'POST', headers: HEADERS }) + + expect(response.status).toBe(200) + expect(cancel).toHaveBeenCalledWith(row.id) + sqlite.close() + }) + + test('rejects a non-numeric download id', async () => { + const { app, cancel, sqlite } = setup() + + const response = await app.request('/downloads/not-a-number/cancel', { method: 'POST', headers: HEADERS }) + + expect(response.status).toBe(400) + expect(cancel).not.toHaveBeenCalled() + sqlite.close() + }) + + test('POST /downloads/:id/retry dispatches from the persisted last failed operation', async () => { + const { app, retryTransfer, retryImport, repository, row, sqlite } = setup() + repository.markFailed(row.id, 'transfer failed', 'transfer') + + const transferResponse = await app.request(`/downloads/${row.id}/retry`, { method: 'POST', headers: HEADERS }) + expect(transferResponse.status).toBe(200) + expect(retryTransfer).toHaveBeenCalledWith(row.id) + + repository.markImportQueued(row.id) + repository.markFailed(row.id, 'import failed', 'import') + // Retry selection is driven by the persisted failed operation, not the display status. + sqlite.query('UPDATE downloads SET status = ? WHERE id = ?').run('imported', row.id) + const importResponse = await app.request(`/downloads/${row.id}/retry`, { method: 'POST', headers: HEADERS }) + expect(importResponse.status).toBe(200) + expect(retryImport).toHaveBeenCalledWith(row.id) + sqlite.close() + }) + + test('DELETE /downloads/:id delegates deletion by record id', async () => { + const { app, deleteDownload, row, sqlite } = setup() + + const response = await app.request(`/downloads/${row.id}`, { method: 'DELETE', headers: HEADERS }) + + expect(response.status).toBe(200) + expect(deleteDownload).toHaveBeenCalledWith(row.id) + sqlite.close() + }) +}) diff --git a/apps/backend/src/__tests__/peer-download.test.ts b/apps/backend/src/__tests__/peer-download.test.ts index 638ee6f..996dfc8 100644 --- a/apps/backend/src/__tests__/peer-download.test.ts +++ b/apps/backend/src/__tests__/peer-download.test.ts @@ -245,6 +245,35 @@ describe('PeerConnector.downloadFile', () => { } }) + test('does not rename a completed part when cancelled during finalization', async () => { + server.use( + http.get(`${PEER_JACK_URL}/peer/items/:itemId/file`, () => + new Response(streamOf([1, 2, 3, 4]), { headers: { 'Content-Length': '4' } })), + ) + const peer = markInitialized(new PeerConnector({ url: PEER_JACK_URL, apiKey: 'peer-api-key', name: 'Friend Jack' })) + const controller = new AbortController() + const dir = await mkdtemp(join(tmpdir(), 'jack-peer-late-cancel-')) + const destPath = join(dir, 'Movie.mkv') + const partPath = `${destPath}.part` + + try { + await expect(peer.downloadFile('remote1:movie:99', destPath, { + partPath, + releaseSize: 4, + signal: controller.signal, + onProgress: (event) => { + if (event.type === 'progress') + controller.abort(new Error('cancelled before rename')) + }, + })).rejects.toThrow('cancelled before rename') + expect(await Bun.file(partPath).exists()).toBe(true) + expect(await Bun.file(destPath).exists()).toBe(false) + } + finally { + await rm(dir, { recursive: true, force: true }) + } + }) + test('does not leave the response body locked when opening the .part file fails', async () => { let body: Response['body'] = null const fetchSpy = spyOn(globalThis, 'fetch').mockImplementation((async () => { @@ -360,6 +389,29 @@ describe('PeerConnector.downloadFile', () => { }) describe('PeerConnector.downloadFile resume', () => { + test('does not finalize an already-complete part when the signal is aborted', async () => { + const peer = markInitialized(new PeerConnector({ url: PEER_JACK_URL, apiKey: 'peer-api-key', name: 'Friend Jack' })) + const controller = new AbortController() + controller.abort(new Error('cancelled before fast-path rename')) + const dir = await mkdtemp(join(tmpdir(), 'jack-resume-cancelled-')) + const destPath = join(dir, 'Movie.mkv') + const partPath = `${destPath}.part` + await writeFile(partPath, new Uint8Array([0, 1, 2, 3])) + + try { + await expect(peer.downloadFile('remote1:movie:99', destPath, { + partPath, + releaseSize: 4, + signal: controller.signal, + })).rejects.toThrow('cancelled before fast-path rename') + expect(await Bun.file(partPath).exists()).toBe(true) + expect(await Bun.file(destPath).exists()).toBe(false) + } + finally { + await rm(dir, { recursive: true, force: true }) + } + }) + test('resumes from an existing .part via a Range request and appends', async () => { const seen: { range: string | null } = { range: null } server.use( diff --git a/apps/backend/src/__tests__/qbittorrent-mapper.test.ts b/apps/backend/src/__tests__/qbittorrent-mapper.test.ts index 7d37c25..86d53e6 100644 --- a/apps/backend/src/__tests__/qbittorrent-mapper.test.ts +++ b/apps/backend/src/__tests__/qbittorrent-mapper.test.ts @@ -24,6 +24,8 @@ function baseRecord(overrides: Partial = {}): DownloadRecord { updatedAt: '2026-06-06T00:00:00.000Z', completedAt: null, error: null, + lastOperation: 'transfer', + operationFailed: false, qbCategory: 'jack-abc12345', qbSourceServer: 'My Radarr', sourceServerId: 'abc12345', diff --git a/apps/backend/src/__tests__/retry.test.ts b/apps/backend/src/__tests__/retry.test.ts index 2e1a978..8b5eb3b 100644 --- a/apps/backend/src/__tests__/retry.test.ts +++ b/apps/backend/src/__tests__/retry.test.ts @@ -103,6 +103,27 @@ describe('retry', () => { })).rejects.toThrow('maxAttempts must be at least 1') expect(calls).toBe(0) }) + + test('aborts immediately while sleeping between attempts', async () => { + const controller = new AbortController() + let calls = 0 + const result = retry(async () => { + calls++ + throw new Error('transient') + }, { + maxAttempts: 3, + baseDelayMs: 60_000, + maxDelayMs: 60_000, + isRetryable: () => true, + random: () => 1, + signal: controller.signal, + }) + + await Bun.sleep(0) + controller.abort(new Error('cancelled during backoff')) + await expect(result).rejects.toThrow('cancelled during backoff') + expect(calls).toBe(1) + }) }) describe('download retry backoff schedule', () => { diff --git a/apps/backend/src/__tests__/semaphore.test.ts b/apps/backend/src/__tests__/semaphore.test.ts index 8c7c461..863b888 100644 --- a/apps/backend/src/__tests__/semaphore.test.ts +++ b/apps/backend/src/__tests__/semaphore.test.ts @@ -39,6 +39,34 @@ describe('Semaphore', () => { expect(ran).toBe('next') }) + test('removes an aborted waiter without consuming the next released permit', async () => { + const sem = new Semaphore(1) + await sem.acquire() + const controller = new AbortController() + const aborted = sem.acquire(controller.signal) + const next = sem.acquire() + + controller.abort(new Error('cancelled while queued')) + await expect(aborted).rejects.toThrow('cancelled while queued') + sem.release() + await next + }) + + test('does not run work when its signal aborts while queued', async () => { + const sem = new Semaphore(1) + await sem.acquire() + const controller = new AbortController() + let ran = false + const queued = sem.run(async () => { + ran = true + }, controller.signal) + + controller.abort(new Error('cancelled while queued')) + await expect(queued).rejects.toThrow('cancelled while queued') + expect(ran).toBe(false) + sem.release() + }) + test('rejects a non-positive permit count', () => { expect(() => new Semaphore(0)).toThrow() }) diff --git a/apps/backend/src/database/schema.ts b/apps/backend/src/database/schema.ts index d808f92..d7677fe 100644 --- a/apps/backend/src/database/schema.ts +++ b/apps/backend/src/database/schema.ts @@ -6,6 +6,7 @@ import { check, index, integer, sqliteTable, text } from 'drizzle-orm/sqlite-cor export const DOWNLOAD_STATUSES = ['downloading', 'import_queued', 'imported', 'failed'] as const export type DownloadStatus = typeof DOWNLOAD_STATUSES[number] export type ExpectedBytesSource = 'content_length' | 'content_range' | 'release_size' +export type DownloadOperation = 'transfer' | 'import' export const downloads = sqliteTable('downloads', { id: integer('id').primaryKey({ autoIncrement: true }), @@ -28,6 +29,8 @@ export const downloads = sqliteTable('downloads', { updatedAt: text('updated_at').notNull(), completedAt: text('completed_at'), error: text('error'), + lastOperation: text('last_operation').$type().notNull().default('transfer'), + operationFailed: integer('operation_failed', { mode: 'boolean' }).notNull().default(false), // qBittorrent emulation: the category *arr sent on add, and the server // connector that added it. Presence of qbSourceServer marks a qB-added // download (→ *arr-pull import, no jack push). Null for blackhole-added rows. @@ -43,6 +46,7 @@ export const downloads = sqliteTable('downloads', { }, t => [ check('downloads_status_check', sql`${t.status} in ('downloading', 'import_queued', 'imported', 'failed')`), check('downloads_expected_bytes_source_check', sql`${t.expectedBytesSource} is null or ${t.expectedBytesSource} in ('content_length', 'content_range', 'release_size')`), + check('downloads_last_operation_check', sql`${t.lastOperation} in ('transfer', 'import')`), index('downloads_status_idx').on(t.status), index('downloads_updated_at_idx').on(t.updatedAt), ]) diff --git a/apps/backend/src/index.ts b/apps/backend/src/index.ts index 031a75c..9799d36 100644 --- a/apps/backend/src/index.ts +++ b/apps/backend/src/index.ts @@ -67,6 +67,16 @@ logger.info({ destinations: connectorManager.destinations.length, }, 'Server listening') +// Detect *arr imports of finished downloads and flip them import_queued → imported. +// Construct this before the management app so failed imports can be retried there. +const importWatcher = config.downloads + ? new ImportWatcher(downloadsRepository, connectorManager, config.downloads.importPollIntervalMs, { + maxAttempts: config.downloads.maxManualImportAttempts, + backoffBaseMs: config.downloads.manualImportBackoffBaseMs, + backoffMaxMs: config.downloads.manualImportBackoffMaxMs, + }) + : undefined + function startManagementServer() { if (!envs.MANAGEMENT_KEY) return undefined @@ -83,6 +93,7 @@ function startManagementServer() { configService, downloadsRepository, downloadsService, + importWatcher, apiKeysRepository, tmdbApiKey: config.jack.tmdbApiKey, }) @@ -127,14 +138,6 @@ for (const dest of registrable) { // Drop managed keys for destinations no longer registrable so stale keys can't authenticate. managedApiKeys.prune(registrable.map(d => d.id)) -// Detect *arr imports of finished downloads and flip them import_queued → imported. -const importWatcher = config.downloads - ? new ImportWatcher(downloadsRepository, connectorManager, config.downloads.importPollIntervalMs, { - maxAttempts: config.downloads.maxManualImportAttempts, - backoffBaseMs: config.downloads.manualImportBackoffBaseMs, - backoffMaxMs: config.downloads.manualImportBackoffMaxMs, - }) - : undefined importWatcher?.start() // Re-drive interrupted downloads from a prior run. diff --git a/apps/backend/src/lib/retry.ts b/apps/backend/src/lib/retry.ts index 59c01f3..635a2d9 100644 --- a/apps/backend/src/lib/retry.ts +++ b/apps/backend/src/lib/retry.ts @@ -12,10 +12,29 @@ export interface RetryOptions { sleep?: (ms: number) => Promise /** Injectable for tests (defaults to Math.random). */ random?: () => number + /** Cancels both pending backoff delays and subsequent attempts. */ + signal?: AbortSignal } const defaultSleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)) +function abortableDefaultSleep(ms: number, signal: AbortSignal): Promise { + signal.throwIfAborted() + return new Promise((resolve, reject) => { + const timer = setTimeout(finish, ms) + const onAbort = () => finish(signal.reason) + function finish(error?: unknown) { + clearTimeout(timer) + signal.removeEventListener('abort', onAbort) + if (error === undefined) + resolve() + else + reject(error) + } + signal.addEventListener('abort', onAbort, { once: true }) + }) +} + export async function retry(fn: (attempt: number) => Promise, options: RetryOptions): Promise { const { maxAttempts, baseDelayMs, maxDelayMs, isRetryable, retryAfterMs, onRetry } = options const sleep = options.sleep ?? defaultSleep @@ -26,6 +45,7 @@ export async function retry(fn: (attempt: number) => Promise, options: Ret for (let attempt = 1; ; attempt++) { try { + options.signal?.throwIfAborted() return await fn(attempt) } catch (error) { @@ -37,7 +57,21 @@ export async function retry(fn: (attempt: number) => Promise, options: Ret const backoff = Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1)) const delayMs = explicit != null ? Math.min(explicit, maxDelayMs) : random() * backoff onRetry?.({ attempt, delayMs, error }) - await sleep(delayMs) + if (options.signal && !options.sleep) { + await abortableDefaultSleep(delayMs, options.signal) + continue + } + const delay = sleep(delayMs) + if (!options.signal) { + await delay + continue + } + options.signal.throwIfAborted() + await new Promise((resolve, reject) => { + const onAbort = () => reject(options.signal!.reason) + options.signal!.addEventListener('abort', onAbort, { once: true }) + void delay.then(resolve, reject).finally(() => options.signal!.removeEventListener('abort', onAbort)) + }) } } } diff --git a/apps/backend/src/lib/semaphore.ts b/apps/backend/src/lib/semaphore.ts index 7f41f4c..0d6d228 100644 --- a/apps/backend/src/lib/semaphore.ts +++ b/apps/backend/src/lib/semaphore.ts @@ -1,7 +1,7 @@ /** A counting semaphore with FIFO fairness for limiting concurrent async work. */ export class Semaphore { private available: number - private readonly waiters: Array<() => void> = [] + private readonly waiters: Array<{ resolve: () => void, signal?: AbortSignal, onAbort?: () => void }> = [] constructor(permits: number) { if (!Number.isInteger(permits) || permits < 1) @@ -9,26 +9,41 @@ export class Semaphore { this.available = permits } - async acquire(): Promise { + async acquire(signal?: AbortSignal): Promise { + signal?.throwIfAborted() if (this.available > 0) { this.available-- return } - await new Promise(resolve => this.waiters.push(resolve)) + await new Promise((resolve, reject) => { + const waiter: { resolve: () => void, signal?: AbortSignal, onAbort?: () => void } = { resolve, signal } + if (signal) { + waiter.onAbort = () => { + const index = this.waiters.indexOf(waiter) + if (index >= 0) + this.waiters.splice(index, 1) + reject(signal.reason) + } + signal.addEventListener('abort', waiter.onAbort, { once: true }) + } + this.waiters.push(waiter) + }) } release(): void { const next = this.waiters.shift() if (next) { // Hand the permit straight to the next waiter (keeps `available` at 0). - next() + if (next.signal && next.onAbort) + next.signal.removeEventListener('abort', next.onAbort) + next.resolve() return } this.available++ } - async run(fn: () => Promise): Promise { - await this.acquire() + async run(fn: () => Promise, signal?: AbortSignal): Promise { + await this.acquire(signal) try { return await fn() } diff --git a/apps/backend/src/lib/servers/peer.ts b/apps/backend/src/lib/servers/peer.ts index ce92505..7f4c20f 100644 --- a/apps/backend/src/lib/servers/peer.ts +++ b/apps/backend/src/lib/servers/peer.ts @@ -27,6 +27,7 @@ export type PeerDownloadProgressEvent | { type: 'completed', downloadedBytes: number, expectedBytes: number | null } export interface PeerDownloadOptions { + signal?: AbortSignal idleTimeoutMs?: number torrentFilename?: string partPath?: string @@ -223,6 +224,7 @@ export class PeerConnector extends ServerConnector { // trips it. The abort carries a sentinel reason so only it — not a later real // error — is reclassified as a retryable IdleTimeoutError. const controller = new AbortController() + const signal = options.signal ? AbortSignal.any([controller.signal, options.signal]) : controller.signal const IDLE_ABORT_REASON = 'jack:idle-timeout' let idleTimer: ReturnType | undefined const clearIdle = () => { @@ -243,11 +245,12 @@ export class PeerConnector extends ServerConnector { let existingBytes = await partFile.exists() ? partFile.size : 0 const doFetch = async (withRange: boolean): Promise => { + signal.throwIfAborted() armIdle() try { return await fetch(url, { headers: withRange ? { ...baseHeaders, Range: `bytes=${existingBytes}-` } : baseHeaders, - signal: controller.signal, + signal, }) } catch (err) { @@ -292,6 +295,7 @@ export class PeerConnector extends ServerConnector { existingBytes = 0 } else if (existingBytes === options.releaseSize) { + signal.throwIfAborted() await rename(partPath, destPath) setSpanAttribute(span, 'download.downloaded_bytes', existingBytes) // Emit headers too so the service persists expectedBytes/source (the @@ -458,6 +462,7 @@ export class PeerConnector extends ServerConnector { throw new IncompleteDownloadError(`Incomplete file download: got ${downloadedBytes} bytes, expected ${expectedBytes}`) reader.releaseLock() + signal.throwIfAborted() await rename(partPath, destPath) setSpanAttribute(span, 'download.downloaded_bytes', downloadedBytes) try { diff --git a/apps/backend/src/management-app.ts b/apps/backend/src/management-app.ts index 9e98b74..baf2213 100644 --- a/apps/backend/src/management-app.ts +++ b/apps/backend/src/management-app.ts @@ -3,6 +3,7 @@ import type { ApiKeysRepository } from './modules/api-keys/api-keys.repository' import type { ConfigService } from './modules/config/config.service' import type { DownloadsRepository } from './modules/downloads/downloads.repository' import type { DownloadsService } from './modules/downloads/downloads.service' +import type { ImportWatcher } from './modules/downloads/import-watcher' import { Hono } from 'hono' import { secureHeaders } from 'hono/secure-headers' import { TmdbClient } from './lib/tmdb/client' @@ -14,6 +15,8 @@ import { CatalogController } from './modules/catalog/catalog.controller' import { getCatalogRouter } from './modules/catalog/catalog.router' import { ConfigController } from './modules/config/config.controller' import { getConfigRouter } from './modules/config/config.router' +import { DownloadsManagementController } from './modules/downloads/downloads.controller' +import { getDownloadsManagementRouter } from './modules/downloads/downloads.router' import { logHub } from './modules/logging/log-store' import { LogsController } from './modules/logging/logs.controller' import { getLogsRouter } from './modules/logging/logs.router' @@ -28,6 +31,7 @@ export function getManagementApp(params: { configService?: ConfigService downloadsRepository?: DownloadsRepository downloadsService?: DownloadsService + importWatcher?: ImportWatcher apiKeysRepository?: ApiKeysRepository tmdbApiKey?: string }) { @@ -47,6 +51,12 @@ export function getManagementApp(params: { const statusController = new StatusController(params.connectors, params.downloadsRepository) app.route('/', getStatusRouter(statusController)) + app.route('/downloads', getDownloadsManagementRouter(new DownloadsManagementController( + params.downloadsRepository, + params.downloadsService, + params.importWatcher, + ))) + // Logs read from the shared LogHub singleton the logger writes to (same module // instance), so this exposes the running process's own logs. app.route('/logs', getLogsRouter(new LogsController(logHub))) diff --git a/apps/backend/src/modules/downloads/download-operation-coordinator.ts b/apps/backend/src/modules/downloads/download-operation-coordinator.ts new file mode 100644 index 0000000..801a7d1 --- /dev/null +++ b/apps/backend/src/modules/downloads/download-operation-coordinator.ts @@ -0,0 +1,55 @@ +import type { DownloadsRepository } from './downloads.repository' + +/** Coordinates per-download import/delete work and reserves paths during deletion. */ +export class DownloadOperationCoordinator { + private readonly tails = new Map>() + private readonly deletingPaths = new Map() + + async runExclusive(id: number, operation: () => Promise): Promise { + const previous = this.tails.get(id) ?? Promise.resolve() + const gate = Promise.withResolvers() + const tail = previous.catch(() => {}).then(() => gate.promise) + this.tails.set(id, tail) + await previous.catch(() => {}) + try { + return await operation() + } + finally { + gate.resolve() + if (this.tails.get(id) === tail) + this.tails.delete(id) + } + } + + async runDelete(id: number, paths: string[], operation: () => Promise): Promise { + for (const path of paths) + this.deletingPaths.set(path, (this.deletingPaths.get(path) ?? 0) + 1) + try { + return await this.runExclusive(id, operation) + } + finally { + for (const path of paths) { + const remaining = (this.deletingPaths.get(path) ?? 1) - 1 + if (remaining === 0) + this.deletingPaths.delete(path) + else + this.deletingPaths.set(path, remaining) + } + } + } + + isPathDeleting(path: string): boolean { + return this.deletingPaths.has(path) + } +} + +const coordinators = new WeakMap() + +export function coordinatorFor(repository: DownloadsRepository): DownloadOperationCoordinator { + let coordinator = coordinators.get(repository) + if (!coordinator) { + coordinator = new DownloadOperationCoordinator() + coordinators.set(repository, coordinator) + } + return coordinator +} diff --git a/apps/backend/src/modules/downloads/downloads.controller.ts b/apps/backend/src/modules/downloads/downloads.controller.ts new file mode 100644 index 0000000..955819c --- /dev/null +++ b/apps/backend/src/modules/downloads/downloads.controller.ts @@ -0,0 +1,44 @@ +import type { DownloadsRepository } from './downloads.repository' +import type { DownloadsService } from './downloads.service' +import type { ImportWatcher } from './import-watcher' +import { ConflictError } from '../../lib/errors/ConflictError' +import { NotFoundError } from '../../lib/errors/NotFoundError' + +export class DownloadsManagementController { + constructor( + private readonly downloadsRepository?: DownloadsRepository, + private readonly downloadsService?: DownloadsService, + private readonly importWatcher?: ImportWatcher, + ) {} + + private requireDownloadsService(): DownloadsService { + if (!this.downloadsService) + throw new ConflictError('Download management is unavailable') + return this.downloadsService + } + + async cancel(id: number) { + return { download: await this.requireDownloadsService().cancel(id) } + } + + async retry(id: number) { + const record = this.downloadsRepository?.get(id) + if (!record) + throw new NotFoundError(`Download ${id} not found`) + if (!record.operationFailed) + throw new ConflictError(`Download ${id} has no failed operation to retry`) + + if (record.lastOperation === 'import') { + if (!this.importWatcher) + throw new ConflictError('Import retry is unavailable') + return { download: await this.importWatcher.retry(id) } + } + + return { download: this.requireDownloadsService().retry(id) } + } + + async delete(id: number) { + await this.requireDownloadsService().delete(id) + return { ok: true } + } +} diff --git a/apps/backend/src/modules/downloads/downloads.repository.ts b/apps/backend/src/modules/downloads/downloads.repository.ts index d976d14..956bf63 100644 --- a/apps/backend/src/modules/downloads/downloads.repository.ts +++ b/apps/backend/src/modules/downloads/downloads.repository.ts @@ -1,5 +1,5 @@ import type { AppDatabase } from '../../database/connection' -import type { DownloadRow, DownloadStatus, ExpectedBytesSource, NewDownloadRow } from '../../database/schema' +import type { DownloadOperation, DownloadRow, DownloadStatus, ExpectedBytesSource, NewDownloadRow } from '../../database/schema' import type { Release } from '../../lib/release' import type { ManualImportTarget } from '../../lib/servers/arr/base' import { desc, eq, sql } from 'drizzle-orm' @@ -26,6 +26,8 @@ export interface DownloadRecord { updatedAt: string completedAt: string | null error: string | null + lastOperation: DownloadOperation + operationFailed: boolean qbCategory: string | null qbSourceServer: string | null sourceServerId: string | null @@ -78,6 +80,8 @@ function toRecord(row: DownloadRow): DownloadRecord { updatedAt: row.updatedAt, completedAt: row.completedAt, error: row.error, + lastOperation: row.lastOperation, + operationFailed: row.operationFailed, qbCategory: row.qbCategory ?? null, qbSourceServer: row.qbSourceServer ?? null, sourceServerId: row.sourceServerId ?? null, @@ -110,6 +114,8 @@ export class DownloadsRepository { manualImportCommandId: input.manualImportCommandId ?? null, downloadedBytes: 0, status: 'downloading', + lastOperation: 'transfer', + operationFailed: false, startedAt: timestamp, updatedAt: timestamp, } @@ -155,6 +161,8 @@ export class DownloadsRepository { completedAt: timestamp, updatedAt: timestamp, error: null, + lastOperation: 'import', + operationFailed: false, }) .where(eq(downloads.id, id)) .run() @@ -164,7 +172,7 @@ export class DownloadsRepository { // is kept (not deleted) so the downloads list doubles as a history. markImported(id: number): void { this.db.update(downloads) - .set({ status: 'imported', updatedAt: nowIso() }) + .set({ status: 'imported', lastOperation: 'import', operationFailed: false, error: null, updatedAt: nowIso() }) .where(eq(downloads.id, id)) .run() } @@ -176,9 +184,23 @@ export class DownloadsRepository { .run() } - markFailed(id: number, error: string): void { + markImportRetryStarted(id: number): void { this.db.update(downloads) - .set({ status: 'failed', error, updatedAt: nowIso() }) + .set({ + status: 'import_queued', + manualImportCommandId: null, + lastOperation: 'import', + operationFailed: false, + error: null, + updatedAt: nowIso(), + }) + .where(eq(downloads.id, id)) + .run() + } + + markFailed(id: number, error: string, operation: DownloadOperation = 'transfer'): void { + this.db.update(downloads) + .set({ status: 'failed', error, lastOperation: operation, operationFailed: true, updatedAt: nowIso() }) .where(eq(downloads.id, id)) .run() } @@ -192,6 +214,13 @@ export class DownloadsRepository { return row?.attempts ?? 0 } + markTransferStarted(id: number): void { + this.db.update(downloads) + .set({ status: 'downloading', lastOperation: 'transfer', operationFailed: false, error: null, updatedAt: nowIso() }) + .where(eq(downloads.id, id)) + .run() + } + markResumeReset(id: number): void { this.db.update(downloads) .set({ downloadedBytes: 0, error: 'resume validation failed; restarted from byte 0', updatedAt: nowIso() }) @@ -238,6 +267,8 @@ export class DownloadsRepository { this.db.update(downloads) .set({ status: 'failed', + lastOperation: 'transfer', + operationFailed: true, downloadedBytes, error: partExists ? `stale download after Jack restart; found .part file with ${downloadedBytes} bytes` diff --git a/apps/backend/src/modules/downloads/downloads.router.ts b/apps/backend/src/modules/downloads/downloads.router.ts new file mode 100644 index 0000000..baccc14 --- /dev/null +++ b/apps/backend/src/modules/downloads/downloads.router.ts @@ -0,0 +1,22 @@ +import type { DownloadsManagementController } from './downloads.controller' +import { Hono } from 'hono' +import { BadRequestError } from '../../lib/errors/BadRequestError' + +const POSITIVE_INTEGER_REGEX = /^[1-9]\d*$/ + +function numericId(raw: string): number { + if (!POSITIVE_INTEGER_REGEX.test(raw)) + throw new BadRequestError('Download id must be a positive integer') + const id = Number(raw) + if (!Number.isSafeInteger(id)) + throw new BadRequestError('Download id must be a safe integer') + return id +} + +export function getDownloadsManagementRouter(controller: DownloadsManagementController) { + const app = new Hono() + app.post('/:id/cancel', async c => c.json(await controller.cancel(numericId(c.req.param('id'))))) + app.post('/:id/retry', async c => c.json(await controller.retry(numericId(c.req.param('id'))))) + app.delete('/:id', async c => c.json(await controller.delete(numericId(c.req.param('id'))))) + return app +} diff --git a/apps/backend/src/modules/downloads/downloads.service.ts b/apps/backend/src/modules/downloads/downloads.service.ts index 3f1dd0f..2b2a9da 100644 --- a/apps/backend/src/modules/downloads/downloads.service.ts +++ b/apps/backend/src/modules/downloads/downloads.service.ts @@ -2,11 +2,16 @@ import type { AppConfig } from '../../lib/config' import type { ConnectorManager } from '../../lib/servers' import type { ManualImportTarget } from '../../lib/servers/arr/base' import type { PeerDownloadProgressEvent } from '../../lib/servers/peer' +import type { DownloadOperationCoordinator } from './download-operation-coordinator' import type { DownloadRecord, DownloadsRepository } from './downloads.repository' -import { basename, join } from 'node:path' +import { unlink } from 'node:fs/promises' +import { basename, isAbsolute, join, relative, resolve } from 'node:path' +import { ConflictError } from '../../lib/errors/ConflictError' +import { NotFoundError } from '../../lib/errors/NotFoundError' import { retry } from '../../lib/retry' import { Semaphore } from '../../lib/semaphore' import { logger } from '../../logger' +import { coordinatorFor } from './download-operation-coordinator' import { downloadRetryAfterMs, isTransientDownloadError } from './retry-policy' type DownloadsServiceConfig = NonNullable @@ -34,6 +39,8 @@ export class DownloadsService { // Dest paths with a download in flight — guards two concurrent live drops that // resolve to the same destination (no duplicate rows / writers). private readonly active = new Set() + private readonly transfers = new Map }>() + private readonly coordinator?: DownloadOperationCoordinator constructor( private readonly config: DownloadsServiceConfig, @@ -41,14 +48,22 @@ export class DownloadsService { // ConnectorManager (live) or a test stub both satisfy it. private readonly connectorManager: { peers: ConnectorManager['peers'] }, private readonly downloadsRepository?: DownloadsRepository, + coordinator?: DownloadOperationCoordinator, ) { this.semaphore = new Semaphore(config.maxConcurrentDownloads) + this.coordinator = coordinator ?? (downloadsRepository ? coordinatorFor(downloadsRepository) : undefined) } private get peers() { return this.connectorManager.peers } + private requireRepository(): DownloadsRepository { + if (!this.downloadsRepository) + throw new ConflictError('Download management is unavailable') + return this.downloadsRepository + } + /** * Shared creation core for the qB add path. Returns the created record, a * benign duplicate (a download for the same destination is already active), @@ -86,7 +101,7 @@ export class DownloadsService { const destPath = join(this.config.completedPath, safeName) const partPath = `${destPath}.part` - if (this.active.has(destPath)) { + if (this.active.has(destPath) || this.coordinator?.isPathDeleting(resolve(destPath))) { logger.debug({ torrentFilename, destPath }, 'A download for this destination is already active; skipping duplicate') return { kind: 'duplicate' } } @@ -131,6 +146,8 @@ export class DownloadsService { updatedAt: '', completedAt: null, error: null, + lastOperation: 'transfer', + operationFailed: false, qbCategory: input.qbCategory ?? null, qbSourceServer: input.qbSourceServer ?? null, sourceServerId: input.sourceServerId ?? null, @@ -169,7 +186,7 @@ export class DownloadsService { // A duplicate is already in flight: no new row, but a success — don't make *arr retry. if (outcome.kind === 'duplicate') return 'duplicate' - void this.runDownload(outcome.record).catch((err) => { + void this.enqueue(outcome.record).catch((err) => { const message = err instanceof Error ? err.message : String(err) logger.error({ itemId: input.itemId, error: message }, 'qB download failed') }) @@ -210,7 +227,7 @@ export class DownloadsService { return 'failed' if (outcome.kind === 'duplicate') return 'duplicate' - void this.runDownload(outcome.record).catch((err) => { + void this.enqueue(outcome.record).catch((err) => { const message = err instanceof Error ? err.message : String(err) logger.error({ itemId: input.itemId, error: message }, 'Direct download failed') }) @@ -238,7 +255,7 @@ export class DownloadsService { } for (const record of resumable) { // Fire-and-forget: the semaphore caps concurrency. - void this.runDownload(record).catch((err) => { + void this.enqueue(record).catch((err) => { const message = err instanceof Error ? err.message : String(err) logger.error({ torrentFilename: record.torrentFilename, error: message }, 'Failed to resume stale download') }) @@ -248,19 +265,108 @@ export class DownloadsService { return resumable.length } - private async runDownload(record: DownloadRecord): Promise { + private enqueue(record: DownloadRecord): Promise { + const controller = new AbortController() + const task = this.runDownload(record, controller) + this.transfers.set(record.id, { controller, task }) + void task.finally(() => { + if (this.transfers.get(record.id)?.task === task) + this.transfers.delete(record.id) + }).catch(() => {}) + return task + } + + async cancel(id: number): Promise { + const repo = this.requireRepository() + const record = repo.get(id) + if (!record) + throw new NotFoundError(`Download ${id} not found`) + if (record.status === 'failed' && record.operationFailed && record.lastOperation === 'transfer' && record.error?.includes('cancelled')) + return record + if (record.status !== 'downloading') + throw new ConflictError(`Download ${id} is not active`) + const transfer = this.transfers.get(id) + if (!transfer) + throw new ConflictError(`Download ${id} is not active in this process`) + transfer.controller.abort(new Error('Download cancelled by user')) + await transfer.task + return repo.get(id) ?? record + } + + retry(id: number): DownloadRecord { + const repo = this.requireRepository() + const record = repo.get(id) + if (!record) + throw new NotFoundError(`Download ${id} not found`) + if (!record.operationFailed || record.lastOperation !== 'transfer') + throw new ConflictError(`Download ${id} has no failed transfer to retry`) + if (this.transfers.has(id) || this.active.has(record.destPath)) + throw new ConflictError(`Download ${id} is already active`) + repo.markTransferStarted(id) + void this.enqueue(record).catch((err) => { + const message = err instanceof Error ? err.message : String(err) + logger.error({ id, error: message }, 'Retried download failed') + }) + return repo.get(id) ?? record + } + + async delete(id: number): Promise { + const repo = this.requireRepository() + const record = repo.get(id) + if (!record) + throw new NotFoundError(`Download ${id} not found`) + const paths = [resolve(record.partPath), resolve(record.destPath)] + const operation = async () => { + const current = repo.get(id) + if (!current) + throw new NotFoundError(`Download ${id} not found`) + if (current.status === 'downloading') + await this.cancel(id) + + const root = resolve(this.config.completedPath) + for (const artifact of [record.partPath, record.destPath]) { + const resolved = resolve(artifact) + const fromRoot = relative(root, resolved) + const isSafe = fromRoot !== '' && !fromRoot.startsWith('..') && !isAbsolute(fromRoot) + // Re-read ownership immediately before each unlink. A sibling that began + // before the path reservation was installed must keep the shared file. + const live = repo.get(id) + const stillOwned = live != null && [resolve(live.partPath), resolve(live.destPath)].includes(resolved) + const siblingOwns = repo.list().some(row => row.id !== id + && [resolve(row.destPath), resolve(row.partPath)].includes(resolved)) + if (isSafe && stillOwned && !siblingOwns) { + await unlink(resolved).catch((err: unknown) => { + if (err instanceof Error && 'code' in err && err.code === 'ENOENT') + return + throw err + }) + } + } + repo.delete(id) + } + if (this.coordinator) + await this.coordinator.runDelete(id, paths, operation) + else + await operation() + } + + private async runDownload(record: DownloadRecord, controller: AbortController): Promise { if (this.active.has(record.destPath)) return this.active.add(record.destPath) try { - await this.semaphore.run(() => this.downloadWithRetry(record)) + await this.semaphore.run(() => this.downloadWithRetry(record, controller.signal), controller.signal) + } + catch (err) { + const message = err instanceof Error ? err.message : String(err) + this.downloadsRepository?.markFailed(record.id, message, 'transfer') } finally { this.active.delete(record.destPath) } } - private async downloadWithRetry(record: DownloadRecord): Promise { + private async downloadWithRetry(record: DownloadRecord, signal: AbortSignal): Promise { const repo = this.downloadsRepository const peer = this.peers.find(p => p.id === record.peerId) if (!peer) { @@ -289,6 +395,7 @@ export class DownloadsService { try { await retry(async () => { + signal.throwIfAborted() repo?.incrementAttempts(record.id) await peer.downloadFile(record.itemId, record.destPath, { torrentFilename: record.torrentFilename, @@ -296,6 +403,7 @@ export class DownloadsService { releaseSize: record.releaseSize, idleTimeoutMs: this.config.idleTimeoutMs, onProgress, + signal, }) }, { maxAttempts: this.config.maxDownloadAttempts, @@ -307,8 +415,10 @@ export class DownloadsService { const message = error instanceof Error ? error.message : String(error) logger.warn({ torrentFilename: record.torrentFilename, attempt, delayMs, error: message }, 'Retrying peer download after transient failure') }, + signal, }) + signal.throwIfAborted() repo?.markImportQueued(record.id) logger.info({ torrentFilename: record.torrentFilename, filename: record.filename }, 'Download complete') } diff --git a/apps/backend/src/modules/downloads/import-watcher.ts b/apps/backend/src/modules/downloads/import-watcher.ts index 4d34794..61079a7 100644 --- a/apps/backend/src/modules/downloads/import-watcher.ts +++ b/apps/backend/src/modules/downloads/import-watcher.ts @@ -1,9 +1,13 @@ import type { ArrServerConnector } from '../../lib/servers/arr/base' +import type { DownloadOperationCoordinator } from './download-operation-coordinator' import type { DownloadRecord, DownloadsRepository } from './downloads.repository' import { dirname } from 'node:path' +import { ConflictError } from '../../lib/errors/ConflictError' +import { NotFoundError } from '../../lib/errors/NotFoundError' import { PermanentManualImportError } from '../../lib/servers/arr/base' import { logger } from '../../logger' import { deriveHash } from '../qbittorrent/qbittorrent.mapper' +import { coordinatorFor } from './download-operation-coordinator' /** * Periodically reconciles `import_queued` downloads against the history of each @@ -33,6 +37,8 @@ export class ImportWatcher { // *arr 500s because the movie folder is missing) backs off instead of re-firing // every tick. In-memory by design: a restart is a fair signal to try again. private readonly triggerFailures = new Map() + private readonly coordinator: DownloadOperationCoordinator + private tickInFlight?: Promise constructor( private readonly repository: DownloadsRepository, @@ -41,7 +47,10 @@ export class ImportWatcher { private readonly connectorManager: { servers: ArrServerConnector[] }, private readonly intervalMs: number, private readonly retryPolicy: ManualImportRetryPolicy = DEFAULT_RETRY_POLICY, - ) {} + coordinator?: DownloadOperationCoordinator, + ) { + this.coordinator = coordinator ?? coordinatorFor(repository) + } private connectorFor(row: DownloadRecord): ArrServerConnector | undefined { if (row.sourceServerId) { @@ -72,8 +81,54 @@ export class ImportWatcher { this.timer = undefined } + async retry(id: number): Promise { + return this.coordinator.runExclusive(id, async () => { + const row = this.repository.get(id) + if (!row) + throw new NotFoundError(`Download ${id} not found`) + if (!row.operationFailed || row.lastOperation !== 'import') + throw new ConflictError(`Download ${id} has no failed import to retry`) + if (row.importMode !== 'jack_manual' || !row.importTarget) + throw new ConflictError(`Download ${id} does not support a manual import retry`) + const connector = this.connectorFor(row) + if (!connector?.isInitialized) + throw new ConflictError(`Destination for download ${id} is unavailable`) + + this.repository.markImportRetryStarted(id) + try { + const commandId = await connector.manualImport({ + folder: dirname(row.destPath), + paths: [row.destPath], + target: row.importTarget, + downloadId: deriveHash(row.release.title, row.releaseSize), + release: row.release, + }) + this.repository.setManualImportCommand(id, commandId) + this.triggerFailures.delete(id) + return this.repository.get(id) ?? row + } + catch (err) { + const message = err instanceof Error ? err.message : String(err) + this.repository.markFailed(id, message, 'import') + throw err + } + }) + } + /** One reconciliation pass. Returns how many downloads it marked imported. */ - async tick(): Promise { + tick(): Promise { + if (this.tickInFlight) + return this.tickInFlight + const tick = this.tickOnce() + this.tickInFlight = tick + void tick.finally(() => { + if (this.tickInFlight === tick) + this.tickInFlight = undefined + }).catch(() => {}) + return tick + } + + private async tickOnce(): Promise { // Only rows with an owning destination can be reconciled; blackhole rows have // no server to poll, so they're left as-is. const queued = this.repository.listByStatus('import_queued').filter(r => r.sourceServerId || r.qbSourceServer) @@ -91,95 +146,98 @@ export class ImportWatcher { const skippedConnectors = new Set() let importedCount = 0 - for (const row of queued) { - const connector = this.connectorFor(row) - if (!connector || !connector.isInitialized) - continue - - if (!importedIdsByConnector.has(connector.id) && !skippedConnectors.has(connector.id)) { - try { - importedIdsByConnector.set(connector.id, await connector.recentlyImportedDownloadIds()) + for (const queuedRow of queued) { + importedCount += await this.coordinator.runExclusive(queuedRow.id, async () => { + // Another tick/retry/delete may have changed or removed the row while this + // operation waited for the per-ID lock. Never act on the stale snapshot. + const row = this.repository.get(queuedRow.id) + if (!row || row.status !== 'import_queued') + return 0 + const connector = this.connectorFor(row) + if (!connector || !connector.isInitialized) + return 0 + + if (!importedIdsByConnector.has(connector.id) && !skippedConnectors.has(connector.id)) { + try { + importedIdsByConnector.set(connector.id, await connector.recentlyImportedDownloadIds()) + } + catch (err) { + const message = err instanceof Error ? err.message : String(err) + skippedConnectors.add(connector.id) + logger.warn({ server: connector.name, error: message }, 'Could not read import history; will retry next tick') + } } - catch (err) { - const message = err instanceof Error ? err.message : String(err) - skippedConnectors.add(connector.id) - logger.warn({ server: connector.name, error: message }, 'Could not read import history; will retry next tick') + const importedIds = importedIdsByConnector.get(connector.id) + if (!importedIds) + return 0 + + const hash = deriveHash(row.release.title, row.releaseSize).toLowerCase() + if (importedIds.has(hash)) { + this.repository.markImported(row.id) + logger.info({ id: row.id, filename: row.filename, server: connector.name }, 'Download imported by *arr') + return 1 } - } - const importedIds = importedIdsByConnector.get(connector.id) - if (!importedIds) - continue - - const hash = deriveHash(row.release.title, row.releaseSize).toLowerCase() - if (importedIds.has(hash)) { - this.repository.markImported(row.id) - importedCount++ - logger.info({ id: row.id, filename: row.filename, server: connector.name }, 'Download imported by *arr') - continue - } - - if (row.importMode !== 'jack_manual' || !row.importTarget) - continue - if (row.manualImportCommandId != null) { - try { - const status = await connector.manualImportCommandStatus(row.manualImportCommandId) - if (status.state === 'completed') { - this.repository.markImported(row.id) - importedCount++ - logger.info({ id: row.id, filename: row.filename, server: connector.name, commandId: row.manualImportCommandId }, 'Manual import command completed') + if (row.importMode !== 'jack_manual' || !row.importTarget) + return 0 + + if (row.manualImportCommandId != null) { + try { + const status = await connector.manualImportCommandStatus(row.manualImportCommandId) + if (status.state === 'completed') { + this.repository.markImported(row.id) + logger.info({ id: row.id, filename: row.filename, server: connector.name, commandId: row.manualImportCommandId }, 'Manual import command completed') + return 1 + } + if (status.state === 'failed') { + this.repository.markFailed(row.id, status.error, 'import') + logger.warn({ id: row.id, server: connector.name, commandId: row.manualImportCommandId, error: status.error }, 'Manual import command failed') + } } - else if (status.state === 'failed') { - this.repository.markFailed(row.id, status.error) - logger.warn({ id: row.id, server: connector.name, commandId: row.manualImportCommandId, error: status.error }, 'Manual import command failed') + catch (err) { + const message = err instanceof Error ? err.message : String(err) + logger.warn({ id: row.id, server: connector.name, commandId: row.manualImportCommandId, error: message }, 'Could not read manual import command status; will retry next tick') } + return 0 } - catch (err) { - const message = err instanceof Error ? err.message : String(err) - logger.warn({ id: row.id, server: connector.name, commandId: row.manualImportCommandId, error: message }, 'Could not read manual import command status; will retry next tick') - } - continue - } - // Still inside a back-off window from a recent trigger failure: skip so a - // persistently failing import (e.g. *arr 500s on a missing folder) doesn't - // re-fire every tick. The item/history checks above still ran this tick, so - // we'll notice the moment *arr actually gains the file. - const backoff = this.triggerFailures.get(row.id) - if (backoff && now < backoff.nextAttemptAt) - continue + const backoff = this.triggerFailures.get(row.id) + if (backoff && now < backoff.nextAttemptAt) + return 0 - try { - const commandId = await connector.manualImport({ - folder: dirname(row.destPath), - paths: [row.destPath], - target: row.importTarget, - downloadId: deriveHash(row.release.title, row.releaseSize), - release: row.release, - }) - this.repository.setManualImportCommand(row.id, commandId) - this.triggerFailures.delete(row.id) - logger.info({ id: row.id, filename: row.filename, server: connector.name, commandId }, 'Triggered *arr manual import') - } - catch (err) { - const message = err instanceof Error ? err.message : String(err) - if (err instanceof PermanentManualImportError) { - this.repository.markFailed(row.id, message) + try { + const commandId = await connector.manualImport({ + folder: dirname(row.destPath), + paths: [row.destPath], + target: row.importTarget, + downloadId: deriveHash(row.release.title, row.releaseSize), + release: row.release, + }) + this.repository.setManualImportCommand(row.id, commandId) this.triggerFailures.delete(row.id) - logger.warn({ id: row.id, server: connector.name, error: message }, 'Manual import cannot be retried') - continue + logger.info({ id: row.id, filename: row.filename, server: connector.name, commandId }, 'Triggered *arr manual import') } - const failures = (this.triggerFailures.get(row.id)?.failures ?? 0) + 1 - if (failures >= this.retryPolicy.maxAttempts) { - this.repository.markFailed(row.id, `manual import failed after ${failures} attempts: ${message}`) - this.triggerFailures.delete(row.id) - logger.warn({ id: row.id, server: connector.name, attempts: failures, error: message }, 'Manual import gave up after repeated failures') - continue + catch (err) { + const message = err instanceof Error ? err.message : String(err) + if (err instanceof PermanentManualImportError) { + this.repository.markFailed(row.id, message, 'import') + this.triggerFailures.delete(row.id) + logger.warn({ id: row.id, server: connector.name, error: message }, 'Manual import cannot be retried') + return 0 + } + const failures = (this.triggerFailures.get(row.id)?.failures ?? 0) + 1 + if (failures >= this.retryPolicy.maxAttempts) { + this.repository.markFailed(row.id, `manual import failed after ${failures} attempts: ${message}`, 'import') + this.triggerFailures.delete(row.id) + logger.warn({ id: row.id, server: connector.name, attempts: failures, error: message }, 'Manual import gave up after repeated failures') + return 0 + } + const delayMs = Math.min(this.retryPolicy.backoffMaxMs, this.retryPolicy.backoffBaseMs * 2 ** (failures - 1)) + this.triggerFailures.set(row.id, { failures, nextAttemptAt: now + delayMs }) + logger.warn({ id: row.id, server: connector.name, attempts: failures, retryInMs: delayMs, error: message }, 'Manual import trigger failed; backing off') } - const delayMs = Math.min(this.retryPolicy.backoffMaxMs, this.retryPolicy.backoffBaseMs * 2 ** (failures - 1)) - this.triggerFailures.set(row.id, { failures, nextAttemptAt: now + delayMs }) - logger.warn({ id: row.id, server: connector.name, attempts: failures, retryInMs: delayMs, error: message }, 'Manual import trigger failed; backing off') - } + return 0 + }) } return importedCount } diff --git a/apps/ui/app/components/DownloadActions.vue b/apps/ui/app/components/DownloadActions.vue new file mode 100644 index 0000000..3bfcf8d --- /dev/null +++ b/apps/ui/app/components/DownloadActions.vue @@ -0,0 +1,41 @@ + + + diff --git a/apps/ui/app/pages/downloads.vue b/apps/ui/app/pages/downloads.vue index 1172bed..4ab1ec7 100644 --- a/apps/ui/app/pages/downloads.vue +++ b/apps/ui/app/pages/downloads.vue @@ -1,16 +1,60 @@ @@ -162,6 +207,14 @@ const columns: TableColumn[] = [ + +