FN-7018: retry transient ntfy notification failures

Ntfy sends now retry bounded transient failures before giving up.

- Add per-attempt timeouts and bounded retries for ntfy network, timeout, 5xx, and 429 failures.\n- Preserve best-effort abort behavior while retrying only retryable publish failures.\n- Cover retry, timeout, abort, payload, and non-retryable failure behavior in notifier tests.\n- Add a patch changeset for the published Fusion package.\n\nFiles changed:\n .changeset/fn-7018-ntfy-retry.md               |   7 +\n packages/engine/src/__tests__/notifier.test.ts | 259 ++++++++++++++++++++++++-\n packages/engine/src/notifier.ts                | 173 +++++++++++++----\n 3 files changed, 399 insertions(+), 40 deletions(-)

Fusion-Task-Id: FN-7018

Fusion-Task-Lineage: 0d0b7313-532a-4583-8bb7-67fc4b2e98bf

.changeset/fn-7018-ntfy-retry.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Retry transient ntfy publish failures so one-shot task notifications are less likely to be lost.
+category: fix
+dev: Adds bounded ntfy fetch retries for network, timeout, 5xx, and 429 failures with a per-attempt timeout.

packages/engine/src/__tests__/notifier.test.ts

@@ -133,6 +133,7 @@ describe("sendNtfyNotificationWithResult", () => {
   let fetchMock: ReturnType<typeof vi.fn>;
 
   beforeEach(() => {
+    vi.useRealTimers();
     fetchMock = vi.fn(async (_input: string | URL, init?: RequestInit) => {
       new Headers(init?.headers);
       return { ok: true, status: 200, statusText: "OK" } as Response;
@@ -140,6 +141,10 @@ describe("sendNtfyNotificationWithResult", () => {
     global.fetch = fetchMock;
   });
 
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
   it("calling with a unicode title does not throw and posts JSON with auth preserved", async () => {
     const signal = new AbortController().signal;
 
@@ -157,7 +162,7 @@ describe("sendNtfyNotificationWithResult", () => {
       "https://ntfy.sh/",
       expect.objectContaining({
         method: "POST",
-        signal,
+        signal: expect.any(AbortSignal),
         headers: expect.objectContaining({
           "Content-Type": "application/json",
           Authorization: "Bearer secret-token",
@@ -257,7 +262,7 @@ describe("sendNtfyNotificationWithResult", () => {
       "https://ntfy.sh/ascii-topic",
       expect.objectContaining({
         method: "POST",
-        signal,
+        signal: expect.any(AbortSignal),
         headers: expect.objectContaining({
           "Content-Type": "text/plain",
           Title: "Task FN-1 merged",
@@ -269,6 +274,256 @@ describe("sendNtfyNotificationWithResult", () => {
     );
   });
 
+  it("does not retry when the first ntfy publish succeeds", async () => {
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "single-success",
+      title: "Task FN-1 merged",
+      message: "Task merged",
+      retryDelayMs: 0,
+    });
+
+    expect(result).toEqual({ ok: true, status: 200, statusText: "OK" });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("retries retryable network failures and returns the recovered success", async () => {
+    fetchMock
+      .mockRejectedValueOnce(new Error("ECONNRESET"))
+      .mockResolvedValueOnce({ ok: true, status: 200, statusText: "OK" } as Response);
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      ntfyAccessToken: "secret-token",
+      topic: "retry-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      retryDelayMs: 0,
+    });
+
+    expect(result).toEqual({ ok: true, status: 200, statusText: "OK" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    for (const call of fetchMock.mock.calls) {
+      const request = call[1] as RequestInit;
+      const headers = new Headers(request.headers as HeadersInit);
+      expect(headers.get("Authorization")).toBe("Bearer secret-token");
+      expect(call[0]).toBe("https://ntfy.sh/retry-topic");
+    }
+  });
+
+  it.each([
+    { status: 503, statusText: "Service Unavailable" },
+    { status: 429, statusText: "Too Many Requests" },
+  ])("retries retryable HTTP status $status", async ({ status, statusText }) => {
+    fetchMock
+      .mockResolvedValueOnce({ ok: false, status, statusText } as Response)
+      .mockResolvedValueOnce({ ok: true, status: 200, statusText: "OK" } as Response);
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "http-retry-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      retryDelayMs: 0,
+    });
+
+    expect(result).toEqual({ ok: true, status: 200, statusText: "OK" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it.each([
+    { status: 400, statusText: "Bad Request" },
+    { status: 401, statusText: "Unauthorized" },
+    { status: 403, statusText: "Forbidden" },
+    { status: 404, statusText: "Not Found" },
+  ])("does not retry non-429 client status $status", async ({ status, statusText }) => {
+    fetchMock.mockResolvedValueOnce({ ok: false, status, statusText } as Response);
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "client-error-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      retryDelayMs: 0,
+    });
+
+    expect(result).toEqual({ ok: false, status, statusText });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("returns the last HTTP failure after exhausting retryable responses", async () => {
+    fetchMock
+      .mockResolvedValueOnce({ ok: false, status: 503, statusText: "first" } as Response)
+      .mockResolvedValueOnce({ ok: false, status: 503, statusText: "second" } as Response);
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "exhaust-http-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      retryDelayMs: 0,
+      maxAttempts: 2,
+    });
+
+    expect(result).toEqual({ ok: false, status: 503, statusText: "second" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("returns null without throwing after exhausting network retries", async () => {
+    fetchMock.mockRejectedValue(new Error("DNS failure"));
+
+    await expect(sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "network-fail-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      retryDelayMs: 0,
+      maxAttempts: 2,
+    })).resolves.toBeNull();
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("aborts a hung attempt with the per-attempt timeout and retries", async () => {
+    vi.useFakeTimers();
+    const abortError = Object.assign(new Error("aborted"), { name: "AbortError" });
+    fetchMock
+      .mockImplementationOnce((_input: string | URL, init?: RequestInit) => new Promise((_resolve, reject) => {
+        init?.signal?.addEventListener("abort", () => reject(abortError), { once: true });
+      }))
+      .mockResolvedValueOnce({ ok: true, status: 200, statusText: "OK" } as Response);
+
+    const resultPromise = sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "timeout-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      attemptTimeoutMs: 5,
+      retryDelayMs: 0,
+      maxAttempts: 2,
+    });
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(resultPromise).resolves.toEqual({ ok: true, status: 200, statusText: "OK" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    vi.useRealTimers();
+  });
+
+  it("returns null after every attempt times out", async () => {
+    vi.useFakeTimers();
+    const abortError = Object.assign(new Error("aborted"), { name: "AbortError" });
+    fetchMock.mockImplementation((_input: string | URL, init?: RequestInit) => new Promise((_resolve, reject) => {
+      init?.signal?.addEventListener("abort", () => reject(abortError), { once: true });
+    }));
+
+    const resultPromise = sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "timeout-null-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      attemptTimeoutMs: 5,
+      retryDelayMs: 0,
+      maxAttempts: 2,
+    });
+
+    await vi.advanceTimersByTimeAsync(5);
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
+    await vi.advanceTimersByTimeAsync(5);
+    await expect(resultPromise).resolves.toBeNull();
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    vi.useRealTimers();
+  });
+
+  it("short-circuits without fetch when the caller signal is already aborted", async () => {
+    const controller = new AbortController();
+    controller.abort();
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "aborted-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      signal: controller.signal,
+      retryDelayMs: 0,
+    });
+
+    expect(result).toBeNull();
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it("does not retry after a caller aborts mid-flight", async () => {
+    const controller = new AbortController();
+    const abortError = Object.assign(new Error("aborted"), { name: "AbortError" });
+    fetchMock.mockImplementationOnce((_input: string | URL, init?: RequestInit) => new Promise((_resolve, reject) => {
+      init?.signal?.addEventListener("abort", () => reject(abortError), { once: true });
+    }));
+
+    const resultPromise = sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "caller-abort-topic",
+      title: "ASCII title",
+      message: "ASCII message",
+      signal: controller.signal,
+      retryDelayMs: 0,
+    });
+
+    controller.abort();
+
+    await expect(resultPromise).resolves.toBeNull();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("posts unicode retries through the JSON publish path", async () => {
+    fetchMock
+      .mockResolvedValueOnce({ ok: false, status: 503, statusText: "Service Unavailable" } as Response)
+      .mockResolvedValueOnce({ ok: true, status: 200, statusText: "OK" } as Response);
+
+    const result = await sendNtfyNotificationWithResult({
+      ntfyBaseUrl: "https://ntfy.sh",
+      topic: "unicode-topic",
+      title: "Triage Bot → Executor Bot",
+      message: "Triage Bot → you: preview text",
+      priority: "high",
+      retryDelayMs: 0,
+    });
+
+    expect(result).toEqual({ ok: true, status: 200, statusText: "OK" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    for (const call of fetchMock.mock.calls) {
+      expect(call[0]).toBe("https://ntfy.sh/");
+      const request = call[1] as RequestInit;
+      const payload = JSON.parse(String(request.body)) as { topic: string; priority: number; message: string };
+      expect(new Headers(request.headers as HeadersInit).get("Content-Type")).toBe("application/json");
+      expect(payload).toEqual(expect.objectContaining({
+        topic: "unicode-topic",
+        priority: 4,
+        message: "Triage Bot → you: preview text",
+      }));
+    }
+  });
+
+  it("lets the provider report success when the primitive retry recovers", async () => {
+    fetchMock
+      .mockRejectedValueOnce(new Error("ECONNRESET"))
+      .mockResolvedValueOnce({ ok: true, status: 200, statusText: "OK" } as Response);
+    const provider = new NtfyNotificationProvider();
+    await provider.initialize({
+      topic: "provider-retry-topic",
+      ntfyBaseUrl: "https://ntfy.sh",
+      events: ["in-review"],
+    });
+
+    const result = await provider.sendNotification("in-review", {
+      taskId: "FN-1",
+      taskTitle: "Retry me",
+      event: "in-review",
+    });
+
+    expect(result).toEqual({ success: true, providerId: "ntfy" });
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    await provider.shutdown();
+  });
+
   it("truncates overlong titles to 250 characters with an ellipsis", async () => {
     await sendNtfyNotificationWithResult({
       ntfyBaseUrl: "https://ntfy.sh",

packages/engine/src/notifier.ts

@@ -19,6 +19,9 @@ const GRIDLOCK_NOTIFICATION_COOLDOWN_MS = 15 * 60 * 1000;
 const NTFY_TITLE_MAX = 250;
 // ntfy documents a 4 KiB message body limit; reserve room for an ellipsis when truncating UTF-8 payloads.
 const NTFY_MESSAGE_MAX = 4096;
+const DEFAULT_NTFY_MAX_ATTEMPTS = 3;
+const DEFAULT_NTFY_ATTEMPT_TIMEOUT_MS = 10_000;
+const DEFAULT_NTFY_RETRY_DELAY_MS = 500;
 
 export const DEFAULT_NTFY_EVENTS: readonly NtfyNotificationEvent[] = [
   "in-review",
@@ -57,6 +60,12 @@ export interface SendNtfyNotificationInput {
   priority?: NtfyNotificationPriority;
   clickUrl?: string;
   signal?: AbortSignal;
+  /** @internal Test seam for exercising timeout behavior without slow wall-clock waits. */
+  attemptTimeoutMs?: number;
+  /** @internal Test seam for exercising retry behavior without slow wall-clock waits. */
+  retryDelayMs?: number;
+  /** @internal Test seam for keeping retry-bound assertions narrow. */
+  maxAttempts?: number;
 }
 
 interface NtfyConfig {
@@ -156,6 +165,40 @@ function ntfyPriorityToInt(priority: NtfyNotificationPriority): number {
   }
 }
 
+function isRetryableNtfyStatus(status: number): boolean {
+  return status === 429 || status >= 500;
+}
+
+function isAbortError(error: unknown): boolean {
+  return error instanceof Error && error.name === "AbortError";
+}
+
+function sleep(ms: number, signal?: AbortSignal): Promise<"slept" | "aborted"> {
+  if (signal?.aborted) {
+    return Promise.resolve("aborted");
+  }
+  if (ms <= 0) {
+    return Promise.resolve("slept");
+  }
+
+  return new Promise((resolve) => {
+    function cleanup() {
+      clearTimeout(timer);
+      signal?.removeEventListener("abort", onAbort);
+    }
+    const onAbort = () => {
+      cleanup();
+      resolve("aborted");
+    };
+
+    const timer = setTimeout(() => {
+      cleanup();
+      resolve("slept");
+    }, ms);
+    signal?.addEventListener("abort", onAbort, { once: true });
+  });
+}
+
 export function resolveNtfyEvents(events?: NtfyNotificationEvent[]): NtfyNotificationEvent[] {
   return events && events.length > 0 ? [...events] : [...DEFAULT_NTFY_EVENTS];
 }
@@ -216,61 +259,115 @@ export async function sendNtfyNotificationWithResult({
   priority = "default",
   clickUrl,
   signal,
+  attemptTimeoutMs = DEFAULT_NTFY_ATTEMPT_TIMEOUT_MS,
+  retryDelayMs = DEFAULT_NTFY_RETRY_DELAY_MS,
+  maxAttempts = DEFAULT_NTFY_MAX_ATTEMPTS,
 }: SendNtfyNotificationInput): Promise<{ ok: boolean; status: number; statusText: string } | null> {
+  const resolvedMaxAttempts = Math.max(1, Math.floor(maxAttempts));
+
   try {
     const resolvedBaseUrl = resolveNtfyBaseUrl(ntfyBaseUrl);
     const trimmedToken = ntfyAccessToken?.trim();
     const truncatedTitle = truncateNtfyTitle(title);
     const truncatedMessage = truncateNtfyMessage(message);
     const latin1Safe = isLatin1Safe(truncatedTitle) && isLatin1Safe(truncatedMessage);
+    const url = latin1Safe ? `${resolvedBaseUrl}/${topic}` : `${resolvedBaseUrl}/`;
+    const body = latin1Safe
+      ? truncatedMessage
+      : JSON.stringify({
+        topic,
+        title: truncatedTitle,
+        message: truncatedMessage,
+        priority: ntfyPriorityToInt(priority),
+        ...(clickUrl ? { click: clickUrl } : {}),
+      });
 
-    const headers: Record<string, string> = {
+    /*
-      "Content-Type": latin1Safe ? "text/plain" : "application/json",
+    FNXC:Notifications 2026-06-25-18:25:
-    };
+    Task notifications are one-shot and a single transient ntfy network failure, timeout, 5xx, or 429 can permanently lose the user-facing event.
+    Keep ntfy best-effort and never-throwing, but bound each attempt with an internal timeout and retry only retryable failures while honoring caller lifecycle aborts immediately.
+    */
+    for (let attempt = 1; attempt <= resolvedMaxAttempts; attempt += 1) {
+      if (signal?.aborted) {
+        return null;
+      }
 
-    if (latin1Safe) {
+      const attemptController = new AbortController();
-      headers.Priority = priority;
+      let timedOut = false;
-      headers.Title = truncatedTitle;
+      let timeout: ReturnType<typeof setTimeout> | undefined;
-      if (clickUrl) {
+      const onCallerAbort = () => attemptController.abort();
-        headers.Click = clickUrl;
+      signal?.addEventListener("abort", onCallerAbort, { once: true });
+      if (attemptTimeoutMs > 0) {
+        timeout = setTimeout(() => {
+          timedOut = true;
+          attemptController.abort();
+        }, attemptTimeoutMs);
+      }
+
+      try {
+        const headers: Record<string, string> = {
+          "Content-Type": latin1Safe ? "text/plain" : "application/json",
+        };
+
+        if (latin1Safe) {
+          headers.Priority = priority;
+          headers.Title = truncatedTitle;
+          if (clickUrl) {
+            headers.Click = clickUrl;
+          }
+        }
+
+        if (trimmedToken) {
+          headers.Authorization = `Bearer ${trimmedToken}`;
+        }
+
+        const response = await fetch(url, {
+          method: "POST",
+          headers,
+          body,
+          signal: attemptController.signal,
+        });
+
+        if (!response.ok) {
+          schedulerLog.log(`Ntfy notification failed: ${response.status} ${response.statusText}`);
+        }
+
+        const result = {
+          ok: response.ok,
+          status: response.status,
+          statusText: response.statusText,
+        };
+        if (response.ok || !isRetryableNtfyStatus(response.status) || attempt === resolvedMaxAttempts) {
+          return result;
+        }
+      } catch (err) {
+        if (signal?.aborted || (isAbortError(err) && !timedOut)) {
+          return null;
+        }
+        if (attempt === resolvedMaxAttempts) {
+          schedulerLog.log(`Failed to send ntfy notification: ${err}`);
+          return null;
+        }
+      } finally {
+        if (timeout) {
+          clearTimeout(timeout);
+        }
+        signal?.removeEventListener("abort", onCallerAbort);
+      }
+
+      const slept = await sleep(retryDelayMs, signal);
+      if (slept === "aborted") {
+        return null;
       }
     }
-
-    if (trimmedToken) {
-      headers.Authorization = `Bearer ${trimmedToken}`;
-    }
-
-    const response = await fetch(latin1Safe ? `${resolvedBaseUrl}/${topic}` : `${resolvedBaseUrl}/`, {
-      method: "POST",
-      headers,
-      body: latin1Safe
-        ? truncatedMessage
-        : JSON.stringify({
-          topic,
-          title: truncatedTitle,
-          message: truncatedMessage,
-          priority: ntfyPriorityToInt(priority),
-          ...(clickUrl ? { click: clickUrl } : {}),
-        }),
-      signal,
-    });
-
-    if (!response.ok) {
-      schedulerLog.log(`Ntfy notification failed: ${response.status} ${response.statusText}`);
-    }
-
-    return {
-      ok: response.ok,
-      status: response.status,
-      statusText: response.statusText,
-    };
   } catch (err) {
-    if (err instanceof Error && err.name === "AbortError") {
+    if (isAbortError(err) || signal?.aborted) {
       return null;
     }
     schedulerLog.log(`Failed to send ntfy notification: ${err}`);
-    return null;
   }
+
+  return null;
 }
 
 export async function sendNtfyNotification(input: SendNtfyNotificationInput): Promise<void> {