diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 4caac949..24a0e96a 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -2,6 +2,11 @@ ### New +- Add opt-in `callHttp({ retryOptions: new HttpRetryOptions(...) })` failure retries using the core + durable activity retry engine, configurable status codes, backoff, attempt limits, and timeout. + Defaults match .NET in-process HTTP retries (non-2xx statuses, six-day maximum interval); + exhaustion throws `TaskFailedError`. Like .NET, `202` Location polls do not inherit the retry + policy. Calls without retry options are unchanged. See README for replay rollout and idempotency guidance. - Expose readonly `context.df.name`, forwarding the core logical orchestration name from history and preserving aliases and case across replay. This does not add `InvocationContext.functionName`. - Add optional orchestration version migration support to `context.df.continueAsNew()`. diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index e4450768..03dd6bdf 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -71,7 +71,8 @@ changed: ([#318](https://github.com/microsoft/durabletask-js/issues/318)) — though **not** as a drop-in, fully v3-equivalent replacement: the known incompatibilities and behavior differences listed below are load-bearing for migration, so review them before relying on it. It accepts the v3 - `CallHttpOptions` (`method`, `url`, `body`, `headers`, `tokenSource`, `enablePolling`) and returns a + `CallHttpOptions` (`method`, `url`, `body`, `headers`, `tokenSource`, `enablePolling`), adds opt-in + [`retryOptions`](#durable-http-failure-retries), and returns a `Task` (`{ statusCode, headers, content }`), including automatic `202 Accepted` polling that honors `Retry-After` via durable timers. **Trust-boundary change:** in v3 the Functions **host** extension executed the HTTP request; here it runs as a durable **activity inside your @@ -130,6 +131,69 @@ changed: sync **generators** (`function*`) using `context.df.*` — are unaffected; convert any non-generator classic orchestrator to generator form, or to the core-native `ctx.*` API. +## Durable HTTP failure retries + +Use `HttpRetryOptions` to retry failures of the **initial** `callHttp` request. It extends the existing +`RetryOptions` API; all intervals are milliseconds, and the attempt count includes the first request. + +```typescript +import * as df from "durable-functions"; +import type { Task } from "@microsoft/durabletask-js"; + +df.app.orchestration("FetchReport", function* (context: df.OrchestrationContext): Generator< + Task, + unknown, + unknown +> { + const retryOptions = new df.HttpRetryOptions(1000, 4); + retryOptions.backoffCoefficient = 2; + retryOptions.maxRetryIntervalInMilliseconds = 10000; + retryOptions.retryTimeoutInMilliseconds = 60000; + retryOptions.statusCodesToRetry = [429, 500, 503]; + + return yield context.df.callHttp({ + method: "GET", + url: context.df.getInput(), + retryOptions, + }); +}); +``` + +The policy and a copy of the status list are recorded with the request. Retries reuse the core +activity retry engine and durable timers, so replay does not reissue completed network requests. +Selected failure responses have their unused body canceled before buffering, so even a body that +never ends does not prevent a durable retry. Returned responses still include their complete body. +Exhaustion throws `TaskFailedError` into the caller (or fails the orchestration if uncaught); it does +**not** return the last failed HTTP response. An excluded status is returned normally. Transport and +other activity failures use the same retry budget regardless of the status list. Retrying a request +can repeat side effects, including when a server processed it but the connection failed before the +response arrived. Enable retries only for operations that are safe to repeat. + +Defaults follow the .NET **in-process Durable Functions extension**, not the generic .NET Durable +Task SDK: backoff `1`, maximum interval **six days**, and no overall retry timeout. An omitted or empty +status list retries non-2xx responses, including 4xx/5xx and any unfollowed 3xx, matching +[`TaskHttpActivityShim`'s `EnsureSuccessStatusCode`](https://github.com/Azure/azure-functions-durable-extension/blob/4437edf785533ccdf219c15f496777ba1c30d41d/src/WebJobs.Extensions.DurableTask/Listener/TaskHttpActivityShim.cs#L63-L88). +An explicit list is exact, even for success codes: listing `202` retries the initial request instead +of following its Location. Existing `fetch` redirect behavior is unchanged. + +**202 polling is separate:** like the extension's +[`CreateLocationPollRequest`](https://github.com/Azure/azure-functions-durable-extension/blob/4437edf785533ccdf219c15f496777ba1c30d41d/src/WebJobs.Extensions.DurableTask/ContextImplementations/DurableOrchestrationContext.cs#L283-L379), +Location polls **do not inherit the failure retry policy**. For example, initial `503 -> 202` can retry, +then a poll returning `503` ends the call with that response, without retrying the poll. The existing +polling `Retry-After` delays and credential-stripping rules remain unchanged. Failure retry delays use +the configured policy, not `Retry-After`. `enablePolling: false` disables only polling, not retries. + +JavaScript-specific behavior: options use millisecond numbers rather than .NET `TimeSpan`; `-1` means +unlimited maximum interval or retry timeout. Scalar validation and timeout boundaries reuse the core +`RetryPolicy`: no retry is scheduled if its delay exceeds the remaining timeout, equality is allowed, +and the timeout does not cancel an in-flight HTTP request or limit the subsequent polling loop. +Long retry delays use the provider's existing three-day durable timer segments. Failures surface as +JavaScript `TaskFailedError`, not .NET exception types. + +Omitting `retryOptions` preserves existing response, failure, and polling behavior. Deploy the updated +worker before enabling the option; do not add it to existing in-flight orchestration code without +versioning or draining those instances, because it changes the durable action history. + ## Requirements This provider reaches the Durable Task backend over the Functions host's **gRPC** channel, which diff --git a/packages/azure-functions-durable/src/http/builtin.ts b/packages/azure-functions-durable/src/http/builtin.ts index ffb8b674..970d1a9e 100644 --- a/packages/azure-functions-durable/src/http/builtin.ts +++ b/packages/azure-functions-durable/src/http/builtin.ts @@ -33,7 +33,7 @@ * durabletask-python design (Andy Staples, durabletask-python#155). */ -import { OrchestrationContext, Task } from "@microsoft/durabletask-js"; +import { OrchestrationContext, RetryPolicy, Task } from "@microsoft/durabletask-js"; import { DurableHttpRequestPayload, DurableHttpResponse } from "./models"; /** @@ -234,10 +234,10 @@ async function acquireBearerToken(resource: string): Promise { * * @remarks * `input` is the JSON form of a durable HTTP request (`method`, `uri`, `content`, `headers`, - * `tokenSource`). Non-2xx responses (including `202`) are captured rather than thrown — the global - * `fetch` only rejects on network errors, not on HTTP status — so the poll orchestrator can inspect - * the status code and headers. Only http/https URIs are permitted (an SSRF guard that closes off - * `file://`, `ftp://`, ... schemes from orchestration-supplied URLs). + * `tokenSource`, `retryOptions`). Responses are returned unless an opt-in retry policy selects the + * status as a failure. Activity failures are retried by the orchestrator's durable retry policy, + * never by an in-process retry loop. Only http/https URIs are permitted (an SSRF guard that closes + * off `file://`, `ftp://`, ... schemes from orchestration-supplied URLs). * * When a `tokenSource` is present, the acquired bearer token **overwrites** any caller-supplied * `Authorization` header (matching v3, which applies the caller's headers first and then the token), @@ -299,6 +299,18 @@ export async function builtinHttpActivity(input: DurableHttpRequestPayload): Pro body: includeBody ? request.content : undefined, }); + if (request.retryOptions !== undefined) { + const codes = request.retryOptions.statusCodesToRetry; + const shouldRetry = codes?.length + ? codes.includes(response.status) + : response.status < 200 || response.status >= 300; + if (shouldRetry) { + // The discarded body may never end. Cancel it before buffering so durable retries can proceed. + await response.body?.cancel(); + throw new Error(`HTTP request failed with status code ${response.status}.`); + } + } + const responseHeaders: { [key: string]: string } = {}; response.headers.forEach((value, key) => { responseHeaders[key] = value; @@ -345,6 +357,7 @@ function buildPollRequest( enablePolling: boolean, ): DurableHttpRequestPayload { const sameOrigin = isSameOrigin(trustAnchorUri, resolved); + // Like .NET CreateLocationPollRequest, Location polls do not inherit the initial retry policy. const pollRequest: DurableHttpRequestPayload = { method: "GET", uri: resolved, enablePolling }; if (request.headers !== undefined) { @@ -390,10 +403,14 @@ export async function* builtinHttpPollOrchestrator( } const request = input ?? ({} as DurableHttpRequestPayload); - // v3 opt-out: when polling is disabled the first response is returned as-is (no 202 loop). + // v3 opt-out: skip the 202 loop when polling is disabled; failure retries are independent. const enablePolling = request.enablePolling !== false; - let response = (yield ctx.callActivity(BUILTIN_HTTP_ACTIVITY_NAME, request)) as BuiltinHttpActivityResult; + const initialTask = + request.retryOptions === undefined + ? ctx.callActivity(BUILTIN_HTTP_ACTIVITY_NAME, request) + : ctx.callActivity(BUILTIN_HTTP_ACTIVITY_NAME, request, { retry: new RetryPolicy(request.retryOptions) }); + let response = (yield initialTask) as BuiltinHttpActivityResult; // Two DISTINCT URIs, deliberately NOT merged into one variable: // - `originalUri` is the credential TRUST ANCHOR: the URI the app author declared. It is captured // once and NEVER reassigned, so a callee-controlled `Location` (or a redirect `fetch` followed) diff --git a/packages/azure-functions-durable/src/http/models.ts b/packages/azure-functions-durable/src/http/models.ts index 82a99179..8e56a10b 100644 --- a/packages/azure-functions-durable/src/http/models.ts +++ b/packages/azure-functions-durable/src/http/models.ts @@ -1,6 +1,8 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. +import { RetryOptions } from "../retry-options"; + /** * Durable HTTP request/response models that **mirror** the durable-functions v3 shapes (they are not * the v3 classes themselves). @@ -43,8 +45,24 @@ export class ManagedIdentityTokenSource { } /** - * Options accepted by `context.df.callHttp` (classic durable-functions v3 shape). + * Opt-in failure retries for the initial `callHttp` request (not subsequent `202` Location polls). + * Uses durable activity retries; exhausting the policy fails `callHttp` with `TaskFailedError`. + * Only enable retries when repeating the request's side effects is safe. */ +export class HttpRetryOptions extends RetryOptions { + /** Maximum delay between retries, in milliseconds. Defaults to six days, matching .NET HTTP retries. */ + public override maxRetryIntervalInMilliseconds?: number = 6 * 24 * 60 * 60 * 1000; + + /** + * Response status codes that fail the activity and trigger retries. Omitted/empty means every + * non-2xx response (including 4xx/5xx and unfollowed redirects), matching .NET EnsureSuccessStatusCode. + * A non-empty list is exact: even `202` can be retried instead of polled. Excluded codes are returned. + * Activity failures, including transport errors, are retried regardless of this list. + */ + public statusCodesToRetry?: number[]; +} + +/** Options accepted by `context.df.callHttp`. */ export interface CallHttpOptions { /** The HTTP request method. */ method: string; @@ -56,6 +74,8 @@ export interface CallHttpOptions { headers?: { [key: string]: string }; /** The source of the OAuth token to add to the request. */ tokenSource?: TokenSource; + /** Optional durable failure retry policy for the initial request only. Omitted means no retries. */ + retryOptions?: HttpRetryOptions; /** * Whether to keep polling the request after receiving a `202 Accepted` response. Replaces the * deprecated `asynchronousPatternEnabled`; if both are specified, `enablePolling` takes precedence. @@ -135,4 +155,6 @@ export interface DurableHttpRequestPayload { tokenSource?: { kind?: string; resource: string }; /** Whether to keep polling on `202 Accepted`. Defaults to `true` when absent. */ enablePolling?: boolean; + /** Serializable retry policy for this request; methods are never included in history. */ + retryOptions?: Omit; } diff --git a/packages/azure-functions-durable/src/index.ts b/packages/azure-functions-durable/src/index.ts index 4c2cc4c2..319c55d2 100644 --- a/packages/azure-functions-durable/src/index.ts +++ b/packages/azure-functions-durable/src/index.ts @@ -56,7 +56,7 @@ export { OrchestrationFilter } from "./orchestration-filter"; // Durable HTTP (classic v3 `context.df.callHttp`) models. Value exports (classes) and type exports // so `import { DurableHttpRequest, ManagedIdentityTokenSource, CallHttpOptions, DurableHttpResponse, // TokenSource } from ...` resolves as it did in durable-functions v3. -export { DurableHttpRequest, ManagedIdentityTokenSource } from "./http/models"; +export { DurableHttpRequest, HttpRetryOptions, ManagedIdentityTokenSource } from "./http/models"; export type { CallHttpOptions, DurableHttpResponse, TokenSource } from "./http/models"; // Legacy durable-functions v3 API compatibility aliases (types only). These let orchestrator/ diff --git a/packages/azure-functions-durable/src/orchestration-context.ts b/packages/azure-functions-durable/src/orchestration-context.ts index fe6dad3b..4158f1be 100644 --- a/packages/azure-functions-durable/src/orchestration-context.ts +++ b/packages/azure-functions-durable/src/orchestration-context.ts @@ -168,6 +168,10 @@ export class DurableOrchestrationContext { * endpoint returns `202 Accepted`, waits on durable timers (honoring `Retry-After`) and re-polls * until completion. The whole flow is a single `yield`, preserving the v3 ergonomics. * + * `retryOptions` enables durable failure retries for the initial request only. Exhaustion throws + * `TaskFailedError` into the calling generator; excluded HTTP statuses are returned unchanged. + * Like the .NET in-process extension, subsequent `202` Location polls do not inherit this policy. + * * Trust-boundary change vs v3: the HTTP request now runs as a durable **activity inside the app / * worker process** (via `fetch`), not in the Functions host extension. Outbound network path, * identity, and firewall/VNet behavior therefore follow the worker process. Managed-identity @@ -195,6 +199,19 @@ export class DurableOrchestrationContext { if (options.tokenSource !== undefined) { request.tokenSource = { kind: options.tokenSource.kind, resource: options.tokenSource.resource }; } + if (options.retryOptions !== undefined) { + const policy = options.retryOptions.toRetryPolicy(); + request.retryOptions = { + firstRetryIntervalInMilliseconds: policy.firstRetryIntervalInMilliseconds, + maxNumberOfAttempts: policy.maxNumberOfAttempts, + backoffCoefficient: policy.backoffCoefficient, + maxRetryIntervalInMilliseconds: policy.maxRetryIntervalInMilliseconds, + retryTimeoutInMilliseconds: policy.retryTimeoutInMilliseconds, + }; + if (options.retryOptions.statusCodesToRetry != null) { + request.retryOptions.statusCodesToRetry = [...options.retryOptions.statusCodesToRetry]; + } + } return this._ctx.callSubOrchestrator( BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, request, diff --git a/packages/azure-functions-durable/test/unit/http-builtin.spec.ts b/packages/azure-functions-durable/test/unit/http-builtin.spec.ts index f556ddb6..a88d44df 100644 --- a/packages/azure-functions-durable/test/unit/http-builtin.spec.ts +++ b/packages/azure-functions-durable/test/unit/http-builtin.spec.ts @@ -8,7 +8,7 @@ import { isSameOrigin, retryAfterSeconds, } from "../../src/http/builtin"; -import { DurableHttpRequestPayload, DurableHttpResponse } from "../../src/http/models"; +import { DurableHttpRequestPayload, DurableHttpResponse, HttpRetryOptions } from "../../src/http/models"; // `@azure/identity` is an OPTIONAL peer dependency loaded lazily via `require` inside the activity. // A `{ virtual: true }` mock stands in so the token-acquisition path can be exercised and the REAL @@ -180,6 +180,44 @@ describe("builtinHttpActivity", () => { }); }); + it.each([429, 503])("cancels a selected %s body without buffering it", async (status) => { + const cancel = jest.fn(async () => undefined); + const text = jest.fn(async () => "discarded"); + const response = { + ...fakeResponse(status, {}, ""), + body: { cancel }, + text, + }; + global.fetch = makeFetchMock(response) as unknown as typeof fetch; + await expect( + builtinHttpActivity({ method: "POST", uri: "http://127.0.0.1/", retryOptions: new HttpRetryOptions(100, 2) }), + ).rejects.toThrow(`HTTP request failed with status code ${status}.`); + expect(cancel).toHaveBeenCalledTimes(1); + expect(text).not.toHaveBeenCalled(); + }); + + it("fails a selected status with a null response body", async () => { + global.fetch = jest.fn(async () => new Response(null, { status: 503 })); + await expect( + builtinHttpActivity({ method: "HEAD", uri: "http://127.0.0.1/", retryOptions: new HttpRetryOptions(100, 2) }), + ).rejects.toThrow("HTTP request failed with status code 503."); + }); + + it("surfaces a body cancellation failure rather than silently discarding it", async () => { + const cleanupError = new Error("body cancellation failed"); + const text = jest.fn(async () => "discarded"); + const response = { + ...fakeResponse(503, {}, ""), + body: { cancel: jest.fn().mockRejectedValue(cleanupError) }, + text, + }; + global.fetch = makeFetchMock(response) as unknown as typeof fetch; + await expect( + builtinHttpActivity({ method: "POST", uri: "http://127.0.0.1/", retryOptions: new HttpRetryOptions(100, 2) }), + ).rejects.toBe(cleanupError); + expect(text).not.toHaveBeenCalled(); + }); + it("sends a body for non-GET/HEAD methods", async () => { const fetchMock = makeFetchMock(fakeResponse(200, {}, "")); global.fetch = fetchMock as unknown as typeof fetch; diff --git a/packages/azure-functions-durable/test/unit/http-retry.spec.ts b/packages/azure-functions-durable/test/unit/http-retry.spec.ts new file mode 100644 index 00000000..c274e846 --- /dev/null +++ b/packages/azure-functions-durable/test/unit/http-retry.spec.ts @@ -0,0 +1,363 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { createServer, IncomingHttpHeaders, Server } from "node:http"; +import { AddressInfo } from "node:net"; +import { + InMemoryOrchestrationBackend, + NoOpLogger, + OrchestrationStatus, + Task, + TaskFailedError, + TestOrchestrationClient, + TestOrchestrationWorker, +} from "@microsoft/durabletask-js"; +import * as pb from "../../../durabletask-js/src/proto/orchestrator_service_pb"; +import * as ph from "../../../durabletask-js/src/utils/pb-helper.util"; +import { + BUILTIN_HTTP_ACTIVITY_NAME, + BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, + builtinHttpActivity, + builtinHttpPollOrchestrator, +} from "../../src/http/builtin"; +import { DurableHttpRequestPayload, HttpRetryOptions } from "../../src/http/models"; +import { ClassicOrchestrationContext, wrapOrchestrator } from "../../src/orchestration-context"; +import { DurableFunctionsWorker } from "../../src/worker"; + +const START = new Date("2026-01-01T00:00:00Z"); + +// The real Functions worker replays protobuf history; only timer delivery is accelerated. +async function runHttp(request: DurableHttpRequestPayload, activityDurationMs = 0) { + const worker = new DurableFunctionsWorker({ logger: new NoOpLogger() }); + worker.addNamedOrchestrator(BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, builtinHttpPollOrchestrator); + const history: pb.HistoryEvent[] = []; + const requests: DurableHttpRequestPayload[] = []; + const delays: number[] = []; + let now = START; + let events = [ + ph.newOrchestratorStartedEvent(now), + ph.newExecutionStartedEvent(BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, "http", JSON.stringify(request), { + name: "parent", + instanceId: "parent", + taskScheduledId: 1, + }), + ]; + const replay = async () => { + const input = new pb.OrchestratorRequest() + .setInstanceid("http") + .setPasteventsList(history) + .setNeweventsList(events); + const output = await worker.handleOrchestratorRequest(Buffer.from(input.serializeBinary()).toString("base64")); + return pb.OrchestratorResponse.deserializeBinary(Buffer.from(output, "base64")).getActionsList(); + }; + for (let turn = 0; turn < 40; turn++) { + const replays: pb.OrchestratorAction[][] = []; + for (let pass = 0; pass < 2; pass++) replays.push(await replay()); + const actions = replays[0]; + // Reprocessing the same history must emit identical actions, without doing network I/O. + expect(replays[1].map((a) => a.toObject())).toEqual(actions.map((a) => a.toObject())); + expect(actions).toHaveLength(1); + history.push(...events); + const action = actions[0]; + const completed = action.getCompleteorchestration(); + if (completed) { + return { completed, requests, delays, history }; + } + const scheduled = action.getScheduletask(); + if (scheduled) { + expect(scheduled.getName()).toBe(BUILTIN_HTTP_ACTIVITY_NAME); + const input: DurableHttpRequestPayload = JSON.parse(scheduled.getInput()!.getValue()); + requests.push(input); + history.push(ph.newTaskScheduledEvent(action.getId(), scheduled.getName(), scheduled.getInput()!.getValue())); + now = new Date(now.getTime() + activityDurationMs); + let result: pb.HistoryEvent; + try { + result = ph.newTaskCompletedEvent(action.getId(), JSON.stringify(await builtinHttpActivity(input))); + } catch (error) { + result = ph.newTaskFailedEvent(action.getId(), error instanceof Error ? error : new Error(String(error))); + } + events = [ph.newOrchestratorStartedEvent(now), result]; + } else { + const fireAt = action.getCreatetimer()!.getFireat()!.toDate(); + delays.push(fireAt.getTime() - now.getTime()); + history.push(ph.newTimerCreatedEvent(action.getId(), fireAt)); + now = fireAt; + events = [ph.newOrchestratorStartedEvent(now), ph.newTimerFiredEvent(action.getId(), now)]; + } + } + throw new Error("HTTP test exceeded its bounded replay turns."); +} + +describe("durable HTTP failure retries", () => { + let server: Server; + let uri: string; + let responses: (number | "disconnect" | "open503")[]; + let discardedBodyClosed: Promise; + let received: { method?: string; url?: string; headers: IncomingHttpHeaders; body: string }[]; + + beforeEach(async () => { + received = []; + responses = [200]; + let notifyBodyClosed: (withoutEnd: boolean) => void; + discardedBodyClosed = new Promise((resolve) => (notifyBodyClosed = resolve)); + server = createServer((req, res) => { + let body = ""; + req.setEncoding("utf8"); + req.on("data", (chunk: string) => (body += chunk)); + req.on("end", () => { + received.push({ method: req.method, url: req.url, headers: req.headers, body }); + const status = responses.shift() ?? 200; + if (status === "disconnect") { + req.socket.destroy(); + return; + } + if (status === "open503") { + res.writeHead(503); + res.write("partial body that the server never ends"); + res.once("close", () => notifyBodyClosed(!res.writableEnded)); + return; + } + res.writeHead(status, status === 202 ? { Location: "/status", "Retry-After": "2" } : {}); + res.end(`response ${status}`); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + uri = `http://127.0.0.1:${(server.address() as AddressInfo).port}/start`; + }); + + afterEach(async () => { + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); + }); + + function request(codes?: number[], attempts = 3) { + return { + method: "POST", + uri, + content: "original body", + retryOptions: Object.assign(new HttpRetryOptions(1000, attempts), { statusCodesToRetry: codes }), + }; + } + + it("cancels an open retryable response body and durably retries before the server ends it", async () => { + responses = ["open503", 200]; + const run = runHttp(request([503])); + let deadline: ReturnType | undefined; + try { + const [{ completed, delays, history }, closedWithoutEnd] = await Promise.race([ + Promise.all([run, discardedBodyClosed]), + new Promise((_, reject) => { + deadline = setTimeout( + () => reject(new Error("Retry blocked waiting for the discarded response body.")), + 2000, + ); + }), + ]); + expect(closedWithoutEnd).toBe(true); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received).toHaveLength(2); + expect(delays).toEqual([1000]); + expect(history.filter((event) => event.hasTaskfailed())).toHaveLength(1); + } finally { + clearTimeout(deadline); + server.closeAllConnections(); + await Promise.allSettled([run]); + } + }); + + it.each([429, 503])( + "retries configured %s then returns success, without duplicate requests on replay", + async (code) => { + responses = [code, 200]; + const { completed, requests, delays, history } = await runHttp(request([code])); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received).toHaveLength(2); + expect(received.map((r) => [r.method, r.body])).toEqual([ + ["POST", "original body"], + ["POST", "original body"], + ]); + expect(requests[1]).toEqual(requests[0]); + expect(delays).toEqual([1000]); + expect(history.filter((e) => e.hasTaskfailed())).toHaveLength(1); + }, + ); + + it("fails with the last status when the total attempt budget is exhausted", async () => { + responses = [503, 503, 503, 200]; + const { completed, delays } = await runHttp(request([503])); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completed.getFailuredetails()?.getErrormessage()).toContain("503"); + expect(received).toHaveLength(3); + expect(delays).toEqual([1000, 1000]); + }); + + it("does not retry when maxNumberOfAttempts is one", async () => { + responses = [503, 200]; + const { completed, delays } = await runHttp(request([503], 1)); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(received).toHaveLength(1); + expect(delays).toEqual([]); + }); + + it.each([undefined, []])("defaults an omitted/empty status list (%j) to non-success responses", async (codes) => { + responses = [400, 500, 200]; + const { completed, delays } = await runHttp(request(codes)); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received).toHaveLength(3); + expect(delays).toEqual([1000, 1000]); + }); + + it("also retries an unfollowed redirect with the default status list, like .NET EnsureSuccessStatusCode", async () => { + responses = [302, 200]; + const { completed, delays } = await runHttp(request()); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received).toHaveLength(2); + expect(delays).toEqual([1000]); + }); + + it("returns an excluded error response without retrying or failing", async () => { + responses = [400]; + const { completed, delays } = await runHttp(request([429, 503])); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(400); + expect(JSON.parse(completed.getResult()!.getValue()).content).toBe("response 400"); + expect(received).toHaveLength(1); + expect(delays).toEqual([]); + }); + + it("preserves error responses when the policy is omitted", async () => { + responses = [503]; + const { completed, delays } = await runHttp({ method: "GET", uri }); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(503); + expect(JSON.parse(completed.getResult()!.getValue()).content).toBe("response 503"); + expect(received).toHaveLength(1); + expect(delays).toEqual([]); + }); + + it("retries a transport failure using the same durable activity policy", async () => { + responses = ["disconnect", 200]; + const { completed, delays } = await runHttp(request([503])); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received).toHaveLength(2); + expect(delays).toEqual([1000]); + }); + + it("preserves transport failure without retries when the policy is omitted", async () => { + responses = ["disconnect", 200]; + const { completed, delays } = await runHttp({ method: "POST", uri }); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(received).toHaveLength(1); + expect(delays).toEqual([]); + }); + + it("uses exponential backoff capped by the maximum interval", async () => { + responses = [503, 503, 503, 200]; + const input = request([503], 4); + input.retryOptions.backoffCoefficient = 3; + input.retryOptions.maxRetryIntervalInMilliseconds = 2000; + const { completed, delays } = await runHttp(input); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(delays).toEqual([1000, 2000, 2000]); + }); + + it.each([ + [0, 2], + [1, 1], + [1000, 1], + ])("enforces the retry timeout with %s ms spent in each activity (%s attempts)", async (duration, attempts) => { + responses = [503, 503, 503]; + const input = request([503]); + input.retryOptions.retryTimeoutInMilliseconds = 1000; + const { completed, delays } = await runHttp(input, duration); + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(received).toHaveLength(attempts); + expect(delays).toEqual(attempts === 2 ? [1000] : []); + }); + + it("keeps 202 polling separate: retries the initial request but not a failed Location poll", async () => { + responses = [503, 202, 503, 200]; + const input = { ...request([503]), headers: { Authorization: "test-token", "x-functions-key": "test-key" } }; + const { completed, requests, delays } = await runHttp(input); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(503); + expect(received.map((r) => [r.method, r.url])).toEqual([ + ["POST", "/start"], + ["POST", "/start"], + ["GET", "/status"], + ]); + expect(requests[2]).not.toHaveProperty("retryOptions"); + expect(received[2].headers.authorization).toBe("test-token"); + expect(received[2].headers["x-functions-key"]).toBeUndefined(); + expect(delays).toEqual([1000, 2000]); + }); + + it("treats an explicitly listed 202 as a failure rather than entering the polling loop", async () => { + responses = [202, 200]; + const { completed, delays } = await runHttp(request([202])); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received.map((r) => r.url)).toEqual(["/start", "/start"]); + expect(delays).toEqual([1000]); + }); + + it("allows default-policy 202 polling to complete normally", async () => { + responses = [202, 200]; + const { completed, delays } = await runHttp(request()); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(received.map((r) => r.url)).toEqual(["/start", "/status"]); + expect(delays).toEqual([2000]); + }); + + it("splits the default six-day HTTP retry cap into durable three-day timer segments", async () => { + responses = [503, 503, 200]; + const input = request([503]); + input.retryOptions = Object.assign(new HttpRetryOptions(2 * 24 * 60 * 60 * 1000, 3), { + backoffCoefficient: 4, + statusCodesToRetry: [503], + }); + const { completed, delays } = await runHttp(input); + expect(JSON.parse(completed.getResult()!.getValue()).statusCode).toBe(200); + expect(delays).toEqual([2, 3, 3].map((days) => days * 24 * 60 * 60 * 1000)); + }); + + it.each([false, true])("runs classic callHttp through the in-memory backend (exhausted: %s)", async (exhausted) => { + responses = exhausted ? [503, 503] : [503, 200]; + const backend = new InMemoryOrchestrationBackend(); + const worker = new TestOrchestrationWorker(backend); + const client = new TestOrchestrationClient(backend); + worker.addNamedActivity(BUILTIN_HTTP_ACTIVITY_NAME, async (_ctx, input: DurableHttpRequestPayload) => + builtinHttpActivity(input), + ); + worker.addNamedOrchestrator(BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, builtinHttpPollOrchestrator); + worker.addNamedOrchestrator( + "parent", + wrapOrchestrator(function* (context: ClassicOrchestrationContext): Generator, unknown, unknown> { + try { + return yield context.df.callHttp({ + method: "POST", + url: uri, + retryOptions: new HttpRetryOptions(1, 2), + }); + } catch (error) { + if (!(error instanceof TaskFailedError)) throw error; + return { failure: error.details.message }; + } + }), + ); + await worker.start(); + try { + const id = await client.scheduleNewOrchestration("parent"); + const state = await client.waitForOrchestrationCompletion(id, true, 10); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + const output = JSON.parse(state!.serializedOutput!); + if (exhausted) { + expect(output.failure).toContain("503"); + } else { + expect(output.statusCode).toBe(200); + } + expect(received).toHaveLength(2); + } finally { + await worker.stop(); + backend.reset(); + } + }); +}); diff --git a/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts index f7a82df6..e39a3e62 100644 --- a/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts +++ b/packages/azure-functions-durable/test/unit/orchestration-context.spec.ts @@ -21,6 +21,8 @@ import { wrapOrchestrator, } from "../../src/orchestration-context"; import { RetryOptions } from "../../src/retry-options"; +import { HttpRetryOptions } from "../../src/http/models"; +import type { HttpRetryOptions as PublicHttpRetryOptions } from "../../src"; import { BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME } from "../../src/http/builtin"; /** Builds a fake core OrchestrationContext whose methods return sentinel values via jest mocks. */ @@ -197,6 +199,75 @@ describe("DurableOrchestrationContext", () => { }); }); + it("snapshots a validated HTTP retry policy and status list into the serialized request", () => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + const retryOptions = Object.assign(new HttpRetryOptions(1000, 3), { statusCodesToRetry: [429, 503] }); + retryOptions.backoffCoefficient = 2; + retryOptions.maxRetryIntervalInMilliseconds = 10000; + retryOptions.retryTimeoutInMilliseconds = 60000; + const options = { method: "POST", url: "https://example.test/api", retryOptions }; + df.callHttp(options); + retryOptions.statusCodesToRetry.push(500); + retryOptions.backoffCoefficient = 5; + const payload = raw.callSubOrchestrator.mock.calls[0][1]; + expect(payload.retryOptions).toEqual({ + firstRetryIntervalInMilliseconds: 1000, + maxNumberOfAttempts: 3, + backoffCoefficient: 2, + maxRetryIntervalInMilliseconds: 10000, + retryTimeoutInMilliseconds: 60000, + statusCodesToRetry: [429, 503], + }); + expect(JSON.parse(JSON.stringify(payload)).retryOptions).toEqual(payload.retryOptions); + }); + + it("treats a runtime null HTTP status list as the default, matching .NET", () => { + const { ctx, raw } = createFakeCoreContext(); + const retryOptions = new HttpRetryOptions(1000, 3); + Reflect.set(retryOptions, "statusCodesToRetry", null); + new DurableOrchestrationContext(ctx, undefined).callHttp({ + method: "GET", + url: "https://example.test/api", + retryOptions, + }); + expect(raw.callSubOrchestrator.mock.calls[0][1].retryOptions).not.toHaveProperty("statusCodesToRetry"); + }); + + it("snapshots HTTP defaults without changing the generic activity retry defaults", () => { + const { ctx, raw } = createFakeCoreContext(); + const retryOptions: PublicHttpRetryOptions = new HttpRetryOptions(1000, 3); + new DurableOrchestrationContext(ctx, undefined).callHttp({ + method: "GET", + url: "https://example.test/api", + retryOptions, + }); + expect(raw.callSubOrchestrator.mock.calls[0][1].retryOptions).toEqual({ + firstRetryIntervalInMilliseconds: 1000, + maxNumberOfAttempts: 3, + backoffCoefficient: 1, + maxRetryIntervalInMilliseconds: 6 * 24 * 60 * 60 * 1000, + retryTimeoutInMilliseconds: -1, + }); + expect(new RetryOptions(1000, 3).toRetryPolicy().maxRetryIntervalInMilliseconds).toBe(3600000); + }); + + it.each([ + ["firstRetryIntervalInMilliseconds", { firstRetryIntervalInMilliseconds: 0 }], + ["maxNumberOfAttempts", { maxNumberOfAttempts: 0 }], + ["maxNumberOfAttempts", { maxNumberOfAttempts: NaN }], + ["backoffCoefficient", { backoffCoefficient: 0.5 }], + ["maxRetryIntervalInMilliseconds", { maxRetryIntervalInMilliseconds: 999 }], + ["retryTimeoutInMilliseconds", { retryTimeoutInMilliseconds: 999 }], + ])("rejects invalid HTTP %s before scheduling any work", (field, invalid) => { + const { ctx, raw } = createFakeCoreContext(); + const df = new DurableOrchestrationContext(ctx, undefined); + const retryOptions = Object.assign(new HttpRetryOptions(1000, 3), invalid); + const options = { method: "GET", url: "https://example.test/api", retryOptions }; + expect(() => df.callHttp(options)).toThrow(field); + expect(raw.callSubOrchestrator).not.toHaveBeenCalled(); + }); + it("honors enablePolling=false (and the deprecated asynchronousPatternEnabled alias) for callHttp", () => { const { ctx, raw } = createFakeCoreContext(); const df = new DurableOrchestrationContext(ctx, undefined);