diff --git a/src/main/disk-space.ts b/src/main/disk-space.ts new file mode 100644 index 0000000..cbf9e51 --- /dev/null +++ b/src/main/disk-space.ts @@ -0,0 +1,177 @@ +import fs from "node:fs"; +import path from "node:path"; + +export type DiskReservationPhase = "download" | "extract" | "remux"; + +export type DiskVolumeStats = { + path: string; + volumeKey: string; + freeBytes: number; + totalBytes: number; +}; + +export type DiskReservationRequest = { + phase: DiskReservationPhase; + ownerId: string; + targetPath: string; + requiredBytes: number | null; + alreadyPresentBytes?: number; +}; + +export type DiskWaitEvent = { + phase: DiskReservationPhase; + ownerId: string; + volumeKey: string; + requiredBytes: number; + availableBytes: number; + deficitBytes: number; + safetyBytes: number; + retryAt: number; +}; + +export class DiskCapacityError extends Error { + public readonly event: DiskWaitEvent; + + public constructor(event: DiskWaitEvent) { + super("Insufficient disk capacity"); + this.name = "DiskCapacityError"; + this.event = event; + } +} + +type DiskReservationCoordinatorOptions = { + safetyBytes?: number; + retryDelayMs?: number; + now?: () => number; + statVolume?: (targetPath: string) => Promise; +}; + +type DiskReservationUpdate = Pick; + +export type DiskReservationLease = { + readonly volumeKey: string | null; + readonly released: boolean; + readonly reservedBytes: number; + update(update: DiskReservationUpdate): Promise; + release(): void; +}; + +export function calculateRemainingReservationBytes(requiredBytes: number | null, alreadyPresentBytes = 0): number | null { + if (!Number.isFinite(requiredBytes) || requiredBytes === null || requiredBytes < 0) return null; + return Math.max(0, Math.floor(requiredBytes) - Math.max(0, Math.floor(alreadyPresentBytes))); +} + +export function calculateExtractionReservationBytes(archiveBytes: Array): number | null { + const known = archiveBytes.filter((value): value is number => typeof value === "number" && Number.isFinite(value) && value > 0); + if (known.length === 0) return null; + return known.reduce((total, value) => total + Math.floor(value), 0); +} + +async function defaultStatVolume(targetPath: string): Promise { + let candidate = path.resolve(targetPath); + while (true) { + try { + const stat = await fs.promises.stat(candidate); + const directory = stat.isDirectory() ? candidate : path.dirname(candidate); + const info = await fs.promises.statfs(directory); + const freeBytes = Number(info.bavail) * Number(info.bsize); + const totalBytes = Number(info.blocks) * Number(info.bsize); + return { path: directory, volumeKey: path.parse(directory).root.toLowerCase(), freeBytes, totalBytes }; + } catch { + const parent = path.dirname(candidate); + if (parent === candidate) throw new Error(`Unable to resolve disk volume for ${targetPath}`); + candidate = parent; + } + } +} + +export class DiskReservationCoordinator { + private readonly safetyBytes: number; + private readonly retryDelayMs: number; + private readonly now: () => number; + private readonly statVolume: (targetPath: string) => Promise; + private readonly reservedByVolume = new Map(); + private readonly leases = new Map(); + private queue = Promise.resolve(); + + public constructor(options: DiskReservationCoordinatorOptions = {}) { + this.safetyBytes = Math.max(0, Math.floor(options.safetyBytes ?? 256 * 1024 * 1024)); + this.retryDelayMs = Math.max(1000, Math.floor(options.retryDelayMs ?? 30000)); + this.now = options.now ?? Date.now; + this.statVolume = options.statVolume ?? defaultStatVolume; + } + + public async reserve(request: DiskReservationRequest): Promise { + return this.enqueue(async () => { + const requiredBytes = calculateRemainingReservationBytes(request.requiredBytes, request.alreadyPresentBytes ?? 0); + if (requiredBytes === null) return this.createLease(request.ownerId, request.targetPath, null, 0); + const volume = await this.statVolume(request.targetPath); + const reserved = this.reservedByVolume.get(volume.volumeKey) ?? 0; + const availableBytes = Math.max(0, Math.floor(volume.freeBytes) - reserved - this.safetyBytes); + if (requiredBytes > availableBytes) { + throw new DiskCapacityError({ + phase: request.phase, + ownerId: request.ownerId, + volumeKey: volume.volumeKey, + requiredBytes, + availableBytes, + deficitBytes: requiredBytes - availableBytes, + safetyBytes: this.safetyBytes, + retryAt: this.now() + this.retryDelayMs + }); + } + return this.createLease(request.ownerId, request.targetPath, volume.volumeKey, requiredBytes); + }); + } + + public getReservedBytesByVolume(): ReadonlyMap { + return new Map(this.reservedByVolume); + } + + private createLease(ownerId: string, targetPath: string, volumeKey: string | null, reservedBytes: number): DiskReservationLease { + const leaseId = `${ownerId}:${Math.random().toString(16).slice(2)}`; + const coordinator = this; + if (volumeKey) { + this.leases.set(leaseId, { volumeKey, reservedBytes, targetPath }); + this.reservedByVolume.set(volumeKey, (this.reservedByVolume.get(volumeKey) ?? 0) + reservedBytes); + } + let released = false; + return { + get volumeKey() { return volumeKey; }, + get released() { return released; }, + get reservedBytes() { return coordinator.leases.get(leaseId)?.reservedBytes ?? 0; }, + update: async (update) => { + await coordinator.enqueue(async () => { + if (released || !volumeKey) return; + const lease = coordinator.leases.get(leaseId); + if (!lease) return; + const nextBytes = calculateRemainingReservationBytes(update.requiredBytes, update.alreadyPresentBytes ?? 0); + if (nextBytes === null) return; + const delta = nextBytes - lease.reservedBytes; + if (delta > 0) { + const volume = await coordinator.statVolume(lease.targetPath); + const available = Math.max(0, Math.floor(volume.freeBytes) - (coordinator.reservedByVolume.get(volumeKey) ?? 0) - coordinator.safetyBytes); + if (delta > available) throw new DiskCapacityError({ phase: "download", ownerId, volumeKey, requiredBytes: nextBytes, availableBytes: available, deficitBytes: delta - available, safetyBytes: coordinator.safetyBytes, retryAt: coordinator.now() + coordinator.retryDelayMs }); + } + lease.reservedBytes = nextBytes; + coordinator.reservedByVolume.set(volumeKey, Math.max(0, (coordinator.reservedByVolume.get(volumeKey) ?? 0) + delta)); + }); + }, + release: () => { + if (released) return; + released = true; + const lease = coordinator.leases.get(leaseId); + coordinator.leases.delete(leaseId); + if (!lease) return; + const next = Math.max(0, (coordinator.reservedByVolume.get(lease.volumeKey) ?? 0) - lease.reservedBytes); + coordinator.reservedByVolume.set(lease.volumeKey, next); + } + }; + } + + private async enqueue(operation: () => Promise): Promise { + const next = this.queue.then(operation, operation); + this.queue = next.then(() => undefined, () => undefined); + return next; + } +} diff --git a/src/main/download-manager.ts b/src/main/download-manager.ts index 0582dea..4b2b755 100644 --- a/src/main/download-manager.ts +++ b/src/main/download-manager.ts @@ -71,6 +71,7 @@ import { logDesktopRename, verifyRename, verifyRenameAsync, type RenameVerificat import { StoragePaths, saveSession, saveSessionAsync, saveSettings, saveSettingsAsync } from "./storage"; import { compactErrorText, ensureDirPath, filenameFromUrl, formatEta, humanSize, looksLikeOpaqueFilename, nowMs, sanitizeFilename, sleep } from "./utils"; import { mergeKnownTotalBytes } from "./download-size"; +import { DiskCapacityError, DiskReservationCoordinator, type DiskReservationLease } from "./disk-space"; import { createRendererState } from "./renderer-state"; type ActiveTask = { @@ -1857,7 +1858,15 @@ export class DownloadManager extends EventEmitter { private lastGlobalProgressAt = 0; - private retryAfterByItem = new Map(); + private retryAfterByItem = new Map(); + + private packageDiskRetryAfterByPackage = new Map(); + + private diskWaitEvents: NonNullable = []; + + private diskReservations = new DiskReservationCoordinator(); + + private diskLeasesByOwner = new Map(); private retryStateByItem = new Map { const out: Record = {}; @@ -4192,19 +4202,46 @@ export class DownloadManager extends EventEmitter { let processed = 0; let failed = 0; for (const sourcePath of targets) { - if (shouldAbort?.() || signal?.aborted) { - return processed; - } - const sourceName = path.basename(sourcePath); - let result: VideoProcessResult; - try { - result = await processVideoFile(sourcePath, { mode, cpuPriority: this.settings.extractCpuPriority, signal }); - } catch (error) { - result = { action: "error", reason: "exception", error: compactErrorText(error) }; - } - if (result.action === "aborted") { - return processed; - } + if (shouldAbort?.() || signal?.aborted) { + return processed; + } + const sourceName = path.basename(sourcePath); + let result: VideoProcessResult | null = null; + let remuxLease: DiskReservationLease | null = null; + try { + try { + const sourceBytes = (await fs.promises.stat(sourcePath)).size; + remuxLease = await this.diskReservations.reserve({ + phase: "remux", + ownerId: pkg?.id || sourcePath, + targetPath: sourcePath, + requiredBytes: sourceBytes + }); + } catch (error) { + if (error instanceof DiskCapacityError) { + this.diskWaitEvents = [{ + ...error.event, + ...(pkg ? { packageId: pkg.id } : {}) + }]; + result = { action: "skipped-no-space", reason: "zu wenig freier Speicher fuer Remux" }; + } else { + throw error; + } + } + if (!result) { + result = await processVideoFile(sourcePath, { mode, cpuPriority: this.settings.extractCpuPriority, signal }); + } + } catch (error) { + result = { action: "error", reason: "exception", error: compactErrorText(error) }; + } finally { + remuxLease?.release(); + } + if (!result) { + result = { action: "error", reason: "exception", error: "Unbekannter Remux-Status" }; + } + if (result.action === "aborted") { + return processed; + } const langs = (result.audioLanguages || []).join(","); if (pkg) { const level = result.action === "error" ? "WARN" : "INFO"; @@ -9027,8 +9064,10 @@ export class DownloadManager extends EventEmitter { void this.processItem(active).catch((err) => { logger.warn(`processItem unbehandelt (${itemId}): ${compactErrorText(err)}`); - }).finally(() => { - if (!this.retryAfterByItem.has(item.id)) { + }).finally(() => { + this.diskLeasesByOwner.get(itemId)?.release(); + this.diskLeasesByOwner.delete(itemId); + if (!this.retryAfterByItem.has(item.id)) { this.releaseTargetPath(item.id); } if (active.nonResumableCounted) { @@ -9206,8 +9245,33 @@ export class DownloadManager extends EventEmitter { const preferredTargetPath = canReuseExistingTarget ? existingTargetPath : path.join(pkg.outputDir, item.fileName); - item.targetPath = this.claimTargetPath(item.id, preferredTargetPath, Boolean(canReuseExistingTarget)); + item.targetPath = this.claimTargetPath(item.id, preferredTargetPath, Boolean(canReuseExistingTarget)); item.totalBytes = mergeKnownTotalBytes(item.totalBytes, unrestricted.fileSize); + try { + const diskLease = await this.diskReservations.reserve({ + phase: "download", + ownerId: item.id, + targetPath: item.targetPath, + requiredBytes: item.totalBytes, + alreadyPresentBytes: item.downloadedBytes + }); + this.diskLeasesByOwner.get(item.id)?.release(); + this.diskLeasesByOwner.set(item.id, diskLease); + } catch (error) { + if (error instanceof DiskCapacityError) { + this.diskWaitEvents = [{ + ...error.event, + itemId: item.id, + packageId: pkg.id + }]; + this.releaseTargetPath(item.id); + this.queueRetry(item, active, Math.max(1000, error.event.retryAt - nowMs()), "Warte auf Festplatte"); + this.persistSoon(); + this.emitState(); + return; + } + throw error; + } item.status = "downloading"; const pLabel = unrestricted.providerLabel; item.fullStatus = "Starte..."; @@ -9345,10 +9409,12 @@ export class DownloadManager extends EventEmitter { : `Fertig (${humanSize(item.downloadedBytes)})`; item.progressPercent = 100; item.speedBps = 0; - item.updatedAt = completedAt; - this.notePackageDownloadCompleted(pkg, completedAt); - pkg.updatedAt = completedAt; - this.recordRunOutcome(item.id, "completed"); + item.updatedAt = completedAt; + this.notePackageDownloadCompleted(pkg, completedAt); + pkg.updatedAt = completedAt; + this.diskLeasesByOwner.get(item.id)?.release(); + this.diskLeasesByOwner.delete(item.id); + this.recordRunOutcome(item.id, "completed"); logger.info(`Download fertig: ${item.fileName} (${humanSize(item.downloadedBytes)}), pkg=${pkg.name}`); this.logPackageForItem(item, "INFO", "Download abgeschlossen", { downloadedBytes: item.downloadedBytes, @@ -12169,13 +12235,47 @@ export class DownloadManager extends EventEmitter { const fullArchiveSet = await this.findFullExtractArchiveSet(pkg, completedItems); const fullExtractItemIds = new Set(); - for (const archivePath of fullArchiveSet) { - const archiveItems = resolveArchiveItems(path.basename(archivePath)); - for (const entry of archiveItems) { - fullExtractItemIds.add(entry.id); - } - } - const pendingAt = nowMs(); + for (const archivePath of fullArchiveSet) { + const archiveItems = resolveArchiveItems(path.basename(archivePath)); + for (const entry of archiveItems) { + fullExtractItemIds.add(entry.id); + } + } + const archiveSizes = await Promise.all([...fullArchiveSet].map(async (archivePath) => { + try { + return (await fs.promises.stat(archivePath)).size; + } catch { + return null; + } + })); + try { + const diskLease = await this.diskReservations.reserve({ + phase: "extract", + ownerId: packageId, + targetPath: pkg.extractDir, + requiredBytes: archiveSizes.every((size) => size === null) ? null : archiveSizes.reduce((total, size) => total + Math.max(0, size || 0), 0) + }); + diskLease.release(); + } catch (error) { + if (error instanceof DiskCapacityError) { + this.diskWaitEvents = [{ ...error.event, packageId }]; + const retryAt = error.event.retryAt; + this.packageDiskRetryAfterByPackage.set(packageId, retryAt); + for (const entry of completedItems) { + entry.fullStatus = "Warte auf Festplatte"; + entry.lastError = "Zu wenig Speicherplatz"; + entry.updatedAt = nowMs(); + } + pkg.postProcessLabel = undefined; + pkg.status = "queued"; + pkg.updatedAt = nowMs(); + this.persistSoon(); + this.emitState(); + return; + } + throw error; + } + const pendingAt = nowMs(); for (const entry of completedItems) { if (!fullExtractItemIds.has(entry.id) || isExtractedLabel(entry.fullStatus)) { continue; diff --git a/src/shared/types.ts b/src/shared/types.ts index aba5ea9..3734ec2 100644 --- a/src/shared/types.ts +++ b/src/shared/types.ts @@ -469,8 +469,19 @@ export interface UiSnapshot { canPause: boolean; clipboardActive: boolean; reconnectSeconds: number; - packageSpeedBps: Record; - payloadKind?: "full" | "delta"; + packageSpeedBps: Record; + diskWaitEvents?: Array<{ + phase: "download" | "extract" | "remux"; + ownerId: string; + itemId?: string; + packageId?: string; + volumeKey: string; + requiredBytes: number; + availableBytes: number; + deficitBytes: number; + retryAt: number; + }>; + payloadKind?: "full" | "delta"; removedItemIds?: string[]; removedPackageIds?: string[]; rotationEvents?: RotationEvent[]; diff --git a/tests/disk-space.test.ts b/tests/disk-space.test.ts new file mode 100644 index 0000000..9ca6a33 --- /dev/null +++ b/tests/disk-space.test.ts @@ -0,0 +1,140 @@ +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + DiskReservationCoordinator, + calculateExtractionReservationBytes, + calculateRemainingReservationBytes +} from "../src/main/disk-space"; + +describe("disk reservation coordinator", () => { + it("reserves only the remaining known bytes and keeps the configured safety margin free", async () => { + const coordinator = new DiskReservationCoordinator({ + safetyBytes: 100, + now: () => 10, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "volume-a", + freeBytes: 1_000, + totalBytes: 2_000 + }) + }); + + const first = await coordinator.reserve({ + phase: "download", + ownerId: "item-a", + targetPath: path.join("C:\\", "downloads", "a.bin"), + requiredBytes: 800, + alreadyPresentBytes: 300 + }); + + expect(first.reservedBytes).toBe(500); + await expect(coordinator.reserve({ + phase: "download", + ownerId: "item-b", + targetPath: path.join("C:\\", "downloads", "b.bin"), + requiredBytes: 450, + alreadyPresentBytes: 0 + })).rejects.toMatchObject({ + event: expect.objectContaining({ + phase: "download", + ownerId: "item-b", + volumeKey: "volume-a", + requiredBytes: 450, + availableBytes: 400, + deficitBytes: 50, + safetyBytes: 100, + retryAt: 30_010 + }) + }); + }); + + it("lets unknown sizes pass without consuming volume capacity", async () => { + const coordinator = new DiskReservationCoordinator({ + safetyBytes: 100, + statVolume: async () => { + throw new Error("capacity lookup should not run for unknown sizes"); + } + }); + + const lease = await coordinator.reserve({ + phase: "download", + ownerId: "unknown-item", + targetPath: path.join("D:\\", "downloads", "unknown.bin"), + requiredBytes: null, + alreadyPresentBytes: 0 + }); + + expect(lease.reservedBytes).toBe(0); + expect(coordinator.getReservedBytesByVolume().size).toBe(0); + }); + + it("updates and releases lease pressure on the owning volume", async () => { + const coordinator = new DiskReservationCoordinator({ + safetyBytes: 100, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "volume-b", + freeBytes: 1_200, + totalBytes: 2_000 + }) + }); + + const lease = await coordinator.reserve({ + phase: "download", + ownerId: "item-c", + targetPath: path.join("E:\\", "downloads", "c.bin"), + requiredBytes: 900, + alreadyPresentBytes: 100 + }); + + await lease.update({ requiredBytes: 600, alreadyPresentBytes: 400 }); + expect(lease.reservedBytes).toBe(200); + expect(coordinator.getReservedBytesByVolume().get("volume-b")).toBe(200); + + lease.release(); + + expect(lease.released).toBe(true); + expect(coordinator.getReservedBytesByVolume().get("volume-b")).toBe(0); + }); + + it("serializes parallel reservations so one volume cannot be overbooked", async () => { + const coordinator = new DiskReservationCoordinator({ + safetyBytes: 100, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "volume-c", + freeBytes: 1_000, + totalBytes: 2_000 + }) + }); + + const results = await Promise.allSettled([ + coordinator.reserve({ + phase: "download", + ownerId: "item-d", + targetPath: path.join("F:\\", "downloads", "d.bin"), + requiredBytes: 600, + alreadyPresentBytes: 0 + }), + coordinator.reserve({ + phase: "download", + ownerId: "item-e", + targetPath: path.join("F:\\", "downloads", "e.bin"), + requiredBytes: 600, + alreadyPresentBytes: 0 + }) + ]); + + expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1); + expect(results.filter((result) => result.status === "rejected")).toHaveLength(1); + expect(coordinator.getReservedBytesByVolume().get("volume-c")).toBe(600); + }); + + it("calculates conservative download and extraction reservation sizes", () => { + expect(calculateRemainingReservationBytes(1_000, 250)).toBe(750); + expect(calculateRemainingReservationBytes(1_000, 1_500)).toBe(0); + expect(calculateRemainingReservationBytes(null, 250)).toBeNull(); + expect(calculateExtractionReservationBytes([500, null, 200, 0])).toBe(700); + expect(calculateExtractionReservationBytes([null, 0])).toBeNull(); + }); +}); diff --git a/tests/download-manager.test.ts b/tests/download-manager.test.ts index d3d7e45..8302b80 100644 --- a/tests/download-manager.test.ts +++ b/tests/download-manager.test.ts @@ -7,18 +7,20 @@ import { EventEmitter, once } from "node:events"; import AdmZip from "adm-zip"; import { afterEach, describe, expect, it, vi } from "vitest"; import { DownloadManager, buildAutoRenameBaseNameFromFoldersWithOptions, extractArchiveNameFromExtractorLogMessage, getAuthoritativeRealDebridTotal, getDiskWriteWaitReason, resolveArchiveItemsFromList, runWithLimitedConcurrency } from "../src/main/download-manager"; -import { planDownloadCompletion, validateDownloadedFileCompletion } from "../src/main/download-completion"; -import { defaultSettings } from "../src/main/constants"; +import { planDownloadCompletion, validateDownloadedFileCompletion } from "../src/main/download-completion"; +import { DiskReservationCoordinator } from "../src/main/disk-space"; +import { defaultSettings } from "../src/main/constants"; import { parseDebridLinkApiKeys } from "../src/shared/debrid-link-keys"; import { getProviderUsageDayKey } from "../src/shared/provider-daily-limits"; import { getItemLogPath, initItemLogs, shutdownItemLogs } from "../src/main/item-log"; import { initPackageLogs, shutdownPackageLogs } from "../src/main/package-log"; import { createStoragePaths, emptySession } from "../src/main/storage"; import { primeDebridLinkRuntimeCooldownForTests, resetDebridLinkRuntimeStateForTests, primeMegaDebridRuntimeCooldownForTests, resetMegaDebridRuntimeStateForTests, primeMegaDebridInFlightForTests } from "../src/main/debrid"; -import { getMegaDebridAccountId } from "../src/shared/mega-debrid-accounts"; -import { getRenameLogPath, initRenameLog, shutdownRenameLog } from "../src/main/rename-log"; +import { getMegaDebridAccountId } from "../src/shared/mega-debrid-accounts"; +import { getRenameLogPath, initRenameLog, shutdownRenameLog } from "../src/main/rename-log"; import { UnrestrictedLink } from "../src/main/realdebrid"; -import type { HistoryEntry } from "../src/shared/types"; +import { resetVideoToolingCache } from "../src/main/video-processor"; +import type { HistoryEntry, PackageEntry } from "../src/shared/types"; const tempDirs: string[] = []; const originalFetch = globalThis.fetch; @@ -124,6 +126,201 @@ describe("disk write recovery", () => { expect(session.packages[packageId].status).toBe("queued"); }); + it("parks downloads before opening the target file when the final known size does not fit", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-disk-reserve-download-")); + tempDirs.push(root); + const session = emptySession(); + const packageId = "reserve-download-package"; + const itemId = "reserve-download-item"; + const outputDir = path.join(root, "downloads", "reserve-download"); + const extractDir = path.join(root, "extract", "reserve-download"); + const createdAt = Date.now(); + session.running = true; + session.packageOrder = [packageId]; + session.packages[packageId] = { + id: packageId, + name: "reserve-download", + outputDir, + extractDir, + status: "downloading", + itemIds: [itemId], + cancelled: false, + enabled: true, + createdAt, + updatedAt: createdAt + }; + session.items[itemId] = { + id: itemId, + packageId, + url: "https://rapidgator.net/file/reserve-download", + provider: "realdebrid", + status: "downloading", + retries: 0, + speedBps: 0, + downloadedBytes: 0, + totalBytes: null, + progressPercent: 0, + fileName: "reserve-download.bin", + targetPath: "", + resumable: true, + attempts: 0, + lastError: "", + fullStatus: "Download läuft", + createdAt, + updatedAt: createdAt + }; + const manager = new DownloadManager( + { ...defaultSettings(), token: "rd-token", outputDir: path.join(root, "downloads"), extractDir: path.join(root, "extract"), autoExtract: false }, + session, + createStoragePaths(path.join(root, "state")) + ); + (manager as any).diskReservations = new DiskReservationCoordinator({ + safetyBytes: 128, + retryDelayMs: 45_000, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "download-volume", + freeBytes: 512, + totalBytes: 2_048 + }) + }); + (manager as any).debridService.unrestrictLink = async () => ({ + fileName: "reserve-download.bin", + directUrl: "https://dummy/reserve-download", + fileSize: 1_024, + retriesUsed: 0, + provider: "realdebrid", + providerLabel: "Real-Debrid" + }); + globalThis.fetch = vi.fn(async () => new Response(Buffer.alloc(16, 1), { + status: 200, + headers: { + "content-length": "1024", + "accept-ranges": "bytes" + } + })) as typeof fetch; + const active = { itemId, packageId, abortController: new AbortController(), abortReason: "none", resumable: true, nonResumableCounted: false, blockedOnDiskWrite: false, blockedOnDiskSince: 0 }; + (manager as any).activeTasks.set(itemId, active); + + const before = Date.now(); + await (manager as any).processItem(active); + (manager as any).activeTasks.delete(itemId); + + const item = session.items[itemId]; + expect(item).toEqual(expect.objectContaining({ + status: "queued", + retries: 0, + speedBps: 0, + fullStatus: "Warte auf Festplatte" + })); + expect((manager as any).retryAfterByItem.get(itemId)).toBeGreaterThanOrEqual(before + 45_000); + expect(fs.existsSync(path.join(outputDir, "reserve-download.bin"))).toBe(false); + expect(manager.getSnapshot().diskWaitEvents?.[0]).toEqual(expect.objectContaining({ + phase: "download", + ownerId: itemId, + itemId, + packageId, + volumeKey: "download-volume", + requiredBytes: 1_024, + availableBytes: 384, + deficitBytes: 640 + })); + }); + + it("keeps disk-wait downloads out of the scheduler until their capacity retry is due", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-disk-reserve-resume-")); + tempDirs.push(root); + const session = emptySession(); + const packageId = "reserve-resume-package"; + const itemId = "reserve-resume-item"; + const outputDir = path.join(root, "downloads", "reserve-resume"); + const extractDir = path.join(root, "extract", "reserve-resume"); + const createdAt = Date.now(); + let freeBytes = 512; + session.running = true; + session.packageOrder = [packageId]; + session.packages[packageId] = { + id: packageId, + name: "reserve-resume", + outputDir, + extractDir, + status: "downloading", + itemIds: [itemId], + cancelled: false, + enabled: true, + createdAt, + updatedAt: createdAt + }; + session.items[itemId] = { + id: itemId, + packageId, + url: "https://rapidgator.net/file/reserve-resume", + provider: "realdebrid", + status: "downloading", + retries: 0, + speedBps: 0, + downloadedBytes: 0, + totalBytes: null, + progressPercent: 0, + fileName: "reserve-resume.bin", + targetPath: "", + resumable: true, + attempts: 0, + lastError: "", + fullStatus: "Download läuft", + createdAt, + updatedAt: createdAt + }; + const manager = new DownloadManager( + { ...defaultSettings(), token: "rd-token", outputDir: path.join(root, "downloads"), extractDir: path.join(root, "extract"), autoExtract: false }, + session, + createStoragePaths(path.join(root, "state")) + ); + (manager as any).diskReservations = new DiskReservationCoordinator({ + safetyBytes: 128, + retryDelayMs: 45_000, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "download-volume", + freeBytes, + totalBytes: 2_048 + }) + }); + (manager as any).debridService.unrestrictLink = async () => ({ + fileName: "reserve-resume.bin", + directUrl: "https://dummy/reserve-resume", + fileSize: 1_024, + retriesUsed: 0, + provider: "realdebrid", + providerLabel: "Real-Debrid" + }); + globalThis.fetch = vi.fn(async () => new Response(Buffer.alloc(1_024, 2), { + status: 200, + headers: { + "content-length": "1024", + "accept-ranges": "bytes" + } + })) as typeof fetch; + const firstActive = { itemId, packageId, abortController: new AbortController(), abortReason: "none", resumable: true, nonResumableCounted: false, blockedOnDiskWrite: false, blockedOnDiskSince: 0 }; + (manager as any).activeTasks.set(itemId, firstActive); + await (manager as any).processItem(firstActive); + (manager as any).activeTasks.delete(itemId); + + expect((manager as any).findNextQueuedItem()).toBeNull(); + + freeBytes = 2_048; + (manager as any).retryAfterByItem.set(itemId, Date.now() - 1); + expect((manager as any).findNextQueuedItem()).toEqual({ packageId, itemId }); + const secondActive = { itemId, packageId, abortController: new AbortController(), abortReason: "none", resumable: true, nonResumableCounted: false, blockedOnDiskWrite: false, blockedOnDiskSince: 0 }; + (manager as any).activeTasks.set(itemId, secondActive); + await (manager as any).processItem(secondActive); + (manager as any).activeTasks.delete(itemId); + + expect(session.items[itemId].status).toBe("completed"); + expect(fs.statSync(path.join(outputDir, "reserve-resume.bin")).size).toBe(1_024); + expect((manager as any).diskReservations.getReservedBytesByVolume().get("download-volume")).toBe(0); + }); + it("marks a fully downloaded package as failed when post-processing failed", () => { const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-extract-status-")); tempDirs.push(root); @@ -174,6 +371,165 @@ describe("disk write recovery", () => { expect(session.packages[packageId].status).toBe("failed"); }); + + it("parks extraction before the extractor starts when the extract volume is short", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-disk-reserve-extract-")); + tempDirs.push(root); + const session = emptySession(); + const packageId = "reserve-extract-package"; + const itemId = "reserve-extract-item"; + const outputDir = path.join(root, "downloads", "reserve-extract"); + const extractDir = path.join(root, "extract", "reserve-extract"); + const archivePath = path.join(outputDir, "reserve-extract.zip"); + const createdAt = Date.now(); + fs.mkdirSync(outputDir, { recursive: true }); + const zip = new AdmZip(); + zip.addFile("episode.mkv", Buffer.alloc(1_024, 3)); + zip.writeZip(archivePath); + const archiveBytes = fs.statSync(archivePath).size; + session.running = true; + session.packageOrder = [packageId]; + session.packages[packageId] = { + id: packageId, + name: "reserve-extract", + outputDir, + extractDir, + status: "completed", + itemIds: [itemId], + cancelled: false, + enabled: true, + createdAt, + updatedAt: createdAt + }; + session.items[itemId] = { + id: itemId, + packageId, + url: "https://rapidgator.net/file/reserve-extract", + provider: "realdebrid", + status: "completed", + retries: 0, + speedBps: 0, + downloadedBytes: archiveBytes, + totalBytes: archiveBytes, + progressPercent: 100, + fileName: "reserve-extract.zip", + targetPath: archivePath, + resumable: true, + attempts: 1, + lastError: "", + fullStatus: "Entpacken - Ausstehend", + createdAt, + updatedAt: createdAt + }; + const manager = new DownloadManager( + { ...defaultSettings(), token: "rd-token", outputDir: path.join(root, "downloads"), extractDir: path.join(root, "extract"), autoExtract: true }, + session, + createStoragePaths(path.join(root, "state")) + ); + (manager as any).diskReservations = new DiskReservationCoordinator({ + safetyBytes: 0, + retryDelayMs: 60_000, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "extract-volume", + freeBytes: Math.max(0, archiveBytes - 1), + totalBytes: archiveBytes * 2 + }) + }); + + const before = Date.now(); + await (manager as any).handlePackagePostProcessing(packageId); + + expect(session.packages[packageId].status).toBe("queued"); + expect(session.items[itemId]).toEqual(expect.objectContaining({ + status: "completed", + fullStatus: "Warte auf Festplatte", + lastError: "Zu wenig Speicherplatz" + })); + expect((manager as any).packageDiskRetryAfterByPackage.get(packageId)).toBeGreaterThanOrEqual(before + 60_000); + expect(manager.getSnapshot().diskWaitEvents?.[0]).toEqual(expect.objectContaining({ + phase: "extract", + ownerId: packageId, + packageId, + volumeKey: "extract-volume", + requiredBytes: archiveBytes, + availableBytes: archiveBytes - 1, + deficitBytes: 1 + })); + expect(fs.existsSync(path.join(extractDir, "episode.mkv"))).toBe(false); + }); + + it("skips audio remux before processVideoFile when the source volume is short", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-disk-reserve-remux-")); + tempDirs.push(root); + const extractDir = path.join(root, "extract", "reserve-remux"); + const sourcePath = path.join(extractDir, "Show.S01E01.German.DL.720p.mkv"); + fs.mkdirSync(extractDir, { recursive: true }); + fs.writeFileSync(sourcePath, Buffer.alloc(1_024, 4)); + process.env.RD_FFMPEG_BIN = "pwsh"; + process.env.RD_FFPROBE_BIN = "pwsh"; + resetVideoToolingCache(); + const manager = new DownloadManager( + { + ...defaultSettings(), + token: "rd-token", + outputDir: path.join(root, "downloads"), + extractDir, + keepGermanAudioOnly: true, + germanAudioMode: "tag" + }, + emptySession(), + createStoragePaths(path.join(root, "state")) + ); + const pkg: PackageEntry = { + id: "reserve-remux-package", + name: "reserve-remux", + outputDir: path.join(root, "downloads", "reserve-remux"), + extractDir, + status: "completed", + itemIds: [], + cancelled: false, + enabled: true, + createdAt: Date.now(), + updatedAt: Date.now() + }; + (manager as any).diskReservations = new DiskReservationCoordinator({ + safetyBytes: 128, + retryDelayMs: 30_000, + statVolume: async (targetPath) => ({ + path: targetPath, + volumeKey: "remux-volume", + freeBytes: 512, + totalBytes: 2_048 + }) + }); + + const processed = await (manager as any).keepGermanAudioOnlyImpl(extractDir, pkg); + + expect(processed).toBe(0); + expect(fs.existsSync(sourcePath)).toBe(true); + expect(fs.readdirSync(extractDir)).toEqual(["Show.S01E01.German.DL.720p.mkv"]); + expect(pkg.audioStripSummary).toMatchObject({ + candidates: 1, + remuxed: 0, + failed: 1 + }); + expect(pkg.audioStripSummary!.files[0]).toMatchObject({ + name: "Show.S01E01.German.DL.720p.mkv", + action: "skipped-no-space", + reason: "zu wenig freier Speicher fuer Remux" + }); + expect(manager.getSnapshot().diskWaitEvents?.[0]).toEqual(expect.objectContaining({ + phase: "remux", + ownerId: "reserve-remux-package", + packageId: "reserve-remux-package", + volumeKey: "remux-volume", + requiredBytes: 1_024, + availableBytes: 384, + deficitBytes: 640 + })); + expect((manager as any).diskReservations.getReservedBytesByVolume().get("remux-volume") ?? 0).toBe(0); + }); }); describe("download start account gate", () => { @@ -402,11 +758,14 @@ async function removeDirWithRetries(dir: string): Promise { } } -afterEach(async () => { - globalThis.fetch = originalFetch; - resetDebridLinkRuntimeStateForTests(); - resetMegaDebridRuntimeStateForTests(); - shutdownItemLogs(); +afterEach(async () => { + globalThis.fetch = originalFetch; + delete process.env.RD_FFMPEG_BIN; + delete process.env.RD_FFPROBE_BIN; + resetVideoToolingCache(); + resetDebridLinkRuntimeStateForTests(); + resetMegaDebridRuntimeStateForTests(); + shutdownItemLogs(); shutdownPackageLogs(); shutdownRenameLog(); for (const dir of tempDirs.splice(0)) {