diff --git a/.changeset/fn-7018-ntfy-retry.md b/.changeset/fn-7018-ntfy-retry.md new file mode 100644 index 0000000000..f891e586f6 --- /dev/null +++ b/.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. diff --git a/packages/engine/src/__tests__/notifier.test.ts b/packages/engine/src/__tests__/notifier.test.ts index a886482483..7d955f6c16 100644 --- a/packages/engine/src/__tests__/notifier.test.ts +++ b/packages/engine/src/__tests__/notifier.test.ts @@ -133,6 +133,7 @@ describe("sendNtfyNotificationWithResult", () => { let fetchMock: ReturnType; 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", diff --git a/packages/engine/src/notifier.ts b/packages/engine/src/notifier.ts index 0827380475..7409177603 100644 --- a/packages/engine/src/notifier.ts +++ b/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 = { - "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) { - headers.Priority = priority; - headers.Title = truncatedTitle; - if (clickUrl) { - headers.Click = clickUrl; + const attemptController = new AbortController(); + let timedOut = false; + let timeout: ReturnType | undefined; + const onCallerAbort = () => attemptController.abort(); + signal?.addEventListener("abort", onCallerAbort, { once: true }); + if (attemptTimeoutMs > 0) { + timeout = setTimeout(() => { + timedOut = true; + attemptController.abort(); + }, attemptTimeoutMs); + } + + try { + const headers: Record = { + "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 {