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

fix(notifications): harden active run lifecycle tracking

Browse Source
This commit is contained in:
Sucukdeluxe
2026-08-22 05:51:08 +02:00
parent 514d1b1d86
commit 96966c3eca
2 changed files with 208 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/download-manager.ts
+61 -10
View File
@@ -1945,6 +1945,8 @@ export class DownloadManager extends EventEmitter {
private standalonePackageResults = new Set<string>();
private suppressedPackageResults = new Set<string>();
private successDigestResults = new Map<string, PackageResultEnvelope>();
private successDigestTimer: NodeJS.Timeout | null = null;
@@ -3129,6 +3131,7 @@ export class DownloadManager extends EventEmitter {
this.runContexts.clear();
this.activeRunContextId = null;
this.standalonePackageResults.clear();
this.suppressedPackageResults.clear();
this.successDigestResults.clear();
if (this.successDigestTimer) {
clearTimeout(this.successDigestTimer);
@@ -6340,6 +6343,9 @@ export class DownloadManager extends EventEmitter {
const abortReason: "stop" | "shutdown" = parkForRestart ? "shutdown" : "stop";
const keepExtraction = this.settings.autoExtractWhenStopped;
const wasRunning = this.session.running;
const stoppedRunContext = wasRunning
? this.stopActiveRunContext(this.runPackageIds, this.session.runStartedAt)
: null;
this.schedulerGeneration += 1;
this.session.running = false;
this.session.paused = false;
@@ -6385,20 +6391,20 @@ export class DownloadManager extends EventEmitter {
pkg.updatedAt = nowMs();
}
}
if (wasRunning && !parkForRestart && this.settings.notifyOnRunFinished && this.runItemIds.size > 0) {
const packageResults = [...this.runPackageIds]
.flatMap((packageId) => {
const result = this.finalizedPackageResults.get(this.packageResultKey(packageId, this.getPackageResultGeneration(packageId)));
if (stoppedRunContext && !parkForRestart && this.settings.notifyOnRunFinished && this.runItemIds.size > 0) {
const packageResults = [...stoppedRunContext.packageGenerations]
.flatMap(([packageId, generation]) => {
const result = this.finalizedPackageResults.get(this.packageResultKey(packageId, generation));
return result ? [result] : [];
});
this.flushPackageSuccessDigest();
this.queueNotificationEvent(buildRunNotificationEvent(buildRunResult({
id: uuidv4(),
id: stoppedRunContext.id,
stopped: true,
startedAt: this.session.runStartedAt,
startedAt: stoppedRunContext.startedAt,
completedAt: nowMs(),
packages: packageResults,
totalPackages: this.runPackageIds.size
totalPackages: stoppedRunContext.packageGenerations.size
})));
}
this.runItemIds.clear();
@@ -8005,6 +8011,7 @@ export class DownloadManager extends EventEmitter {
}
private runPackagePostProcessing(packageId: string): Promise<void> {
this.trackPackagePostProcessResult(packageId);
const existing = this.packagePostProcessTasks.get(packageId);
if (existing) {
this.hybridExtractRequeue.add(packageId);
@@ -8269,6 +8276,7 @@ export class DownloadManager extends EventEmitter {
}
}
logger.info(`Entpacken via Start ausgelöst: pkg=${pkg.name}`);
this.trackPackagePostProcessResult(packageId);
void this.runPackagePostProcessing(packageId).catch((err) => logger.warn(`runPackagePostProcessing Fehler (triggerPending): ${compactErrorText(err)}`));
}
continue;
@@ -8288,6 +8296,7 @@ export class DownloadManager extends EventEmitter {
}
}
logger.info(`Hybrid-Entpacken via Start ausgelöst: pkg=${pkg.name}, completed=${success}/${items.length}`);
this.trackPackagePostProcessResult(packageId);
void this.runPackagePostProcessing(packageId).catch((err) => logger.warn(`runPackagePostProcessing Fehler (triggerPendingHybrid): ${compactErrorText(err)}`));
}
}
@@ -11743,6 +11752,9 @@ export class DownloadManager extends EventEmitter {
private beginActiveRunContext(packageIds: Iterable<string>, startedAt: number): RunLifecycleContext {
const context = this.createRunContext(packageIds, startedAt, false);
for (const [packageId, generation] of context.packageGenerations) {
this.suppressedPackageResults.delete(this.packageResultKey(packageId, generation));
}
this.activeRunContextId = context.id;
return context;
}
@@ -11752,8 +11764,10 @@ export class DownloadManager extends EventEmitter {
return;
}
const context = this.runContexts.get(this.activeRunContextId);
if (context && !context.packageGenerations.has(packageId)) {
context.packageGenerations.set(packageId, this.getPackageResultGeneration(packageId));
if (context) {
const generation = this.getPackageResultGeneration(packageId);
context.packageGenerations.set(packageId, generation);
this.suppressedPackageResults.delete(this.packageResultKey(packageId, generation));
}
}
@@ -11779,6 +11793,26 @@ export class DownloadManager extends EventEmitter {
return context;
}
private stopActiveRunContext(packageIds: Iterable<string>, startedAt: number): RunLifecycleContext {
const active = this.activeRunContextId ? this.runContexts.get(this.activeRunContextId) : undefined;
const context = active || this.createRunContext(packageIds, startedAt, false);
for (const packageId of packageIds) {
if (!context.packageGenerations.has(packageId)) {
context.packageGenerations.set(packageId, this.getPackageResultGeneration(packageId));
}
}
for (const [packageId, generation] of context.packageGenerations) {
const key = this.packageResultKey(packageId, generation);
this.standalonePackageResults.delete(key);
this.suppressedPackageResults.add(key);
}
this.runContexts.delete(context.id);
if (this.activeRunContextId === context.id) {
this.activeRunContextId = null;
}
return context;
}
private isPackageResultTracked(packageId: string, generation: number): boolean {
const key = this.packageResultKey(packageId, generation);
if (this.standalonePackageResults.has(key)) {
@@ -11793,7 +11827,22 @@ export class DownloadManager extends EventEmitter {
}
private trackStandalonePackageResult(packageId: string): void {
this.standalonePackageResults.add(this.packageResultKey(packageId, this.getPackageResultGeneration(packageId)));
const key = this.packageResultKey(packageId, this.getPackageResultGeneration(packageId));
this.suppressedPackageResults.delete(key);
this.standalonePackageResults.add(key);
}
private trackPackagePostProcessResult(packageId: string): void {
const generation = this.getPackageResultGeneration(packageId);
const key = this.packageResultKey(packageId, generation);
if (this.suppressedPackageResults.has(key)) {
return;
}
if (this.activeRunContextId) {
this.trackActiveRunPackage(packageId);
} else if (!this.isPackageResultTracked(packageId, generation)) {
this.standalonePackageResults.add(key);
}
}
private queueNotificationEvent(notification: NotificationEvent): void {
@@ -12683,6 +12732,7 @@ export class DownloadManager extends EventEmitter {
}
}
if (result.extracted > 0) {
this.trackPackagePostProcessResult(packageId);
const hybridController = new AbortController();
let hybridSet = this.packageHybridPostProcessControllers.get(packageId);
if (!hybridSet) {
@@ -13357,6 +13407,7 @@ export class DownloadManager extends EventEmitter {
alreadyMarkedExtracted: boolean,
extractedCount: number
): Promise<void> {
this.trackPackagePostProcessResult(packageId);
const task = this.executeDeferredPostExtraction(packageId, pkg, success, failed, alreadyMarkedExtracted, extractedCount)
.finally(() => {
const tasks = this.packageDeferredPostProcessTasks.get(packageId);
tests/notify-hooks.test.ts
+134
View File
@@ -488,6 +488,140 @@ describe("authoritative run completion", () => {
expect(history[0]).toMatchObject({ name: pkg.name, status: "completed", fileCount: 0 });
});
it("updates an active run from generation one to generation two after resetting the same package", async () => {
const { manager, session, events, history } = setup();
const pkg = addPackage(session);
const state = internal(manager);
session.running = true;
session.runStartedAt = Date.now() - 20_000;
state.runItemIds = new Set(pkg.itemIds);
state.runPackageIds = new Set([pkg.id]);
state.runOutcomes = new Map([[pkg.itemIds[0], "completed"]]);
state.beginActiveRunContext(state.runPackageIds, session.runStartedAt);
state.tryFinalizePackageResult(pkg.id);
const generationOne = state.finalizedPackageResults.get(`${pkg.id}:1`);
state.finalizedPackageResults.set(`${pkg.id}:1`, {
...generationOne,
status: "failed",
successfulFiles: 0,
failedFiles: 9,
failurePhase: "download",
errorCategory: "stale"
});
await flushNotifications();
events.length = 0;
history.length = 0;
vi.spyOn(state, "ensureScheduler").mockResolvedValue(undefined);
await manager.resetPackage(pkg.id);
expect(pkg.resultGeneration).toBe(2);
const item = session.items[pkg.itemIds[0]];
item.status = "completed";
item.downloadedBytes = 1_000;
item.totalBytes = 1_000;
item.progressPercent = 100;
item.fullStatus = "Fertig";
pkg.status = "completed";
state.runOutcomes.set(item.id, "completed");
state.tryFinalizePackageResult(pkg.id);
state.finishRun();
await flushNotifications();
expect(events.filter((event) => event.type === "package_completed")).toHaveLength(1);
expect(events.find((event) => event.type === "package_completed")?.id).toContain(":2:");
const completedRun = events.find((event) => event.type === "run_completed");
expect(completedRun?.payload.fields.some((field) => field.name === "Dateien" && field.value === "1 erfolgreich · 0 fehlgeschlagen · 0 abgebrochen")).toBe(true);
expect(history).toHaveLength(1);
expect(history[0].id).toBe(`hist-${pkg.id}-2`);
});
it("tracks a main postprocess task created by triggerPendingExtractions after start begins", async () => {
const { manager, session, events, history } = setup({ autoExtract: true });
const pkg = addPackage(session);
const state = internal(manager);
pkg.status = "completed";
session.items[pkg.itemIds[0]].fullStatus = "Fertig";
let releasePostProcess = (): void => {};
const postProcessGate = new Promise<void>((resolve) => {
releasePostProcess = resolve;
});
state.handlePackagePostProcessing = vi.fn(async () => postProcessGate);
await manager.start();
const postProcess = state.packagePostProcessTasks.get(pkg.id);
expect(postProcess).toBeDefined();
releasePostProcess();
await postProcess;
await flushNotifications();
expect(events.filter((event) => event.type === "package_completed")).toHaveLength(1);
expect(history).toHaveLength(1);
expect(history[0].name).toBe(pkg.name);
});
it("tracks a deferred-only startup task at creation without an active run", async () => {
const { manager, session, events, history } = setup();
const pkg = addPackage(session);
const state = internal(manager);
pkg.status = "completed";
session.items[pkg.itemIds[0]].fullStatus = "Entpackt - Done (1.0s)";
state.executeDeferredPostExtraction = vi.fn(async () => undefined);
await state.runDeferredPostExtraction(pkg.id, pkg, 1, 0, true, 1);
await flushNotifications();
expect(events.filter((event) => event.type === "package_completed")).toHaveLength(1);
expect(history).toHaveLength(1);
expect(history[0].name).toBe(pkg.name);
});
it("stops and removes the active run context before late package work can emit", async () => {
const { manager, session, events, history } = setup({ autoExtractWhenStopped: true });
const packageA = addPackage(session, ["completed"], "stopped-package");
const state = internal(manager);
session.running = true;
session.runStartedAt = Date.now() - 20_000;
state.runItemIds = new Set(packageA.itemIds);
state.runPackageIds = new Set([packageA.id]);
state.runOutcomes = new Map([[packageA.itemIds[0], "completed"]]);
const stoppedContext = state.beginActiveRunContext(state.runPackageIds, session.runStartedAt);
let releasePostProcess = (): void => {};
const postProcessGate = new Promise<void>((resolve) => {
releasePostProcess = resolve;
});
state.handlePackagePostProcessing = vi.fn(async () => postProcessGate);
const latePostProcess = state.runPackagePostProcessing(packageA.id);
await Promise.resolve();
manager.stop();
await flushNotifications();
const stoppedEvent = events.find((event) => event.type === "run_stopped");
expect(stoppedEvent?.id).toBe(`run:${stoppedContext.id}:run_stopped`);
expect(state.activeRunContextId).toBeNull();
expect(state.runContexts.has(stoppedContext.id)).toBe(false);
releasePostProcess();
await latePostProcess;
await flushNotifications();
expect(events.filter((event) => event.type === "package_completed")).toHaveLength(0);
expect(history).toHaveLength(0);
const packageB = addPackage(session, ["completed"], "follow-up-package");
session.running = true;
session.runStartedAt = Date.now() - 5_000;
state.runItemIds = new Set(packageB.itemIds);
state.runPackageIds = new Set([packageB.id]);
state.runOutcomes = new Map([[packageB.itemIds[0], "completed"]]);
state.beginActiveRunContext(state.runPackageIds, session.runStartedAt);
state.finishRun();
await flushNotifications();
expect(events.filter((event) => event.type === "run_stopped")).toHaveLength(1);
expect(events.filter((event) => event.type === "run_completed")).toHaveLength(1);
expect(events.filter((event) => event.type === "package_completed")).toHaveLength(1);
expect(history.map((entry) => entry.name)).toEqual([packageB.name]);
});
it("finalizes overlapping runs independently when the earlier run finishes deferred work last", async () => {
const { manager, session, events, history } = setup();
const packageA = addPackage(session, ["completed"], "package-a");