From b3f60eada814b71d56796e8f28652c4892416975 Mon Sep 17 00:00:00 2001 From: Sucukdeluxe Date: Sat, 22 Aug 2026 07:37:40 +0200 Subject: [PATCH] fix(notifications): harden stall monitor lifecycle --- src/main/app-controller.ts | 3 + src/main/download-manager.ts | 165 +++++++++++++++----------- tests/download-manager.test.ts | 90 +++++++++++++- tests/main-shutdown-lifecycle.test.ts | 122 +++++++++++++++++++ 4 files changed, 309 insertions(+), 71 deletions(-) diff --git a/src/main/app-controller.ts b/src/main/app-controller.ts index 62f1f40..89ac69f 100644 --- a/src/main/app-controller.ts +++ b/src/main/app-controller.ts @@ -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; diff --git a/src/main/download-manager.ts b/src/main/download-manager.ts index cc6a743..5798a11 100644 --- a/src/main/download-manager.ts +++ b/src/main/download-manager.ts @@ -6190,11 +6190,10 @@ export class DownloadManager extends EventEmitter { this.speedBytesPerPackage.clear(); this.speedEventsHead = 0; this.lastGlobalProgressBytes = 0; - this.lastGlobalProgressAt = nowMs(); - this.lastReconnectMarkAt = 0; - this.consecutiveReconnects = 0; - this.globalSpeedLimitQueue = Promise.resolve(); - this.globalSpeedLimitNextAt = 0; + this.lastGlobalProgressAt = nowMs(); + this.lastReconnectMarkAt = 0; + this.consecutiveReconnects = 0; + this.resetGlobalSpeedLimitState(); this.summary = null; this.nonResumableActive = 0; this.persistSoon(); @@ -6304,11 +6303,10 @@ export class DownloadManager extends EventEmitter { this.speedBytesPerPackage.clear(); this.speedEventsHead = 0; this.lastGlobalProgressBytes = 0; - this.lastGlobalProgressAt = nowMs(); - this.lastReconnectMarkAt = 0; - this.consecutiveReconnects = 0; - this.globalSpeedLimitQueue = Promise.resolve(); - this.globalSpeedLimitNextAt = 0; + this.lastGlobalProgressAt = nowMs(); + this.lastReconnectMarkAt = 0; + this.consecutiveReconnects = 0; + this.resetGlobalSpeedLimitState(); this.summary = null; this.nonResumableActive = 0; this.persistSoon(); @@ -6461,11 +6459,10 @@ export class DownloadManager extends EventEmitter { this.speedEvents = []; this.speedBytesLastWindow = 0; this.speedBytesPerPackage.clear(); - this.speedEventsHead = 0; - this.lastGlobalProgressBytes = 0; - this.lastGlobalProgressAt = nowMs(); - this.globalSpeedLimitQueue = Promise.resolve(); - this.globalSpeedLimitNextAt = 0; + this.speedEventsHead = 0; + this.lastGlobalProgressBytes = 0; + this.lastGlobalProgressAt = nowMs(); + this.resetGlobalSpeedLimitState(); this.summary = null; this.nonResumableActive = 0; this.persistSoon(); @@ -12359,9 +12356,23 @@ export class DownloadManager extends EventEmitter { private cachedSpeedLimitAt = 0; - private globalSpeedLimitQueue: Promise = Promise.resolve(); - - private globalSpeedLimitNextAt = 0; + private globalSpeedLimitQueue: Promise = Promise.resolve(); + + 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(); @@ -12407,62 +12418,79 @@ export class DownloadManager extends EventEmitter { private async applyGlobalSpeedLimit(chunkBytes: number, bytesPerSecond: number, active?: ActiveTask): Promise { 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 (signal?.aborted) { - throw new Error("aborted:speed_limit"); - } + .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((resolve, reject) => { - let timer: NodeJS.Timeout | null = setTimeout(() => { + await new Promise((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; - } + throw new Error("aborted:speed_limit"); + } + + const startAt = Math.max(nowMs(), this.globalSpeedLimitNextAt); + this.globalSpeedLimitNextAt = startAt + durationMs; + }); + + this.globalSpeedLimitQueue = 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 { const signal = active?.abortController.signal; @@ -14092,11 +14120,10 @@ export class DownloadManager extends EventEmitter { this.claimedTargetPathByItem.clear(); this.itemContributedBytes.clear(); this.speedEvents = []; - this.speedEventsHead = 0; - this.speedBytesLastWindow = 0; - this.speedBytesPerPackage.clear(); - this.globalSpeedLimitQueue = Promise.resolve(); - this.globalSpeedLimitNextAt = 0; + this.speedEventsHead = 0; + this.speedBytesLastWindow = 0; + this.speedBytesPerPackage.clear(); + this.resetGlobalSpeedLimitState(); this.nonResumableActive = 0; this.lastGlobalProgressBytes = this.session.totalDownloadedBytes; this.lastGlobalProgressAt = nowMs(); diff --git a/tests/download-manager.test.ts b/tests/download-manager.test.ts index 84c703f..4ef53b4 100644 --- a/tests/download-manager.test.ts +++ b/tests/download-manager.test.ts @@ -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); diff --git a/tests/main-shutdown-lifecycle.test.ts b/tests/main-shutdown-lifecycle.test.ts index 51fa439..619b79f 100644 --- a/tests/main-shutdown-lifecycle.test.ts +++ b/tests/main-shutdown-lifecycle.test.ts @@ -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; 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 }); + } + }); });