Administrator/Multi-Debrid-Downloader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 596 Wiki Activity

fix(notifications): harden remaining threshold scope

Browse Source
This commit is contained in:
Sucukdeluxe
2026-08-22 06:55:57 +02:00
parent 2142ae8bf0
commit 4fd30802e5
3 changed files with 183 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/download-manager.ts
+22 -6
View File
@@ -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) {
src/main/notification-events.ts
+3
View File
@@ -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 };
tests/remaining-notification.test.ts
+158
View File
@@ -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<RunRemainingSnapshot> = {}): 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<void>((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();