From cce4123289450545d03e2428790fc656ed65f96f Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 29 Sep 2026 12:48:07 -0400 Subject: [PATCH 1/2] feat(functions): add opt-in durable HTTP failure retries Add HttpRetryOptions and serialize its policy and status-code snapshot for the built-in HTTP activity. Reuse durable RetryPolicy execution, preserve calls without options, and match .NET in-process initial-request-only retry semantics. Closes #367 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0af01a5-0dfa-4e71-a660-c4186e65d7e0 --- packages/azure-functions-durable/CHANGELOG.md | 6 + packages/azure-functions-durable/README.md | 64 +++- .../src/http/builtin.ts | 33 +- .../src/http/models.ts | 24 +- packages/azure-functions-durable/src/index.ts | 2 +- .../src/orchestration-context.ts | 17 + .../test/unit/http-retry.spec.ts | 326 ++++++++++++++++++ .../test/unit/orchestration-context.spec.ts | 71 ++++ 8 files changed, 530 insertions(+), 13 deletions(-) create mode 100644 packages/azure-functions-durable/test/unit/http-retry.spec.ts diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 4caac949..f3bc0657 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()`. @@ -34,6 +39,7 @@ and `TaskCancelledError` from canceled results. Timer `result` now aliases `getResult()`, including errors while pending or failed. Drain affected instances before mixing versions or rollback; cancellation branching and already-segmented histories can change replay. + ## v4.0.0-beta.1 (2026-07-31) ### Changes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index e4450768..3baa5be2 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,67 @@ 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. +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..92a0c732 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), @@ -267,9 +267,7 @@ export async function builtinHttpActivity(input: DurableHttpRequestPayload): Pro // The Fetch standard forbids a request body on GET/HEAD. v3 (the host extension) attached content // regardless of method; rather than silently drop it and change the request, fail loudly. if (request.content !== undefined && (method === "GET" || method === "HEAD")) { - throw new Error( - `callHttp: an HTTP ${method} request cannot carry a body; remove 'body' or use POST/PUT/PATCH.`, - ); + throw new Error(`callHttp: an HTTP ${method} request cannot carry a body; remove 'body' or use POST/PUT/PATCH.`); } const headers: { [key: string]: string } = { ...(request.headers ?? {}) }; @@ -305,6 +303,16 @@ export async function builtinHttpActivity(input: DurableHttpRequestPayload): Pro }); const content = await response.text(); + 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) { + throw new Error(`HTTP request failed with status code ${response.status}.`); + } + } + const result: BuiltinHttpActivityResult = { statusCode: response.status, headers: responseHeaders, content }; // Record the post-redirect URL so a relative `Location` is resolved against the URI that actually // produced this response (fetch follows redirects by default). @@ -345,6 +353,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 +399,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-retry.spec.ts b/packages/azure-functions-durable/test/unit/http-retry.spec.ts new file mode 100644 index 00000000..7f3f33ad --- /dev/null +++ b/packages/azure-functions-durable/test/unit/http-retry.spec.ts @@ -0,0 +1,326 @@ +// 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")[]; + let received: { method?: string; url?: string; headers: IncomingHttpHeaders; body: string }[]; + + beforeEach(async () => { + received = []; + responses = [200]; + 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; + } + 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.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(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(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); From 57efe8bda358ad450a9b92af2f4b8f3a519df379 Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 2 Oct 2026 11:21:22 -0400 Subject: [PATCH 2/2] fix(functions): cancel retryable HTTP bodies before buffering Classify selected HTTP failures as soon as headers arrive and cancel their unused body so an open response cannot block durable retries. Keep returned response bodies intact and restore unrelated baseline formatting. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0af01a5-0dfa-4e71-a660-c4186e65d7e0 --- packages/azure-functions-durable/CHANGELOG.md | 1 - packages/azure-functions-durable/README.md | 2 + .../src/http/builtin.ts | 18 +++++---- .../test/unit/http-builtin.spec.ts | 40 ++++++++++++++++++- .../test/unit/http-retry.spec.ts | 39 +++++++++++++++++- 5 files changed, 90 insertions(+), 10 deletions(-) diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index f3bc0657..24a0e96a 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -39,7 +39,6 @@ and `TaskCancelledError` from canceled results. Timer `result` now aliases `getResult()`, including errors while pending or failed. Drain affected instances before mixing versions or rollback; cancellation branching and already-segmented histories can change replay. - ## v4.0.0-beta.1 (2026-07-31) ### Changes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index 3baa5be2..03dd6bdf 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -161,6 +161,8 @@ df.app.orchestration("FetchReport", function* (context: df.OrchestrationContext) 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 diff --git a/packages/azure-functions-durable/src/http/builtin.ts b/packages/azure-functions-durable/src/http/builtin.ts index 92a0c732..970d1a9e 100644 --- a/packages/azure-functions-durable/src/http/builtin.ts +++ b/packages/azure-functions-durable/src/http/builtin.ts @@ -267,7 +267,9 @@ export async function builtinHttpActivity(input: DurableHttpRequestPayload): Pro // The Fetch standard forbids a request body on GET/HEAD. v3 (the host extension) attached content // regardless of method; rather than silently drop it and change the request, fail loudly. if (request.content !== undefined && (method === "GET" || method === "HEAD")) { - throw new Error(`callHttp: an HTTP ${method} request cannot carry a body; remove 'body' or use POST/PUT/PATCH.`); + throw new Error( + `callHttp: an HTTP ${method} request cannot carry a body; remove 'body' or use POST/PUT/PATCH.`, + ); } const headers: { [key: string]: string } = { ...(request.headers ?? {}) }; @@ -297,22 +299,24 @@ export async function builtinHttpActivity(input: DurableHttpRequestPayload): Pro body: includeBody ? request.content : undefined, }); - const responseHeaders: { [key: string]: string } = {}; - response.headers.forEach((value, key) => { - responseHeaders[key] = value; - }); - const content = await response.text(); - 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; + }); + const content = await response.text(); + const result: BuiltinHttpActivityResult = { statusCode: response.status, headers: responseHeaders, content }; // Record the post-redirect URL so a relative `Location` is resolved against the URI that actually // produced this response (fetch follows redirects by default). 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 index 7f3f33ad..c274e846 100644 --- a/packages/azure-functions-durable/test/unit/http-retry.spec.ts +++ b/packages/azure-functions-durable/test/unit/http-retry.spec.ts @@ -91,12 +91,15 @@ async function runHttp(request: DurableHttpRequestPayload, activityDurationMs = describe("durable HTTP failure retries", () => { let server: Server; let uri: string; - let responses: (number | "disconnect")[]; + 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"); @@ -108,6 +111,12 @@ describe("durable HTTP failure retries", () => { 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}`); }); @@ -130,6 +139,32 @@ describe("durable HTTP failure retries", () => { }; } + 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) => { @@ -185,6 +220,7 @@ describe("durable HTTP failure retries", () => { 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([]); }); @@ -193,6 +229,7 @@ describe("durable HTTP failure retries", () => { 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([]); });