fix(notifications): harden final delivery lifecycle

Enforce one shared shutdown deadline while persisting late digests during blocked sends. Start stall cooldowns from serialized delivery acknowledgements, atomically rewrite cleaned legacy outboxes, retain run summaries for 24 hours, and prevent duplicate history or synthetic post-process start times.
2026-08-22 08:58:36 +02:00
parent 6715f42ff8
commit 3ec9085444
10 changed files with 733 additions and 77 deletions
@@ -173,7 +173,14 @@ export class AppController {
        color: event.payload.color ?? (event.priority === "error" ? 0xe74c3c : 0x2ecc71),
        fields: event.payload.fields,
        timestamp: event.createdAt
-      })
+      }),
+      onDelivered: (event, deliveredAt) => {
+        return this.downloadHealthMonitor?.acknowledgeDelivery(
+          event,
+          deliveredAt,
+          this.settings.notifyStallCooldownMinutes * 60_000
+        );
+      }
    });
    this.downloadHealthMonitor = new DownloadHealthMonitor(this.storagePaths.notificationHealthFile);
    void this.notificationOutbox.drain().catch((error) => {
@@ -1338,17 +1345,12 @@ export class AppController {
  }

  public async shutdown(): Promise<void> {
+    const deadlineAt = Date.now() + 3000;
    if (this.downloadHealthTimer) {
      clearInterval(this.downloadHealthTimer);
      this.downloadHealthTimer = null;
    }
    this.manager.suspendDownloadHealthMonitoring?.();
-    if (this.downloadHealthEvaluation) {
-      await this.downloadHealthEvaluation;
-    }
-    if (this.downloadHealthMonitor) {
-      await this.evaluateDownloadHealth();
-    }
    if (this.runtimeStatsTimer) {
      clearInterval(this.runtimeStatsTimer);
      this.runtimeStatsTimer = null;
@@ -1356,14 +1358,20 @@ export class AppController {
    stopDebugServer();
    abortActiveUpdateDownload();
    cancelPendingAsyncSaves();
-    const notificationFlush = this.manager.flushNotificationsForShutdown?.();
-    if (notificationFlush) {
-      await notificationFlush;
+    this.manager.prepareForShutdown();
+    if (this.downloadHealthEvaluation) {
+      await this.waitForShutdownTask(this.downloadHealthEvaluation, deadlineAt);
    }
-    await this.notificationOutbox.drainForShutdown(3000).catch((error) => {
+    if (this.downloadHealthMonitor && Date.now() < deadlineAt) {
+      await this.waitForShutdownTask(this.evaluateDownloadHealth(), deadlineAt);
+    }
+    const notificationFlush = this.manager.flushNotificationsForShutdown?.();
+    if (notificationFlush && Date.now() < deadlineAt) {
+      await this.waitForShutdownTask(notificationFlush, deadlineAt);
+    }
+    await this.notificationOutbox.drainForShutdown(Math.max(0, deadlineAt - Date.now())).catch((error) => {
      logger.warn(`Notification-Outbox konnte beim Beenden nicht geleert werden: ${String(error)}`);
    });
-    this.manager.prepareForShutdown();
    this.megaWebFallback.dispose();
    for (const fallback of this.realDebridWebFallbacks.values()) {
      fallback.dispose();
@@ -1381,7 +1389,22 @@ export class AppController {
    if (this.settings.historyRetentionMode === "session") {
      clearHistory(this.storagePaths);
    }
-    logger.info("App beendet");
+    logger.info("App beendet");
+  }
+
+  private async waitForShutdownTask(task: Promise<unknown>, deadlineAt: number): Promise<void> {
+    const remainingMs = Math.max(0, deadlineAt - Date.now());
+    if (remainingMs <= 0) {
+      return;
+    }
+    let timer: NodeJS.Timeout | null = null;
+    const timeout = new Promise<void>((resolve) => {
+      timer = setTimeout(resolve, remainingMs);
+    });
+    await Promise.race([task.then(() => undefined, () => undefined), timeout]);
+    if (timer) {
+      clearTimeout(timer);
+    }
  }

  private getDesktopDirectory(): string | null {
@@ -53,6 +53,7 @@ export interface DownloadHealthState {
  alertedAt: number;
  lastAlertAt: number;
  cooldownUntil: number;
+  lastDeliveredStallEventId: string | null;
  recoverySamples: number;
  lastSampleAt: number | null;
  downloadProgressSequence: number;
@@ -87,6 +88,7 @@ const HEALTH_STATUSES = new Set<DownloadHealthStatus>([
 ]);
 const INCIDENT_TYPES = new Set<DownloadHealthIncidentType>(["scheduler", "no_data"]);
 const FINGERPRINT_PATTERN = /^[a-f0-9]{64}$/;
+const STALL_EVENT_ID_PATTERN = /^health:stall:[a-f0-9]{16}:\d+$/;
 const MIN_SUSPICIOUS_SAMPLES = 3;
 const SCHEDULER_STALE_AFTER_MS = 30_000;
 const ERROR_EVENT_TTL_MS = 24 * 60 * 60 * 1000;
@@ -101,6 +103,10 @@ function validFingerprint(value: unknown): string | null {
  return typeof value === "string" && FINGERPRINT_PATTERN.test(value) ? value : null;
 }

+function validStallEventId(value: unknown): string | null {
+  return typeof value === "string" && STALL_EVENT_ID_PATTERN.test(value) ? value : null;
+}
+
 function durationText(durationMs: number): string {
  const totalSeconds = Math.max(0, Math.floor(durationMs / 1000));
  if (totalSeconds < 60) {
@@ -195,6 +201,7 @@ export function createDownloadHealthState(overrides: Partial<DownloadHealthState
    alertedAt: 0,
    lastAlertAt: 0,
    cooldownUntil: 0,
+    lastDeliveredStallEventId: null,
    recoverySamples: 0,
    lastSampleAt: null,
    downloadProgressSequence: 0,
@@ -217,6 +224,7 @@ function resetForSnapshot(
    queueFingerprint: snapshot.queueFingerprint,
    lastAlertAt: state.lastAlertAt,
    cooldownUntil: state.cooldownUntil,
+    lastDeliveredStallEventId: state.lastDeliveredStallEventId,
    lastSampleAt: now,
    downloadProgressSequence: finiteInteger(snapshot.downloadProgressSequence),
    itemCompletionSequence: finiteInteger(snapshot.itemCompletionSequence),
@@ -229,6 +237,7 @@ function endIncident(state: DownloadHealthState): DownloadHealthState {
  return createDownloadHealthState({
    lastAlertAt: state.lastAlertAt,
    cooldownUntil: state.cooldownUntil,
+    lastDeliveredStallEventId: state.lastDeliveredStallEventId,
    downloadProgressSequence: state.downloadProgressSequence,
    itemCompletionSequence: state.itemCompletionSequence,
    lastPositiveByteAt: state.lastPositiveByteAt,
@@ -378,14 +387,11 @@ export function evaluateDownloadHealth(
  state.lastSampleAt = now;

  const stallAfterMs = Math.max(0, finiteInteger(options.stallAfterMs, 90_000));
-  const cooldownMs = Math.max(0, finiteInteger(options.cooldownMs, 600_000));
  const confirmed = state.suspiciousDurationMs >= stallAfterMs
    && state.suspiciousSamples >= MIN_SUSPICIOUS_SAMPLES;
  if (confirmed && options.notifyOnStall && now >= state.cooldownUntil) {
    state.status = "alerted";
    state.alertedAt = now;
-    state.lastAlertAt = now;
-    state.cooldownUntil = now + cooldownMs;
    state.recoverySamples = 0;
    events.push(incidentEvent(state, snapshot, now));
  }
@@ -406,6 +412,7 @@ function persistedState(state: DownloadHealthState): Omit<DownloadHealthState, "
    alertedAt: finiteInteger(state.alertedAt),
    lastAlertAt: finiteInteger(state.lastAlertAt),
    cooldownUntil: finiteInteger(state.cooldownUntil),
+    lastDeliveredStallEventId: validStallEventId(state.lastDeliveredStallEventId),
    recoverySamples: finiteInteger(state.recoverySamples),
    lastSampleAt: state.lastSampleAt === null ? null : finiteInteger(state.lastSampleAt),
    downloadProgressSequence: finiteInteger(state.downloadProgressSequence),
@@ -455,6 +462,7 @@ export function loadDownloadHealthState(filePath: string): DownloadHealthState {
      alertedAt: finiteInteger(raw.alertedAt),
      lastAlertAt: finiteInteger(raw.lastAlertAt),
      cooldownUntil: finiteInteger(raw.cooldownUntil),
+      lastDeliveredStallEventId: validStallEventId(raw.lastDeliveredStallEventId),
      recoverySamples: finiteInteger(raw.recoverySamples),
      lastSampleAt: null,
      downloadProgressSequence: finiteInteger(raw.downloadProgressSequence),
@@ -472,6 +480,8 @@ export function loadDownloadHealthState(filePath: string): DownloadHealthState {
 export class DownloadHealthMonitor {
  private state: DownloadHealthState;

+  private operationChain: Promise<void> = Promise.resolve();
+
  public constructor(private readonly filePath: string, initialState?: DownloadHealthState) {
    this.state = initialState ? createDownloadHealthState(initialState) : loadDownloadHealthState(filePath);
  }
@@ -480,18 +490,43 @@ export class DownloadHealthMonitor {
    return createDownloadHealthState(this.state);
  }

-  public async sample(
+  public acknowledgeDelivery(event: NotificationEvent, deliveredAt: number, cooldownMs: number): Promise<void> {
+    return this.runExclusive(async () => {
+      const eventId = validStallEventId(event.id);
+      if (event.type !== "download_stalled" || !eventId || this.state.lastDeliveredStallEventId === eventId) {
+        return;
+      }
+      const acknowledgedAt = finiteInteger(deliveredAt);
+      this.state = createDownloadHealthState({
+        ...this.state,
+        lastAlertAt: acknowledgedAt,
+        cooldownUntil: acknowledgedAt + finiteInteger(cooldownMs),
+        lastDeliveredStallEventId: eventId
+      });
+      saveDownloadHealthState(this.filePath, this.state);
+    });
+  }
+
+  public sample(
    snapshot: DownloadHealthSnapshot,
    now: number,
    options: DownloadHealthOptions,
    enqueue: (event: NotificationEvent) => Promise<void>
  ): Promise<DownloadHealthEvaluation> {
-    const evaluation = evaluateDownloadHealth(this.state, snapshot, now, options);
-    for (const event of evaluation.events) {
-      await enqueue(event);
-    }
-    saveDownloadHealthState(this.filePath, evaluation.state);
-    this.state = evaluation.state;
-    return evaluation;
+    return this.runExclusive(async () => {
+      const evaluation = evaluateDownloadHealth(this.state, snapshot, now, options);
+      for (const event of evaluation.events) {
+        await enqueue(event);
+      }
+      saveDownloadHealthState(this.filePath, evaluation.state);
+      this.state = evaluation.state;
+      return evaluation;
+    });
+  }
+
+  private runExclusive<T>(operation: () => Promise<T>): Promise<T> {
+    const result = this.operationChain.then(operation, operation);
+    this.operationChain = result.then(() => undefined, () => undefined);
+    return result;
  }
 }
@@ -1963,6 +1963,8 @@ export class DownloadManager extends EventEmitter {
  private successDigestTimer: NodeJS.Timeout | null = null;

  private notificationEnqueueChain: Promise<void> = Promise.resolve();
+
+  private notificationsShuttingDown = false;

  private itemCount = 0;

@@ -8807,6 +8809,7 @@ export class DownloadManager extends EventEmitter {
  }

  public async flushNotificationsForShutdown(): Promise<void> {
+    this.notificationsShuttingDown = true;
    this.flushPackageSuccessDigest();
    await this.notificationEnqueueChain;
  }
@@ -12151,6 +12154,10 @@ export class DownloadManager extends EventEmitter {
  private queueSuccessfulPackageResult(envelope: PackageResultEnvelope): void {
    const key = this.packageResultKey(envelope.result.packageId, envelope.generation);
    this.successDigestResults.set(key, envelope);
+    if (this.notificationsShuttingDown) {
+      this.flushPackageSuccessDigest();
+      return;
+    }
    if (this.successDigestTimer) {
      return;
    }
@@ -12219,9 +12226,6 @@ export class DownloadManager extends EventEmitter {
    if (!(pkg.downloadEndedAt || 0)) {
      pkg.downloadEndedAt = completedAt;
    }
-    if (!(pkg.postProcessStartedAt || 0) && (pkg.postProcessQueuedAt || 0) > 0) {
-      pkg.postProcessStartedAt = pkg.postProcessQueuedAt;
-    }
    pkg.postProcessCompletedAt = completedAt;
    pkg.terminalAt = completedAt;
    const result = finalizePackageResult({
@@ -12236,6 +12240,7 @@ export class DownloadManager extends EventEmitter {
    pkg.status = result.status === "partial" ? "failed" : result.status;
    pkg.updatedAt = completedAt;
    if (this.onHistoryEntryCallback) {
+      this.historyRecordedPackages.add(packageId);
      this.onHistoryEntryCallback(buildHistoryEntry(result, {
        generation,
        outputDir: pkg.outputDir,
@@ -166,7 +166,7 @@ function event(
    type,
    priority,
    createdAt,
-    expiresAt: createdAt + (priority === "success" ? SUCCESS_TTL_MS : IMPORTANT_TTL_MS),
+    expiresAt: createdAt + (type === "run_completed" || priority === "error" ? IMPORTANT_TTL_MS : SUCCESS_TTL_MS),
    attempts: 0,
    nextAttemptAt: createdAt,
    payload: { title, description, color, fields }
@@ -44,6 +44,7 @@ export interface NotificationOutboxStatus {
 export interface NotificationOutboxOptions {
  filePath: string;
  send: (event: NotificationEvent) => Promise<boolean>;
+  onDelivered?: (event: NotificationEvent, deliveredAt: number) => void | Promise<void>;
  now?: () => number;
  autoDrain?: boolean;
 }
@@ -166,14 +167,17 @@ export class NotificationOutbox {
  private operationChain: Promise<void> = Promise.resolve();
  private readonly filePath: string;
  private readonly sendEvent: (event: NotificationEvent) => Promise<boolean>;
+  private readonly onDelivered: ((event: NotificationEvent, deliveredAt: number) => void | Promise<void>) | null;
  private readonly clock: () => number;
  private readonly autoDrain: boolean;
  private retryTimer: NodeJS.Timeout | null = null;
  private shutdownRequested = false;
+  private drainOperation: Promise<void> | null = null;

  public constructor(options: NotificationOutboxOptions) {
    this.filePath = options.filePath;
    this.sendEvent = options.send;
+    this.onDelivered = options.onDelivered || null;
    this.clock = options.now || Date.now;
    this.autoDrain = Boolean(options.autoDrain);
    this.load();
@@ -196,45 +200,16 @@ export class NotificationOutbox {
  }

  public drain(now?: number): Promise<void> {
-    return this.runExclusive(async () => {
-      let currentNow = finiteInteger(now ?? this.clock());
-      this.enforceLimits(currentNow);
-      while (this.events.length > 0) {
-        const current = this.events[0];
-        if (current.nextAttemptAt > currentNow) {
-          await this.persist(currentNow);
-          if (this.autoDrain) {
-            this.scheduleDrain(Math.max(0, current.nextAttemptAt - this.clock()));
-          }
-          break;
-        }
-        let sent = false;
-        try {
-          sent = await this.sendEvent(current);
-        } catch {
-          sent = false;
-        }
-        const outcomeAt = finiteInteger(this.clock(), currentNow);
-        if (!sent) {
-          current.attempts += 1;
-          current.nextAttemptAt = outcomeAt + retryDelayMs(current.attempts);
-          this.lastFailureAt = outcomeAt;
-          await this.persist(outcomeAt);
-          if (this.autoDrain) {
-            this.scheduleDrain(Math.max(0, current.nextAttemptAt - this.clock()));
-          }
-          break;
-        }
-        this.events.shift();
-        this.lastSuccessAt = outcomeAt;
-        await this.persist(outcomeAt);
-        currentNow = finiteInteger(this.clock(), outcomeAt);
-      }
-      if (this.events.length === 0) {
-        this.clearRetryTimer();
-        await this.persist(finiteInteger(this.clock(), currentNow));
+    if (this.drainOperation) {
+      return this.drainOperation;
+    }
+    const operation = this.performDrain(now).finally(() => {
+      if (this.drainOperation === operation) {
+        this.drainOperation = null;
      }
    });
+    this.drainOperation = operation;
+    return operation;
  }

  public async drainForShutdown(timeoutMs = DEFAULT_SHUTDOWN_TIMEOUT_MS): Promise<void> {
@@ -258,12 +233,76 @@ export class NotificationOutbox {
    };
  }

-  private runExclusive(operation: () => Promise<void>): Promise<void> {
+  private runExclusive<T>(operation: () => Promise<T>): Promise<T> {
    const result = this.operationChain.then(operation, operation);
-    this.operationChain = result.catch(() => {});
+    this.operationChain = result.then(() => undefined, () => undefined);
    return result;
  }

+  private async performDrain(now?: number): Promise<void> {
+    let currentNow = finiteInteger(now ?? this.clock());
+    while (true) {
+      const current = await this.runExclusive(async () => {
+        this.enforceLimits(currentNow);
+        const next = this.events[0] || null;
+        if (!next) {
+          this.clearRetryTimer();
+          await this.persist(finiteInteger(this.clock(), currentNow));
+          return null;
+        }
+        if (next.nextAttemptAt > currentNow) {
+          await this.persist(currentNow);
+          if (this.autoDrain) {
+            this.scheduleDrain(Math.max(0, next.nextAttemptAt - this.clock()));
+          }
+          return null;
+        }
+        return next;
+      });
+      if (!current) {
+        return;
+      }
+      let sent = false;
+      try {
+        sent = await this.sendEvent(current);
+      } catch {
+        sent = false;
+      }
+      const outcomeAt = finiteInteger(this.clock(), currentNow);
+      const delivered = await this.runExclusive(async () => {
+        const index = this.events.findIndex((event) => event.id === current.id);
+        if (index < 0) {
+          return false;
+        }
+        if (!sent) {
+          const queued = this.events[index];
+          queued.attempts += 1;
+          queued.nextAttemptAt = outcomeAt + retryDelayMs(queued.attempts);
+          this.lastFailureAt = outcomeAt;
+          await this.persist(outcomeAt);
+          if (this.autoDrain) {
+            this.scheduleDrain(Math.max(0, queued.nextAttemptAt - this.clock()));
+          }
+          return false;
+        }
+        this.events.splice(index, 1);
+        this.lastSuccessAt = outcomeAt;
+        await this.persist(outcomeAt);
+        return true;
+      });
+      if (!sent) {
+        return;
+      }
+      if (delivered && this.onDelivered) {
+        try {
+          await this.onDelivered(current, outcomeAt);
+        } catch {
+        }
+      }
+      currentNow = finiteInteger(this.clock(), outcomeAt);
+    }
+  }
+
  private scheduleDrain(delayMs: number): void {
    if (this.shutdownRequested) {
      return;
@@ -284,10 +323,10 @@ export class NotificationOutbox {
  }

  private load(): void {
+    if (!fs.existsSync(this.filePath)) {
+      return;
+    }
    try {
-      if (!fs.existsSync(this.filePath)) {
-        return;
-      }
      const parsed = JSON.parse(fs.readFileSync(this.filePath, "utf8")) as Partial<PersistedNotificationOutbox>;
      this.events = Array.isArray(parsed.events)
        ? parsed.events.flatMap((event) => {
@@ -303,6 +342,7 @@ export class NotificationOutbox {
      this.lastSuccessAt = 0;
      this.lastFailureAt = 0;
    }
+    this.persistSync(this.clock());
  }

  private enforceLimits(now: number): void {
@@ -332,4 +372,26 @@ export class NotificationOutbox {
      throw error;
    }
  }
+
+  private persistSync(now: number): void {
+    this.enforceLimits(now);
+    fs.mkdirSync(path.dirname(this.filePath), { recursive: true });
+    const tempPath = `${this.filePath}.tmp`;
+    const state: PersistedNotificationOutbox = {
+      version: 1,
+      events: this.events,
+      lastSuccessAt: this.lastSuccessAt,
+      lastFailureAt: this.lastFailureAt
+    };
+    try {
+      fs.writeFileSync(tempPath, JSON.stringify(state), "utf8");
+      fs.renameSync(tempPath, this.filePath);
+    } catch (error) {
+      try {
+        fs.rmSync(tempPath, { force: true });
+      } catch {
+      }
+      throw error;
+    }
+  }
 }