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

fix(notifications): harden stall monitor lifecycle

Browse Source
This commit is contained in:
Sucukdeluxe
2026-08-22 07:37:40 +02:00
parent 311475c93b
commit b3f60eada8
4 changed files with 309 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/app-controller.ts
+3
View File
@@ -1336,6 +1336,9 @@ export class AppController {
if (this.downloadHealthEvaluation) {
await this.downloadHealthEvaluation;
}
if (this.downloadHealthMonitor) {
await this.evaluateDownloadHealth();
}
if (this.runtimeStatsTimer) {
clearInterval(this.runtimeStatsTimer);
this.runtimeStatsTimer = null;
src/main/download-manager.ts
+67 -40
View File
@@ -6193,8 +6193,7 @@ export class DownloadManager extends EventEmitter {
this.lastGlobalProgressAt = nowMs();
this.lastReconnectMarkAt = 0;
this.consecutiveReconnects = 0;
this.globalSpeedLimitQueue = Promise.resolve();
this.globalSpeedLimitNextAt = 0;
this.resetGlobalSpeedLimitState();
this.summary = null;
this.nonResumableActive = 0;
this.persistSoon();
@@ -6307,8 +6306,7 @@ export class DownloadManager extends EventEmitter {
this.lastGlobalProgressAt = nowMs();
this.lastReconnectMarkAt = 0;
this.consecutiveReconnects = 0;
this.globalSpeedLimitQueue = Promise.resolve();
this.globalSpeedLimitNextAt = 0;
this.resetGlobalSpeedLimitState();
this.summary = null;
this.nonResumableActive = 0;
this.persistSoon();
@@ -6464,8 +6462,7 @@ export class DownloadManager extends EventEmitter {
this.speedEventsHead = 0;
this.lastGlobalProgressBytes = 0;
this.lastGlobalProgressAt = nowMs();
this.globalSpeedLimitQueue = Promise.resolve();
this.globalSpeedLimitNextAt = 0;
this.resetGlobalSpeedLimitState();
this.summary = null;
this.nonResumableActive = 0;
this.persistSoon();
@@ -12363,6 +12360,20 @@ export class DownloadManager extends EventEmitter {
private globalSpeedLimitNextAt = 0;
private globalSpeedLimitHealthNextAt = 0;
private globalSpeedLimitPending = 0;
private globalSpeedLimitGeneration = 0;
private resetGlobalSpeedLimitState(): void {
this.globalSpeedLimitGeneration += 1;
this.globalSpeedLimitQueue = Promise.resolve();
this.globalSpeedLimitNextAt = 0;
this.globalSpeedLimitHealthNextAt = 0;
this.globalSpeedLimitPending = 0;
}
private getEffectiveSpeedLimitKbps(): number {
const now = nowMs();
if (now - this.cachedSpeedLimitAt < 2000) {
@@ -12407,61 +12418,78 @@ export class DownloadManager extends EventEmitter {
private async applyGlobalSpeedLimit(chunkBytes: number, bytesPerSecond: number, active?: ActiveTask): Promise<void> {
const signal = active?.abortController.signal;
const generation = this.globalSpeedLimitGeneration;
const queuedAt = nowMs();
const durationMs = Math.max(1, Math.ceil((chunkBytes / bytesPerSecond) * 1000));
const healthReadyAt = Math.max(queuedAt, this.globalSpeedLimitNextAt, this.globalSpeedLimitHealthNextAt);
this.globalSpeedLimitHealthNextAt = healthReadyAt + durationMs;
this.globalSpeedLimitPending += 1;
if (active && healthReadyAt > queuedAt) {
active.blockedOnThrottleUntil = Math.max(active.blockedOnThrottleUntil || 0, healthReadyAt);
}
const task = this.globalSpeedLimitQueue
.catch(() => undefined)
.then(async () => {
if (generation !== this.globalSpeedLimitGeneration) {
throw new Error("aborted:speed_limit_generation");
}
if (signal?.aborted) {
throw new Error("aborted:speed_limit");
}
const now = nowMs();
const waitMs = Math.max(0, this.globalSpeedLimitNextAt - now);
if (waitMs > 0) {
if (active) {
active.blockedOnThrottleUntil = now + waitMs;
}
try {
await new Promise<void>((resolve, reject) => {
let timer: NodeJS.Timeout | null = setTimeout(() => {
await new Promise<void>((resolve, reject) => {
let timer: NodeJS.Timeout | null = setTimeout(() => {
timer = null;
signal?.removeEventListener("abort", onAbort);
resolve();
}, waitMs);
const onAbort = (): void => {
if (timer) {
clearTimeout(timer);
timer = null;
signal?.removeEventListener("abort", onAbort);
resolve();
}, waitMs);
const onAbort = (): void => {
if (timer) {
clearTimeout(timer);
timer = null;
}
signal?.removeEventListener("abort", onAbort);
reject(new Error("aborted:speed_limit"));
};
if (signal) {
if (signal.aborted) {
onAbort();
return;
}
signal.addEventListener("abort", onAbort, { once: true });
}
});
} finally {
if (active) {
active.blockedOnThrottleUntil = 0;
signal?.removeEventListener("abort", onAbort);
reject(new Error("aborted:speed_limit"));
};
if (signal) {
if (signal.aborted) {
onAbort();
return;
}
signal.addEventListener("abort", onAbort, { once: true });
}
}
});
}
if (generation !== this.globalSpeedLimitGeneration) {
throw new Error("aborted:speed_limit_generation");
}
if (signal?.aborted) {
throw new Error("aborted:speed_limit");
}
const startAt = Math.max(nowMs(), this.globalSpeedLimitNextAt);
const durationMs = Math.max(1, Math.ceil((chunkBytes / bytesPerSecond) * 1000));
this.globalSpeedLimitNextAt = startAt + durationMs;
});
this.globalSpeedLimitQueue = task;
await task;
try {
await task;
} finally {
if (active && active.blockedOnThrottleUntil === healthReadyAt) {
active.blockedOnThrottleUntil = 0;
}
if (generation === this.globalSpeedLimitGeneration) {
this.globalSpeedLimitPending = Math.max(0, this.globalSpeedLimitPending - 1);
if (this.globalSpeedLimitPending === 0) {
this.globalSpeedLimitHealthNextAt = Math.max(nowMs(), this.globalSpeedLimitNextAt);
}
}
}
}
private async applySpeedLimit(chunkBytes: number, localWindowBytes: number, localWindowStarted: number, active?: ActiveTask): Promise<void> {
@@ -14095,8 +14123,7 @@ export class DownloadManager extends EventEmitter {
this.speedEventsHead = 0;
this.speedBytesLastWindow = 0;
this.speedBytesPerPackage.clear();
this.globalSpeedLimitQueue = Promise.resolve();
this.globalSpeedLimitNextAt = 0;
this.resetGlobalSpeedLimitState();
this.nonResumableActive = 0;
this.lastGlobalProgressBytes = this.session.totalDownloadedBytes;
this.lastGlobalProgressAt = nowMs();
tests/download-manager.test.ts
+88 -2
View File
@@ -22,6 +22,7 @@ import { serializeRealDebridApiAccounts } from "../src/shared/real-debrid-accoun
import { getRenameLogPath, initRenameLog, shutdownRenameLog } from "../src/main/rename-log";
import { UnrestrictedLink } from "../src/main/realdebrid";
import { resetVideoToolingCache } from "../src/main/video-processor";
import { createDownloadHealthState, evaluateDownloadHealth } from "../src/main/download-health-monitor";
import type { AppSettings, DownloadItem, HistoryEntry, PackageEntry } from "../src/shared/types";
const tempDirs: string[] = [];
@@ -14698,7 +14699,7 @@ describe("package lifecycle telemetry boundaries", () => {
});
describe("download health snapshot", () => {
function createHealthManager(root: string) {
function createHealthManager(root: string, settings = defaultSettings()) {
const session = emptySession();
const packageId = "private-package-id";
const itemId = "private-item-id";
@@ -14741,7 +14742,7 @@ describe("download health snapshot", () => {
createdAt: 2_000,
updatedAt: now
};
const manager = new DownloadManager(defaultSettings(), session, createStoragePaths(path.join(root, "state")));
const manager = new DownloadManager(settings, session, createStoragePaths(path.join(root, "state")));
const state = manager as any;
session.running = true;
session.items[itemId].status = "downloading";
@@ -14909,6 +14910,91 @@ describe("download health snapshot", () => {
expect(health.nextRetryAt).toBe(0);
});
it("freezes health while two active downloads wait in the real global speed-limit queue", async () => {
vi.useFakeTimers();
vi.setSystemTime(10_000);
try {
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-health-global-throttle-"));
tempDirs.push(root);
const settings = {
...defaultSettings(),
speedLimitEnabled: true,
speedLimitMode: "global" as const,
speedLimitKbps: 1
};
const { manager, session, state, packageId, itemId } = createHealthManager(root, settings);
const secondItemId = "private-throttled-item-2";
session.items[secondItemId] = {
...session.items[itemId],
id: secondItemId
};
session.packages[packageId].itemIds.push(secondItemId);
state.runItemIds.add(secondItemId);
const firstActive = {
itemId,
packageId,
abortController: new AbortController(),
abortReason: "none",
resumable: true,
nonResumableCounted: false,
phase: "downloading",
phaseStartedAt: 10_000,
phaseDeadlineAt: 0,
blockedOnDiskWrite: false,
blockedOnDiskSince: 0,
blockedOnThrottleUntil: 0
};
const secondActive = {
...firstActive,
itemId: secondItemId,
abortController: new AbortController()
};
state.activeTasks.set(itemId, firstActive);
state.activeTasks.set(secondItemId, secondActive);
await state.applySpeedLimit(100 * 1024, 0, 10_000, firstActive);
const secondWait = state.applySpeedLimit(100 * 1024, 0, 10_000, secondActive);
const firstWait = state.applySpeedLimit(100 * 1024, 0, 10_000, firstActive);
const waitsSettled = Promise.allSettled([secondWait, firstWait]);
await Promise.resolve();
let healthState = createDownloadHealthState();
const events = [];
for (const now of [10_000, 55_000, 105_000]) {
vi.setSystemTime(now);
const result = evaluateDownloadHealth(
healthState,
manager.getDownloadHealthSnapshot(now),
now,
{
stallAfterMs: 90_000,
cooldownMs: 600_000,
notifyOnStall: true,
notifyOnRecovery: true
}
);
healthState = result.state;
events.push(...result.events);
}
expect(firstActive.blockedOnThrottleUntil).toBeGreaterThan(105_000);
expect(secondActive.blockedOnThrottleUntil).toBeGreaterThan(105_000);
expect(manager.getDownloadHealthSnapshot(105_000).blockedOnThrottleUntil).toBeGreaterThan(105_000);
expect(healthState.status).toBe("expected_wait");
expect(healthState.suspiciousDurationMs).toBe(0);
expect(events).toEqual([]);
firstActive.abortController.abort("test-finished");
secondActive.abortController.abort("test-finished");
await vi.runAllTimersAsync();
await waitsSettled;
expect(firstActive.blockedOnThrottleUntil).toBe(0);
expect(secondActive.blockedOnThrottleUntil).toBe(0);
} finally {
vi.useRealTimers();
}
});
it("counts one technical recovery when the existing global watchdog restarts stalled tasks", () => {
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-health-watchdog-"));
tempDirs.push(root);
tests/main-shutdown-lifecycle.test.ts
+122
View File
@@ -1,3 +1,6 @@
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { describe, expect, it, vi } from "vitest";
const electron = vi.hoisted(() => {
@@ -33,6 +36,13 @@ vi.mock("electron", () => ({
}));
import { AppController } from "../src/main/app-controller";
import {
DownloadHealthMonitor,
createDownloadHealthState,
loadDownloadHealthState,
saveDownloadHealthState,
type DownloadHealthSnapshot
} from "../src/main/download-health-monitor";
function deferred(): { promise: Promise<void>; resolve: () => void } {
let resolve = () => {};
@@ -97,4 +107,116 @@ describe("main shutdown lifecycle", () => {
expect(cleanup).toHaveBeenCalledTimes(1);
expect(onError).not.toHaveBeenCalled();
});
it("waits for a running health sample and persistently closes an alerted incident before shutdown returns", async () => {
const root = fs.mkdtempSync(path.join(os.tmpdir(), "rd-health-shutdown-"));
try {
const filePath = path.join(root, "health.json");
const runFingerprint = "a".repeat(64);
const queueFingerprint = "b".repeat(64);
saveDownloadHealthState(filePath, createDownloadHealthState({
status: "alerted",
runFingerprint,
queueFingerprint,
suspiciousDurationMs: 90_000,
suspiciousSamples: 3,
incidentStartedAt: 10_000,
alertedAt: 90_000,
lastAlertAt: 90_000,
cooldownUntil: 690_000
}));
let shuttingDown = false;
const healthSnapshot = (completionSequence: number): DownloadHealthSnapshot => ({
runActive: true,
runFingerprint,
queueFingerprint,
openItems: 1,
openPackages: 1,
knownDownloadedBytes: 4096,
activeTasks: 1,
startableItems: 0,
lastSchedulerTickAt: 100_000,
downloadProgressSequence: 0,
itemCompletionSequence: completionSequence,
lastPositiveByteAt: 0,
technicalRecoveryCount: 0,
paused: false,
reconnectUntil: 0,
nextRetryAt: 0,
providerCooldownUntil: 0,
blockedOnDisk: false,
blockedOnThrottleUntil: 0,
activePhaseDeadlineAt: 0,
terminalFailure: false,
manualStop: false,
shuttingDown,
currentSpeedBps: 0
});
const runningEvaluation = deferred();
const controller = Object.create(AppController.prototype) as any;
controller.downloadHealthTimer = setInterval(() => {}, 60_000);
controller.downloadHealthTimer.unref?.();
controller.downloadHealthMonitor = new DownloadHealthMonitor(filePath);
controller.downloadHealthEvaluation = runningEvaluation.promise.finally(() => {
controller.downloadHealthEvaluation = null;
});
controller.runtimeStatsTimer = null;
controller.notificationOutbox = {
enqueue: vi.fn(async () => undefined),
drainForShutdown: vi.fn(async () => undefined)
};
controller.manager = {
suspendDownloadHealthMonitoring: vi.fn(() => { shuttingDown = true; }),
getDownloadHealthSnapshot: vi.fn(() => healthSnapshot(1)),
flushNotificationsForShutdown: vi.fn(async () => undefined),
prepareForShutdown: vi.fn()
};
controller.megaWebFallback = { dispose: vi.fn() };
controller.realDebridWebFallbacks = new Map();
controller.pendingRealDebridWebAccountIds = new Map();
controller.allDebridWebFallback = { dispose: vi.fn() };
controller.bestDebridWebFallback = { dispose: vi.fn() };
controller.shutdownLogStorage = vi.fn();
controller.audit = vi.fn();
controller.settings = {
historyRetentionMode: "never",
notifyUrl: "https://discord.example.test/webhook",
notifyOnDownloadStall: true,
notifyOnDownloadRecovery: true,
notifyStallAfterSeconds: 90,
notifyStallCooldownMinutes: 10
};
const shutdown = controller.shutdown();
expect(controller.downloadHealthTimer).toBeNull();
expect(controller.notificationOutbox.drainForShutdown).not.toHaveBeenCalled();
runningEvaluation.resolve();
await shutdown;
const closed = loadDownloadHealthState(filePath);
expect(closed.status).toBe("idle");
expect(closed.alertedAt).toBe(0);
expect(closed.restartPending).toBe(false);
const recoveredEvents: unknown[] = [];
const restarted = new DownloadHealthMonitor(filePath);
await restarted.sample(healthSnapshot(1), 120_000, {
stallAfterMs: 90_000,
cooldownMs: 600_000,
notifyOnStall: true,
notifyOnRecovery: true
}, async (event) => { recoveredEvents.push(event); });
await restarted.sample(healthSnapshot(2), 135_000, {
stallAfterMs: 90_000,
cooldownMs: 600_000,
notifyOnStall: true,
notifyOnRecovery: true
}, async (event) => { recoveredEvents.push(event); });
expect(recoveredEvents).toEqual([]);
} finally {
fs.rmSync(root, { recursive: true, force: true });
}
});
});