diff --git a/src/main/download-manager.ts b/src/main/download-manager.ts index 849a285..66890ed 100644 --- a/src/main/download-manager.ts +++ b/src/main/download-manager.ts @@ -11784,10 +11784,12 @@ export class DownloadManager extends EventEmitter { } private trackActiveRunPackage(packageId: string): void { - if (!this.activeRunContextId) { - return; + let context = this.activeRunContextId ? this.runContexts.get(this.activeRunContextId) : undefined; + if (!context && this.session.running && this.runItemIds.size > 0) { + const startedAt = this.session.runStartedAt || nowMs(); + this.session.runStartedAt = startedAt; + context = this.beginActiveRunContext(this.runPackageIds, startedAt); } - const context = this.runContexts.get(this.activeRunContextId); if (context) { const generation = this.getPackageResultGeneration(packageId); context.packageGenerations.set(packageId, generation); @@ -11910,6 +11912,8 @@ export class DownloadManager extends EventEmitter { let remainingBytes = 0; let openItems = 0; let unknownCount = 0; + let finalizingItems = 0; + let speedBps = 0; for (const itemId of this.runItemIds) { const item = this.session.items[itemId]; if (!item || isFinishedStatus(item.status)) { @@ -11919,17 +11923,25 @@ export class DownloadManager extends EventEmitter { if (!pkg || pkg.cancelled || !pkg.enabled) { continue; } + if ((item.status === "downloading" || item.status === "integrity_check") + && item.totalBytes !== null + && item.totalBytes > 0 + && item.downloadedBytes >= item.totalBytes) { + finalizingItems += 1; + continue; + } openItems += 1; openPackages.add(pkg.id); + speedBps += Math.max(0, Math.floor(item.speedBps)); if (item.totalBytes === null) { unknownCount += 1; } else { remainingBytes += Math.max(0, item.totalBytes - item.downloadedBytes); } } - const speedBps = !this.session.running || this.session.paused - ? 0 - : Math.max(0, Math.floor(this.speedBytesLastWindow / SPEED_WINDOW_SECONDS)); + if (!this.session.running || this.session.paused) { + speedBps = 0; + } const etaSeconds = unknownCount === 0 && speedBps > 0 ? Math.ceil(remainingBytes / speedBps) : remainingBytes === 0 && unknownCount === 0 @@ -11940,6 +11952,7 @@ export class DownloadManager extends EventEmitter { openItems, openPackages: openPackages.size, unknownCount, + finalizingItems, speedBps, etaSeconds }; @@ -11951,6 +11964,9 @@ export class DownloadManager extends EventEmitter { return; } const current = this.buildRunRemainingSnapshot(); + if (current.finalizingItems > 0) { + return; + } const previous = context.remainingNotification.snapshot; context.remainingNotification.snapshot = current; if (!this.settings.notifyOnRemainingBelow) { diff --git a/src/main/notification-events.ts b/src/main/notification-events.ts index 1c549a8..fb32c00 100644 --- a/src/main/notification-events.ts +++ b/src/main/notification-events.ts @@ -67,6 +67,7 @@ export interface RunRemainingSnapshot { openItems: number; openPackages: number; unknownCount: number; + finalizingItems: number; speedBps: number; etaSeconds: number; } @@ -104,6 +105,8 @@ export function evaluateRemainingThreshold( || current.openItems <= 0 || previous.unknownCount > 0 || current.unknownCount > 0 + || previous.finalizingItems > 0 + || current.finalizingItems > 0 || previous.remainingBytes <= threshold || current.remainingBytes > threshold) { return { emit: false }; diff --git a/tests/remaining-notification.test.ts b/tests/remaining-notification.test.ts index 5d54b35..686b92c 100644 --- a/tests/remaining-notification.test.ts +++ b/tests/remaining-notification.test.ts @@ -1,6 +1,8 @@ import fs from "node:fs"; +import http from "node:http"; import os from "node:os"; import path from "node:path"; +import { once } from "node:events"; import { afterEach, describe, expect, it, vi } from "vitest"; import { DownloadManager } from "../src/main/download-manager"; import { defaultSettings } from "../src/main/constants"; @@ -37,6 +39,7 @@ function snapshot(overrides: Partial = {}): RunRemainingSn openItems: 2, openPackages: 1, unknownCount: 0, + finalizingItems: 0, speedBps: 1024 ** 2, etaSeconds: 51 * 1024, ...overrides @@ -179,6 +182,158 @@ describe("remaining threshold evaluation", () => { }); describe("run-scoped remaining notifications", () => { + it("waits for real HTTP finalization before crossing and keeps genuine remaining work in the event", async () => { + const payload = Buffer.alloc(256 * 1024, 7); + const server = http.createServer((_request, response) => { + response.statusCode = 200; + response.setHeader("Accept-Ranges", "bytes"); + response.setHeader("Content-Length", String(payload.length)); + response.end(payload); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("server address unavailable"); + } + + const { manager, session, events, settings } = setupManager({ + notifyRemainingThresholdGb: (128 * 1024) / GIB + }); + const downloadingPackage = addPackage(session, "http-final-package", payload.length); + downloadingPackage.outputDir = path.join(settings.outputDir, downloadingPackage.id); + downloadingPackage.extractDir = path.join(settings.extractDir, downloadingPackage.id); + const remainingPackage = addPackage(session, "genuine-remaining-package", 64 * 1024); + remainingPackage.outputDir = path.join(settings.outputDir, remainingPackage.id); + remainingPackage.extractDir = path.join(settings.extractDir, remainingPackage.id); + const downloadingItem = session.items[downloadingPackage.itemIds[0]]; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + state.debridService.unrestrictLink = vi.fn(async () => ({ + fileName: downloadingItem.fileName, + directUrl: `http://127.0.0.1:${address.port}/download`, + fileSize: payload.length, + retriesUsed: 0, + provider: "realdebrid", + providerLabel: "Real-Debrid" + })); + const eventStatuses: string[] = []; + state.enqueueNotificationCallback = async (event: NotificationEvent) => { + events.push(event); + eventStatuses.push(downloadingItem.status); + }; + + try { + await manager.start(); + const active = { + itemId: downloadingItem.id, + packageId: downloadingPackage.id, + abortController: new AbortController(), + abortReason: "none", + resumable: true, + nonResumableCounted: false, + stallRetries: 0, + genericErrorRetries: 0, + unrestrictRetries: 0 + }; + state.activeTasks.set(downloadingItem.id, active); + + await state.processItem(active); + await flushNotifications(); + + expect(downloadingItem.status).toBe("completed"); + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(1); + expect(eventStatuses).toEqual(["completed"]); + expect(events[0].payload.fields).toContainEqual({ name: "Restmenge", value: "64 KB", inline: true }); + expect(events[0].payload.fields).toContainEqual({ name: "Offene Dateien", value: "1", inline: true }); + } finally { + manager.stop(); + server.close(); + await once(server, "close"); + } + }, 15_000); + + it("creates a stable run context when public download work joins a postprocess-only start", async () => { + const { manager, session, events } = setupManager(); + const postprocessPackage = addPackage(session, "postprocess-only-package", GIB, GIB); + const postprocessItem = session.items[postprocessPackage.itemIds[0]]; + postprocessItem.status = "completed"; + postprocessItem.progressPercent = 100; + postprocessItem.fullStatus = "Fertig"; + postprocessPackage.status = "completed"; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + let releasePostprocess = (): void => {}; + const postprocessGate = new Promise((resolve) => { + releasePostprocess = resolve; + }); + state.handlePackagePostProcessing = vi.fn(async () => postprocessGate); + const postprocessTask = state.runPackagePostProcessing(postprocessPackage.id); + await Promise.resolve(); + + await manager.start(); + expect(session.running).toBe(true); + expect(state.activeRunContextId).toBeNull(); + + manager.addPackages([{ + name: "late-download-package", + links: ["https://dummy/late-download.bin"], + fileNames: ["late-download.bin"] + }]); + const latePackageId = session.packageOrder.find((packageId) => packageId !== postprocessPackage.id); + if (!latePackageId) { + throw new Error("late package missing"); + } + const latePackage = session.packages[latePackageId]; + const lateItem = session.items[latePackage.itemIds[0]]; + lateItem.totalBytes = 51 * GIB; + await manager.startItems([lateItem.id]); + const runContextId = state.activeRunContextId; + expect(runContextId).toEqual(expect.any(String)); + expect(state.runContexts.get(runContextId)?.packageGenerations.has(latePackage.id)).toBe(true); + + lateItem.downloadedBytes = 2 * GIB; + manager.setPackagePriority(latePackage.id, "high"); + await flushNotifications(); + + expect(state.activeRunContextId).toBe(runContextId); + expect(events.filter((event) => event.type === "remaining_threshold_crossed")).toHaveLength(1); + + manager.stop(); + releasePostprocess(); + await postprocessTask; + }); + + it("derives speed and ETA only from open enabled current run items", async () => { + const { manager, session } = setupManager(); + const activePackage = addPackage(session, "scoped-speed-package", 60 * GIB, 10 * GIB); + const disabledPackage = addPackage(session, "disabled-speed-package", 100 * GIB); + const removedPackage = addPackage(session, "removed-speed-package", 200 * GIB); + const activeItem = session.items[activePackage.itemIds[0]]; + const disabledItem = session.items[disabledPackage.itemIds[0]]; + const removedItem = session.items[removedPackage.itemIds[0]]; + activeItem.speedBps = 2 * GIB; + disabledItem.speedBps = 100 * GIB; + removedItem.speedBps = 200 * GIB; + const state = internal(manager); + vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); + + await manager.start(); + disabledPackage.enabled = false; + delete session.items[removedItem.id]; + state.speedBytesLastWindow = 500 * GIB; + + expect(state.buildRunRemainingSnapshot()).toEqual({ + remainingBytes: 50 * GIB, + openItems: 1, + openPackages: 1, + unknownCount: 0, + finalizingItems: 0, + speedBps: 2 * GIB, + etaSeconds: 25 + }); + }); + it("calculates known remainder, speed and ETA only from open enabled items in the active run", async () => { const { manager, session } = setupManager(); const active = addPackage(session, "active-package", 60 * GIB, 9 * GIB); @@ -208,6 +363,7 @@ describe("run-scoped remaining notifications", () => { vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); await manager.start({ excludePackageIds: new Set([notStarted.id]) }); + session.items[active.itemIds[0]].speedBps = GIB; state.speedBytesLastWindow = GIB; expect(state.buildRunRemainingSnapshot()).toEqual({ @@ -215,6 +371,7 @@ describe("run-scoped remaining notifications", () => { openItems: 2, openPackages: 1, unknownCount: 0, + finalizingItems: 0, speedBps: GIB, etaSeconds: 51 }); @@ -230,6 +387,7 @@ describe("run-scoped remaining notifications", () => { vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined); await manager.start(); + item.speedBps = GIB; state.speedBytesLastWindow = GIB; item.downloadedBytes = GIB; state.evaluateRemainingNotification();