From 6fd999b337a81fc7215de581aa676a240410cdf9 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 16 Aug 2026 19:24:54 +0900 Subject: [PATCH 1/6] chore: upgrade typescript to 4.9 and typedoc Signed-off-by: kahirokunn --- package-lock.json | 89 +++++++++++++++++++++-------------------------- package.json | 4 +-- 2 files changed, 41 insertions(+), 52 deletions(-) diff --git a/package-lock.json b/package-lock.json index bbffeea7..dc25c00d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -49,8 +49,8 @@ "remark-preset-lint-recommended": "^5.0.0", "superagent": "^7.1.1", "ts-node": "^10.8.1", - "typedoc": "^0.22.11", - "typescript": "^4.3.5", + "typedoc": "^0.23.28", + "typescript": "^4.9.5", "webpack": "^5.76.0", "webpack-cli": "^4.10.0" }, @@ -1891,6 +1891,13 @@ "node": ">=6" } }, + "node_modules/ansi-sequence-parser": { + "version": "1.1.3", + "resolved": "https://registry.npmjs.org/ansi-sequence-parser/-/ansi-sequence-parser-1.1.3.tgz", + "integrity": "sha512-+fksAx9eG3Ab6LDnLs3ZqZa8KVJ/jYnX+D4Qe1azX+LFGFAXqynCQLOdLpNYN/l9e7l6hMWwZbrnctqr6eSQSw==", + "dev": true, + "license": "MIT" + }, "node_modules/ansi-styles": { "version": "4.3.0", "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-4.3.0.tgz", @@ -8863,15 +8870,16 @@ } }, "node_modules/shiki": { - "version": "0.10.1", - "resolved": "https://registry.npmjs.org/shiki/-/shiki-0.10.1.tgz", - "integrity": "sha512-VsY7QJVzU51j5o1+DguUd+6vmCmZ5v/6gYu4vyYAhzjuNQU6P/vmSy4uQaOhvje031qQMiW0d2BwgMH52vqMng==", + "version": "0.14.7", + "resolved": "https://registry.npmjs.org/shiki/-/shiki-0.14.7.tgz", + "integrity": "sha512-dNPAPrxSc87ua2sKJ3H5dQ/6ZaY8RNnaAqK+t0eG7p0Soi2ydiqbGOTaZCqaYvA/uZYfS1LJnemt3Q+mSfcPCg==", "dev": true, "license": "MIT", "dependencies": { - "jsonc-parser": "^3.0.0", - "vscode-oniguruma": "^1.6.1", - "vscode-textmate": "5.2.0" + "ansi-sequence-parser": "^1.1.0", + "jsonc-parser": "^3.2.0", + "vscode-oniguruma": "^1.7.0", + "vscode-textmate": "^8.0.0" } }, "node_modules/side-channel": { @@ -9725,76 +9733,57 @@ } }, "node_modules/typedoc": { - "version": "0.22.18", - "resolved": "https://registry.npmjs.org/typedoc/-/typedoc-0.22.18.tgz", - "integrity": "sha512-NK9RlLhRUGMvc6Rw5USEYgT4DVAUFk7IF7Q6MYfpJ88KnTZP7EneEa4RcP+tX1auAcz7QT1Iy0bUSZBYYHdoyA==", + "version": "0.23.28", + "resolved": "https://registry.npmjs.org/typedoc/-/typedoc-0.23.28.tgz", + "integrity": "sha512-9x1+hZWTHEQcGoP7qFmlo4unUoVJLB0H/8vfO/7wqTnZxg4kPuji9y3uRzEu0ZKez63OJAUmiGhUrtukC6Uj3w==", "dev": true, "license": "Apache-2.0", "dependencies": { - "glob": "^8.0.3", "lunr": "^2.3.9", - "marked": "^4.0.16", - "minimatch": "^5.1.0", - "shiki": "^0.10.1" + "marked": "^4.2.12", + "minimatch": "^7.1.3", + "shiki": "^0.14.1" }, "bin": { "typedoc": "bin/typedoc" }, "engines": { - "node": ">= 12.10.0" + "node": ">= 14.14" }, "peerDependencies": { - "typescript": "4.0.x || 4.1.x || 4.2.x || 4.3.x || 4.4.x || 4.5.x || 4.6.x || 4.7.x" + "typescript": "4.6.x || 4.7.x || 4.8.x || 4.9.x || 5.0.x" } }, "node_modules/typedoc/node_modules/brace-expansion": { - "version": "2.0.1", - "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.0.1.tgz", - "integrity": "sha512-XnAIvQ8eM+kC6aULx6wuQiwVsnzsi9d3WxzV3FpWTGA19F621kwdbsAcFKXgKUHZWsy+mY6iL1sHTxWEFCytDA==", + "version": "2.1.4", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.1.4.tgz", + "integrity": "sha512-hGfVzPxthbf3+2yjg/RBs60cB0FhqBS/zvdV/4wn4/BmN0bNMMHPc4V/BbFieqf1TKAGGAHnY4eSjajCl0f2Xg==", "dev": true, "license": "MIT", "dependencies": { "balanced-match": "^1.0.0" } }, - "node_modules/typedoc/node_modules/glob": { - "version": "8.1.0", - "resolved": "https://registry.npmjs.org/glob/-/glob-8.1.0.tgz", - "integrity": "sha512-r8hpEjiQEYlF2QU0df3dS+nxxSIreXQS1qRhMJM0Q5NDdR386C7jb7Hwwod8Fgiuex+k0GFjgft18yvxm5XoCQ==", - "deprecated": "Glob versions prior to v9 are no longer supported", + "node_modules/typedoc/node_modules/minimatch": { + "version": "7.4.9", + "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-7.4.9.tgz", + "integrity": "sha512-Brg/fp/iAVDOQoHxkuN5bEYhyQlZhxddI78yWsCbeEwTHXQjlNLtiJDUsp1GIptVqMI7/gkJMz4vVAc01mpoBw==", "dev": true, "license": "ISC", "dependencies": { - "fs.realpath": "^1.0.0", - "inflight": "^1.0.4", - "inherits": "2", - "minimatch": "^5.0.1", - "once": "^1.3.0" + "brace-expansion": "^2.0.2" }, "engines": { - "node": ">=12" + "node": ">=10" }, "funding": { "url": "https://github.com/sponsors/isaacs" } }, - "node_modules/typedoc/node_modules/minimatch": { - "version": "5.1.6", - "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-5.1.6.tgz", - "integrity": "sha512-lKwV/1brpG6mBUFHtb7NUmtABCb2WZZmm2wNiOA5hAb8VdCS4B3dtMWyvcoViccwAW/COERjXLt0zP1zXUN26g==", - "dev": true, - "license": "ISC", - "dependencies": { - "brace-expansion": "^2.0.1" - }, - "engines": { - "node": ">=10" - } - }, "node_modules/typescript": { - "version": "4.7.4", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-4.7.4.tgz", - "integrity": "sha512-C0WQT0gezHuw6AdY1M2jxUO83Rjf0HP7Sk1DtXj6j1EwkQNZrHAg2XPWlq62oqEhYvONq5pkC2Y9oPljWToLmQ==", + "version": "4.9.5", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-4.9.5.tgz", + "integrity": "sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==", "dev": true, "license": "Apache-2.0", "bin": { @@ -10402,9 +10391,9 @@ "license": "MIT" }, "node_modules/vscode-textmate": { - "version": "5.2.0", - "resolved": "https://registry.npmjs.org/vscode-textmate/-/vscode-textmate-5.2.0.tgz", - "integrity": "sha512-Uw5ooOQxRASHgu6C7GVvUxisKXfSgW4oFlO+aa+PAkgmH89O3CXxEEzNRNtHSqtXFTl0nAC1uYj0GMSH27uwtQ==", + "version": "8.0.0", + "resolved": "https://registry.npmjs.org/vscode-textmate/-/vscode-textmate-8.0.0.tgz", + "integrity": "sha512-AFbieoL7a5LMqcnOF04ji+rpXadgOXnZsxQr//r83kLPr7biP7am3g9zbaZIaBGwBRWeSvoMD4mgPdX3e4NWBg==", "dev": true, "license": "MIT" }, diff --git a/package.json b/package.json index 214d34a5..1729988e 100644 --- a/package.json +++ b/package.json @@ -150,8 +150,8 @@ "remark-preset-lint-recommended": "^5.0.0", "superagent": "^7.1.1", "ts-node": "^10.8.1", - "typedoc": "^0.22.11", - "typescript": "^4.3.5", + "typedoc": "^0.23.28", + "typescript": "^4.9.5", "webpack": "^5.76.0", "webpack-cli": "^4.10.0" }, From 282927466a60cd5d2321d18ffb07a1cee54f20bc Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Sun, 16 Aug 2026 23:18:43 +0900 Subject: [PATCH 2/6] chore: upgrade @types/node to match the supported Node.js versions Signed-off-by: kahirokunn --- package-lock.json | 20 +++++++++++++++----- package.json | 2 +- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/package-lock.json b/package-lock.json index dc25c00d..9fcdac6d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -23,7 +23,7 @@ "@types/got": "^9.6.11", "@types/json-bigint": "^1.0.1", "@types/mocha": "^7.0.2", - "@types/node": "^14.14.10", + "@types/node": "~22.0.0", "@types/superagent": "^4.1.10", "@types/uuid": "^8.3.4", "@typescript-eslint/eslint-plugin": "^4.29.0", @@ -1295,11 +1295,14 @@ "license": "MIT" }, "node_modules/@types/node": { - "version": "14.18.63", - "resolved": "https://registry.npmjs.org/@types/node/-/node-14.18.63.tgz", - "integrity": "sha512-fAtCfv4jJg+ExtXhvCkCqUKZ+4ok/JQk01qDKhL5BDDoS3AxKXhV5/MAVUZyQnSEd2GT92fkgZl0pz0Q0AzcIQ==", + "version": "22.0.3", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.0.3.tgz", + "integrity": "sha512-/e0NZtK2gs6Vk2DoyrXSZZ4AlamqTkx0CcKx1Aq8/P4ITlRgU9OtVf5fl+LXkWWJce1M89pkSlH6lJJEnK7bQA==", "dev": true, - "license": "MIT" + "license": "MIT", + "dependencies": { + "undici-types": "~6.11.1" + } }, "node_modules/@types/responselike": { "version": "1.0.3", @@ -9813,6 +9816,13 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/undici-types": { + "version": "6.11.1", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.11.1.tgz", + "integrity": "sha512-mIDEX2ek50x0OlRgxryxsenE5XaQD4on5U2inY7RApK3SOJpofyw7uW2AyfMKkhAxXIceo2DeWGVGwyvng1GNQ==", + "dev": true, + "license": "MIT" + }, "node_modules/unified": { "version": "10.1.2", "resolved": "https://registry.npmjs.org/unified/-/unified-10.1.2.tgz", diff --git a/package.json b/package.json index 1729988e..17b31a16 100644 --- a/package.json +++ b/package.json @@ -124,7 +124,7 @@ "@types/got": "^9.6.11", "@types/json-bigint": "^1.0.1", "@types/mocha": "^7.0.2", - "@types/node": "^14.14.10", + "@types/node": "~22.0.0", "@types/superagent": "^4.1.10", "@types/uuid": "^8.3.4", "@typescript-eslint/eslint-plugin": "^4.29.0", From c33b7b527fa7bf25f757c4970f7a7f52e484fc5c Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:19:18 +0900 Subject: [PATCH 3/6] feat!: use Fetch API for HTTP transport Use the platform HTTP client in Node.js and browsers without importing Node-only http and https modules into the SDK's public entry point. Preserve CloudEvent serialization, URL credentials and per-send headers, and expose transport-wide headers and Fetch request options. Keep the transport result inferred through emitterFor(). HTTP status codes still resolve at this point; delivery failure handling is a separate change. BREAKING CHANGE: httpTransport() resolves with { response, body }, where response is the native Fetch Response, instead of { body, headers }. Node.js global HTTP agents no longer configure these requests; a Fetch dispatcher configures proxies, client certificates and custom trust. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 65 ++++ README.md | 25 +- src/index.ts | 17 +- src/message/index.ts | 2 +- src/transport/emitter.ts | 27 +- src/transport/http/index.ts | 247 ++++++++++--- test/integration/emitter_factory_test.ts | 19 +- test/integration/http_transport_test.ts | 424 +++++++++++++++++++++++ webpack.config.js | 4 +- 9 files changed, 743 insertions(+), 87 deletions(-) create mode 100644 test/integration/http_transport_test.ts diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index 4370a732..d33bd8ae 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -100,3 +100,68 @@ new CloudEvent({ type, source, data }).emit(); // You can also have several listener to send the event to several endpoint ``` + +### Upgrading From 10.x to 11.0 + +#### HTTP Transport + +`httpTransport()` uses Fetch and resolves with `{ response, body }`, where +`response` is the native Fetch `Response` and `body` is its text, instead of +`{ body, headers }`. Access headers through `response.headers.get(name)` and +the status through `response.status`. + +```js +const { response, body } = await emit(event); +console.log(response.status, response.headers.get("x-request-id"), body); +``` + +#### Proxies, Custom CAs, mTLS and Connection Pooling + +The old transport went through Node.js `http.request()` or `https.request()`, so a proxy +library or connection options set on `http.globalAgent` or `https.globalAgent` reached +the corresponding requests. Fetch does not use those agents; on Node.js the equivalent +is a dispatcher from [undici](https://www.npmjs.com/package/undici), which `fetchOptions` +passes through: + +```js +import { emitterFor, httpTransport } from "cloudevents"; +import { ProxyAgent } from "undici"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { + fetchOptions: { dispatcher: new ProxyAgent("http://proxy.example.com:3128") }, +})); +``` + +For example, an order service that presents a client certificate to an events receiver +and trusts the receiver's private CA can use an `Agent`: + +```js +import { readFileSync } from "node:fs"; +import { emitterFor, httpTransport } from "cloudevents"; +import { Agent } from "undici"; + +const dispatcher = new Agent({ + connect: { + cert: readFileSync("./certs/order-service-client.pem"), + key: readFileSync("./certs/order-service-client-key.pem"), + ca: readFileSync("./certs/events-receiver-ca.pem"), + }, +}); + +const emit = emitterFor(httpTransport("https://events.example.com/orders", { + fetchOptions: { dispatcher }, +})); +``` + +The `cert` and `key` identify the client when the receiver requires mTLS. The `ca` +controls which server certificates the client trusts, so it is needed here because the +receiver uses a private CA; it is separate from client authentication. An `Agent` also +tunes connection pooling, and `undici.setGlobalDispatcher()` applies to every send rather +than one transport. In the browser the platform handles proxies, certificates and trust, +so `dispatcher` is a Node.js-only option. + +`fetchOptions` takes the Fetch options of whichever declarations type `fetch` in your +project, so TypeScript accepts `dispatcher` only where those are undici's. Against the +DOM `RequestInit` most projects compile with, pass it as +`fetchOptions: { dispatcher } as FetchRequestInit` or reach for +`setGlobalDispatcher()` instead. diff --git a/README.md b/README.md index b40b8de3..60a6874e 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,8 @@ app.post("/", (req, res) => { #### Emitting Events -The easiest way to send events is to use the built-in HTTP emitter. +The easiest way to send events is to use the built-in HTTP emitter, which sends +them with the Fetch API. ```js const { httpTransport, emitterFor, CloudEvent } = require("cloudevents"); @@ -57,10 +58,28 @@ const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); // Create a new CloudEvent const ce = new CloudEvent({ type, source, data }); -// Send it to the endpoint - encoded as HTTP binary by default -emit(ce); +async function main() { + // Send it to the endpoint - encoded as HTTP binary by default + const { response, body } = await emit(ce); + console.log(response.status, body); +} + +main().catch(console.error); ``` +The transport resolves with the native Fetch `Response` and its body as text +for every HTTP status. It sends each event once. Redirects are not followed +unless `fetchOptions: { redirect: "follow" }` enables them; Fetch preserves the +CloudEvent POST and its body only for `307` and `308`. + +`httpTransport()` accepts transport-wide `headers` and other Fetch options under +`fetchOptions`. Headers passed to `emit()` apply to that event only and take +precedence over transport and CloudEvent headers. URL credentials become an +authorization header unless the caller overrides it. + +See the [API transition guide](./API_TRANSITION_GUIDE.md) for response access and +Node.js proxy, certificate and connection settings. + If you prefer to use another transport mechanism for sending events over HTTP, you can use the `HTTP` binding to create a `Message` which has properties for `headers` and `body`, allowing greater flexibility diff --git a/src/index.ts b/src/index.ts index ab0148db..3c3d5bcd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,8 +7,13 @@ import { CloudEvent, V1, V03 } from "./event/cloudevent"; import { ValidationError } from "./event/validation"; import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces"; -import { Options, TransportFunction, EmitterFunction, emitterFor, Emitter } from "./transport/emitter"; -import { httpTransport } from "./transport/http"; +import { + Options, TransportFunction, EmitterFunction, emitterFor, Emitter, +} from "./transport/emitter"; +import { + FetchHeadersInit, FetchRequestInit, HTTPTransportFunction, HTTPTransportOptions, + HTTPTransportResponse, httpTransport, +} from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -26,6 +31,7 @@ export { Kafka, MQTT, MQTTMessageFactory, + // From transport emitterFor, httpTransport, Emitter, @@ -48,5 +54,10 @@ export type { // From transport TransportFunction, EmitterFunction, - Options + Options, + FetchHeadersInit, + FetchRequestInit, + HTTPTransportFunction, + HTTPTransportOptions, + HTTPTransportResponse }; diff --git a/src/message/index.ts b/src/message/index.ts index 4eda7320..bee6ab06 100644 --- a/src/message/index.ts +++ b/src/message/index.ts @@ -3,7 +3,7 @@ SPDX-License-Identifier: Apache-2.0 */ -import { IncomingHttpHeaders } from "http"; +import type { IncomingHttpHeaders } from "http"; import { CloudEventV1 } from ".."; // reexport the protocol bindings diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 2a8b604c..7ddaf7d3 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -20,20 +20,22 @@ export interface Options { * EmitterFunction is an invokable interface returned by {@linkcode emitterFor}. * Invoke an EmitterFunction with a CloudEvent and optional transport * options to send the event as a Message across supported transports. + * TResult is whatever the underlying {@linkcode TransportFunction} resolves with. * @interface */ -export interface EmitterFunction { - (event: CloudEvent, options?: Options): Promise; +export interface EmitterFunction { + (event: CloudEvent, options?: TOptions): Promise; } /** * TransportFunction is an invokable interface provided to the emitterFactory. * A TransportFunction's responsiblity is to send a JSON encoded event Message - * across the wire. + * across the wire. TResult is the value it resolves with, e.g. an HTTP client's + * response, or void for a transport that has nothing to hand back. * @interface */ -export interface TransportFunction { - (message: Message, options?: Options): Promise; +export interface TransportFunction { + (message: Message, options?: TOptions): Promise; } const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; @@ -51,19 +53,24 @@ const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; * @param {Mode} options.mode the encoding mode (Mode.BINARY or Mode.STRUCTURED) * @returns {EmitterFunction} an EmitterFunction to send events with */ -export function emitterFor(fn: TransportFunction, options = emitterDefaults): EmitterFunction { +export function emitterFor( + fn: TransportFunction, options?: Options, +): EmitterFunction; +export function emitterFor( + fn: TransportFunction, options: Options = emitterDefaults, +): EmitterFunction { if (!fn) { throw new TypeError("A TransportFunction is required"); } const { binding, mode }: any = { ...emitterDefaults, ...options }; - return function emit(event: CloudEvent, opts?: Options): Promise { - opts = opts || {}; + return function emit(event: CloudEvent, opts?: TOptions): Promise { + const transportOptions = opts ?? {} as TOptions; switch (mode) { case Mode.BINARY: - return fn(binding.binary(event), opts); + return fn(binding.binary(event), transportOptions); case Mode.STRUCTURED: - return fn(binding.structured(event), opts); + return fn(binding.structured(event), transportOptions); default: throw new TypeError(`Unexpected transport mode: ${mode}`); } diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 2ac2062c..1b5e9604 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -3,61 +3,204 @@ SPDX-License-Identifier: Apache-2.0 */ -import { Socket } from "net"; -import http, { OutgoingHttpHeaders } from "http"; -import https, { RequestOptions } from "https"; - -import { Message, Options } from "../.."; -import { TransportFunction } from "../emitter"; - -/** - * httpTransport provides a simple HTTP Transport function, which can send a CloudEvent, - * encoded as a Message to the endpoint. The returned function can be used with emitterFor() - * to provide an event emitter, for example: - * - * const emitter = emitterFor(httpTransport("http://example.com")); - * emitter.emit(myCloudEvent) - * .then(resp => console.log(resp)); - * +import { Headers as CloudEventHeaders, Message } from "../../message"; +import { Options, TransportFunction } from "../emitter"; + +/** The request options accepted by the Fetch implementation in the current environment */ +export type FetchRequestInit = NonNullable[1]>; + +/** The header forms accepted by the Fetch implementation in the current environment */ +export type FetchHeadersInit = NonNullable; + +/** Options applied to every request sent by an HTTP transport */ +export interface HTTPTransportOptions { + /** HTTP headers applied to every request, per-send headers take precedence */ + headers?: FetchHeadersInit; + /** Fetch options applied to every request, apart from the ones this transport controls */ + fetchOptions?: Omit; +} + +/** The native Fetch response and its body, read as text */ +export interface HTTPTransportResponse { + response: Response; + body: string; +} + +/** A transport that returns an HTTP response */ +export type HTTPTransportFunction = TransportFunction; + +/** + * Send an HTTP-bound CloudEvent with Fetch. The transport sends once and resolves with + * the native response and its body as text, for every HTTP status. Redirects are not + * followed unless fetchOptions.redirect enables them. Credentials in the sink URL become + * an authorization header, overridden by transport or per-send headers. + * + * @param {string|URL} sink the destination endpoint for the event + * @param {HTTPTransportOptions} options headers and Fetch request options + * @returns {HTTPTransportFunction} a function that sends CloudEvents to the sink + */ +export function httpTransport( + sink: string | URL, options: HTTPTransportOptions = {}, +): HTTPTransportFunction { + const url = validateHTTPURL(sink); + // fetch refuses a URL with credentials + const sinkHeaders = credentialsFrom(url); + const transportHeaders = headersFrom(options.headers); + const fetchOptions: FetchRequestInit = { + ...options.fetchOptions, + // fetch reads an explicitly undefined redirect as absent, so the default comes last + redirect: options.fetchOptions?.redirect ?? "manual", + method: "POST", + }; + + return async (message: Message, sendOptions?: Options): Promise => { + const headers = requestHeaders( + sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), + ); + const response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + }); + return { response, body: await response.text() }; + }; +} + +/** + * Check that the sink is a URL this transport knows how to POST to + * * @param {string|URL} sink the destination endpoint for the event - * @returns {TransportFunction} a function which can be used to send CloudEvents to _sink_ + * @returns {URL} the parsed sink */ -export function httpTransport(sink: string | URL): TransportFunction { +function validateHTTPURL(sink: string | URL): URL { const url = new URL(sink); - let base: any; - if (url.protocol === "https:") { - base = https; - } else if (url.protocol === "http:") { - base = http; - } else { + if (url.protocol !== "http:" && url.protocol !== "https:") { throw new TypeError(`unsupported protocol ${url.protocol}`); } - return function(message: Message, options?: Options): Promise { - return new Promise((resolve, reject) => { - options = { ...options }; - - // TODO: Callers should be able to set any Node.js RequestOptions - const opts: RequestOptions = { - method: "POST", - headers: {...message.headers, ...options.headers as OutgoingHttpHeaders}, - }; - try { - const response = { - body: "", - headers: {}, - }; - const req = base.request(url, opts, (res: Socket) => { - res.setEncoding("utf-8"); - response.headers = (res as any).headers; - res.on("data", (chunk) => response.body += chunk); - res.on("end", () => { resolve(response); }); - }); - req.on("error", reject); - req.write(message.body); - req.end(); - } catch (err) { - reject(err); - } - }); - }; + return url; +} + +/** + * Merge the headers for a single request + * + * @param {Headers} sinkHeaders the headers built from the sink URL, if it carried credentials + * @param {CloudEventHeaders} messageHeaders the headers of the event Message + * @param {Headers} transportHeaders the headers configured on this transport, if any + * @param {Headers} sendHeaders the headers supplied with this send, if any + * @returns {Headers} the headers to send + */ +function requestHeaders( + sinkHeaders: Headers | undefined, messageHeaders: CloudEventHeaders, + transportHeaders?: Headers, sendHeaders?: Headers, +): Headers { + const headers = new Headers(); + // lowest precedence first: sink credentials, binding, transport, this send + setHeaders(headers, sinkHeaders); + setHeaders(headers, messageHeaders); + setHeaders(headers, transportHeaders); + setHeaders(headers, sendHeaders); + return headers; +} + +/** + * Move the credentials of a sink URL into the header the old Node.js transport built, since + * Fetch refuses a URL that carries them + * + * @param {URL} url the sink URL, whose credentials are removed + * @returns {Headers|undefined} the authorization header, or undefined for a URL without credentials + */ +function credentialsFrom(url: URL): Headers | undefined { + if (url.username === "" && url.password === "") { + return undefined; + } + const userinfo = `${decodeCredential(url.username)}:${decodeCredential(url.password)}`; + url.username = ""; + url.password = ""; + return new Headers({ authorization: `Basic ${base64(userinfo)}` }); +} + +/** + * Read one percent-encoded credential of a sink URL, as urlToHttpOptions() did for the old + * transport + * + * @param {string} value the credential as the URL holds it + * @returns {string} the decoded credential + */ +function decodeCredential(value: string): string { + try { + return decodeURIComponent(value); + } catch { + throw new TypeError("sink credentials must be percent-encoded values"); + } +} + +/** + * Encode credentials as UTF-8 bytes, the way RFC 7617 and the old transport did, with APIs the + * browser bundle can use as well + * + * @param {string} value the credentials to encode + * @returns {string} the base64 form + */ +function base64(value: string): string { + const utf8 = new TextEncoder().encode(value); + return btoa(Array.from(utf8, (byte) => String.fromCharCode(byte)).join("")); +} + +/** + * Check the headers a caller supplied, either for this transport or for a single send + * + * Fetch does the conversion, so every form its `HeadersInit` accepts arrives intact, a + * `Headers` from another realm included. That leaves an untyped value to Fetch's own coercion, + * which is why a caller's array joins on "," while the array of a Message, normalized by + * {@linkcode headerValueFrom}, joins on ", " + * + * @param {unknown} source the headers supplied by the caller, if any + * @returns {Headers|undefined} the headers, or undefined when none were supplied + */ +function headersFrom(source: unknown): Headers | undefined { + if (source === undefined || source === null) { + return undefined; + } + return new Headers(source as FetchHeadersInit); +} + +/** + * Copy headers onto a target, replacing any that are already there + * + * @param {Headers} target the headers being built + * @param {CloudEventHeaders|Headers} source the headers to copy, undefined ones are ignored + * @returns {void} + */ +function setHeaders(target: Headers, source?: CloudEventHeaders | Headers): void { + if (source === undefined) { + return; + } + if (source instanceof Headers) { + source.forEach((value, name) => target.set(name, value)); + return; + } + for (const [name, value] of Object.entries(source as Record)) { + const headerValue = headerValueFrom(value); + if (headerValue !== undefined) { + target.set(name, headerValue); + } + } +} + +/** + * Read a header value, whatever form it arrived in + * + * @param {unknown} value the value to read + * @returns {string|undefined} the value to send, undefined for a header which is not to be sent + */ +function headerValueFrom(value: unknown): string | undefined { + if (value === undefined || value === null) { + return undefined; + } + // several values join; an absent one is skipped + if (Array.isArray(value)) { + const values = value.filter((item) => item !== undefined && item !== null); + return values.length > 0 ? values.join(", ") : undefined; + } + return String(value); } diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts index 568f6ccc..e73b10e9 100644 --- a/test/integration/emitter_factory_test.ts +++ b/test/integration/emitter_factory_test.ts @@ -11,7 +11,7 @@ import request from "superagent"; import got from "got"; import CONSTANTS from "../../src/constants"; -import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor, httpTransport } +import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } from "../../src"; const DEFAULT_CE_CONTENT_TYPE = CONSTANTS.DEFAULT_CE_CONTENT_TYPE; @@ -101,8 +101,8 @@ describe("emitterFor() defaults", () => { }); }); -function setupMock(uri: string) { - nock(uri) +function setupMock() { + nock(sink) .post("/") .reply(function (uri: string, body: nock.Body) { // return the request body and the headers so they can be @@ -116,18 +116,7 @@ function setupMock(uri: string) { } describe("HTTP Transport Binding for emitterFactory", () => { - beforeEach(() => { setupMock(sink); }); - - describe("HTTPS builtin", () => { - testEmitterBinary(httpTransport(sink), "body"); - }); - - describe("HTTP builtin", () => { - setupMock("http://cloudevents.io"); - testEmitterBinary(httpTransport("http://cloudevents.io"), "body"); - setupMock("http://cloudevents.io"); - testEmitterStructured(httpTransport("http://cloudevents.io"), "body"); - }); + beforeEach(() => { setupMock(); }); describe("Axios", () => { testEmitterBinary(axiosEmitter, "data"); diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts new file mode 100644 index 00000000..cf140d43 --- /dev/null +++ b/test/integration/http_transport_test.ts @@ -0,0 +1,424 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import "mocha"; +import { rejects } from "assert"; +import { expect } from "chai"; +import { createServer, IncomingMessage, ServerResponse } from "http"; +import { AddressInfo } from "net"; +import { json, text } from "stream/consumers"; +import { runInNewContext } from "vm"; + +import { + CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, + Message, Mode, emitterFor, httpTransport, +} from "../../src"; +import { assertStructured, fixture } from "./emitter_factory_test"; + +type Handler = (request: IncomingMessage, response: ServerResponse) => void | Promise; + +// nock cannot intercept fetch, so these tests use a local server + +// a reserved port nothing serves +const UNREACHABLE_URL = "http://127.0.0.1:1"; + +// a sink which accepts every event, for tests that assert on the request rather than the response +const acceptEvent: Handler = (_request, response) => { + response.writeHead(204); + response.end(); +}; + +describe("Built-in HTTP transport", () => { + it("Sends a binary event and resolves with the response for 2xx", async () => { + let receivedBody: Record | undefined; + let receivedHeaders: IncomingMessage["headers"] | undefined; + + await withServer(async (request, response) => { + receivedHeaders = request.headers; + receivedBody = await json(request) as Record; + response.writeHead(202, { "x-request-id": "receipt-7" }); + response.end("accepted"); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/events`)); + const { response, body } = await emit(fixture, { headers: { "x-request-id": "order-123" } }); + + expect(response.status).to.equal(202); + expect(response.headers.get("x-request-id")).to.equal("receipt-7"); + expect(response.bodyUsed).to.equal(true); + expect(body).to.equal("accepted"); + expect(receivedHeaders?.["x-request-id"]).to.equal("order-123"); + expect(receivedHeaders?.["content-type"]).to.equal(CONSTANTS.DEFAULT_CONTENT_TYPE); + expect(receivedHeaders?.["ce-id"]).to.equal(fixture.id); + expect(receivedHeaders?.["ce-specversion"]).to.equal(fixture.specversion); + expect(receivedHeaders?.["ce-type"]).to.equal(fixture.type); + expect(receivedHeaders?.["ce-source"]).to.equal(fixture.source); + // the extensions of the fixture become ce-* headers + expect(receivedHeaders?.["ce-lunch"]).to.equal(fixture.lunch); + expect(receivedHeaders?.["ce-supper"]).to.equal(fixture.supper); + // an extension holding an object has no binary mode form; the old http.request + // transport sent the same String() of it, and only nock kept the object itself + expect(receivedHeaders?.["ce-snack"]).to.equal("[object Object]"); + expect(receivedBody?.lunchBreak).to.equal("noon"); + }); + }); + + it("Resolves with an empty body when the sink sends none", async () => { + await withServer(acceptEvent, async (url) => { + const emit = emitterFor(httpTransport(url)); + const { response, body } = await emit(fixture); + + expect(response.status).to.equal(204); + expect(body).to.equal(""); + }); + }); + + it("Sends to a sink given as a URL", async () => { + let receivedPath: string | undefined; + + await withServer((request, response) => { + receivedPath = request.url; + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(new URL(`${url}/events`))); + const { response, body } = await emit(fixture); + + expect(response.status).to.equal(204); + expect(body).to.equal(""); + expect(receivedPath).to.equal("/events"); + }); + }); + + it("Sends a structured event", async () => { + let received: Record | undefined; + + await withServer(async (request, response) => { + received = { + ...await json(request) as Record, + ...request.headers, + }; + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(url), { mode: Mode.STRUCTURED }); + await emit(fixture); + assertStructured(received as Record>); + }); + }); + + it("Keeps every value of a repeated response header", async () => { + await withServer((_request, response) => { + response.writeHead(202, { + "set-cookie": ["session=abc; Path=/", "tenant=store-42; Path=/"], + }); + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const { response } = await emit(fixture); + const headers = response.headers as Headers & { getSetCookie(): string[] }; + + expect(headers.getSetCookie()).to.deep.equal([ + "session=abc; Path=/", + "tenant=store-42; Path=/", + ]); + }); + }); + + it("Applies transport headers and lets per-send headers override them", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url, { + headers: new Headers({ + authorization: "Bearer transport-token", + "x-tenant-id": "store-42", + }), + })); + + await emit(fixture); + await emit(fixture, { + headers: { "x-tenant-id": "store-99", "x-request-id": "order-123" }, + }); + + expect(receivedHeaders[0]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[0]?.["x-tenant-id"]).to.equal("store-42"); + expect(receivedHeaders[1]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[1]?.["x-tenant-id"]).to.equal("store-99"); + expect(receivedHeaders[1]?.["x-request-id"]).to.equal("order-123"); + // a header neither layer set keeps the value the binding produced + expect(receivedHeaders[1]?.["ce-type"]).to.equal(fixture.type); + }); + }); + + it("Sends credentials from the sink URL as an authorization header", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const { host } = new URL(url); + const sendAs = (credentials: string) => emitterFor(httpTransport(`http://${credentials}@${host}`))(fixture); + + await sendAs("us%40er:p%40ss"); + await sendAs("user:p%C3%A4ss"); + await sendAs("user"); + // anything the caller sets wins + await emitterFor(httpTransport(`http://user:pass@${host}`, { + headers: { authorization: "Bearer transport-token" }, + }))(fixture); + await emitterFor(httpTransport(`http://user:pass@${host}`))(fixture, { + headers: { authorization: "Bearer per-send-token" }, + }); + + // the credentials are percent-decoded, as the old transport decoded them + expect(receivedHeaders[0]?.authorization).to.equal("Basic dXNAZXI6cEBzcw=="); + // non-ASCII credentials are encoded as UTF-8 + expect(receivedHeaders[1]?.authorization).to.equal("Basic dXNlcjpww6Rzcw=="); + // a URL may carry a user without a password + expect(receivedHeaders[2]?.authorization).to.equal("Basic dXNlcjo="); + expect(receivedHeaders[3]?.authorization).to.equal("Bearer transport-token"); + expect(receivedHeaders[4]?.authorization).to.equal("Bearer per-send-token"); + }); + }); + + it("Accepts Fetch header forms without losing their contents", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + const foreignHeaders = withForeignInterfacePrototype(new Headers({ "x-realm": "foreign" })); + expect(foreignHeaders).not.to.be.instanceOf(Headers); + const headerForms: Array<{ headers: FetchHeadersInit; name: string; value: string }> = [ + { headers: new Headers({ "x-headers": "native" }), name: "x-headers", value: "native" }, + { + headers: [["x-tuple", "store-42"], ["x-tuple", "store-99"]], + name: "x-tuple", + value: "store-42, store-99", + }, + { + headers: runInNewContext("({ 'x-record': 'another-realm' })") as FetchHeadersInit, + name: "x-record", + value: "another-realm", + }, + { headers: foreignHeaders, name: "x-realm", value: "foreign" }, + { + // Map is accepted by native Headers at runtime, although it is not in TypeScript's HeadersInit + headers: new Map([["x-map", "native"]]) as unknown as FetchHeadersInit, + name: "x-map", + value: "native", + }, + ]; + + await withServer(handler, async (url) => { + const emit = emitterFor(httpTransport(url)); + + for (const { headers } of headerForms) { + await emit(fixture, { headers }); + } + + headerForms.forEach(({ name, value }, index) => { + expect(receivedHeaders[index]?.[name]).to.equal(value); + }); + }); + }); + + it("Reads Message headers the same way as caller headers", async () => { + const { received: receivedHeaders, handler } = headerCollector(); + + await withServer(handler, async (url) => { + const send = httpTransport(url); + const message: Message = { + headers: { + "content-type": "application/json", + "x-tenant-id": null as unknown as string, + "x-request-id": ["order-123", "order-456"], + }, + body: `{"lunchBreak":"noon"}`, + }; + + await send(message); + await send({ + headers: new Headers({ "x-request-id": "native-headers" }) as unknown as CloudEventHeaders, + body: "event", + }); + + // a null value is skipped rather than sent as the string "null" + expect(receivedHeaders[0]?.["x-tenant-id"]).to.equal(undefined); + expect(receivedHeaders[0]?.["x-request-id"]).to.equal("order-123, order-456"); + expect(receivedHeaders[1]?.["x-request-id"]).to.equal("native-headers"); + }); + }); + + it("Passes standard Fetch options to fetch", async () => { + let receivedOptions: FetchRequestInit | undefined; + + await withFetch(async (_input, options) => { + receivedOptions = options; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com/orders", { + fetchOptions: { + cache: "no-store", + credentials: "include", + redirect: "error", + }, + })); + await emit(fixture); + }); + + expect(receivedOptions?.cache).to.equal("no-store"); + expect(receivedOptions?.credentials).to.equal("include"); + expect(receivedOptions?.redirect).to.equal("error"); + expect(receivedOptions?.method).to.equal("POST"); + }); + + it("Keeps the manual redirect when fetchOptions leaves redirect undefined", async () => { + let receivedOptions: FetchRequestInit | undefined; + + await withFetch(async (_input, options) => { + receivedOptions = options; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com/orders", { + // fetch reads an explicitly undefined redirect as absent, which would follow redirects + fetchOptions: { redirect: undefined }, + })); + await emit(fixture); + }); + + expect(receivedOptions?.redirect).to.equal("manual"); + }); + + // Fetch keeps POST only for 307 and 308, the others become a bodyless GET + for (const { statusCode, method, keepsBody } of [ + { statusCode: 301, method: "GET", keepsBody: false }, + { statusCode: 302, method: "GET", keepsBody: false }, + { statusCode: 303, method: "GET", keepsBody: false }, + { statusCode: 307, method: "POST", keepsBody: true }, + { statusCode: 308, method: "POST", keepsBody: true }, + ]) { + it(`Follows ${statusCode} with Fetch semantics when redirect is "follow"`, async () => { + let originalBody = ""; + let redirectedBody = "not received"; + let redirectedMethod: string | undefined; + + await withServer(async (request, response) => { + if (request.url === "/target") { + redirectedMethod = request.method; + redirectedBody = await text(request); + response.writeHead(204); + } else { + originalBody = await text(request); + response.writeHead(statusCode, { location: "/target" }); + } + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/start`, { + fetchOptions: { redirect: "follow" }, + })); + const { response, body } = await emit(fixture); + + expect(redirectedMethod).to.equal(method); + expect(redirectedBody).to.equal(keepsBody ? originalBody : ""); + expect(response.url).to.equal(`${url}/target`); + expect(body).to.equal(""); + }); + }); + } + + for (const statusCode of [400, 503]) { + it(`Resolves a ${statusCode} response without retrying it`, async () => { + let requestCount = 0; + await withServer((_request, response) => { + requestCount++; + response.writeHead(statusCode); + response.end("rejected"); + }, async (url) => { + const result = await emitterFor(httpTransport(url))(fixture); + expect(result.response.status).to.equal(statusCode); + expect(result.body).to.equal("rejected"); + expect(requestCount).to.equal(1); + }); + }); + } + + it("Rejects a per-send header that is not a valid header", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + // a TypeError, since nothing was sent + await rejects(emit(fixture, { headers: { "invalid header": "value" } }), TypeError); + }); + + it("Uses native Fetch coercion for untyped header values", async () => { + const legacyHeaders: CloudEventHeaders = { "x-tenant-id": ["store-42", "store-99"] }; + + await withServer((request, response) => { + expect(request.headers["x-tenant-id"]).to.equal("store-42,store-99"); + acceptEvent(request, response); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + await emit(fixture, { + + headers: legacyHeaders, + }); + }); + }); + + it("Validates the sink and transport options when the transport is created", () => { + expect(() => httpTransport("not a URL")).to.throw(TypeError); + expect(() => httpTransport("ftp://events.example.com")).to.throw(TypeError, "unsupported protocol ftp:"); + expect(() => httpTransport("https://a%zz@events.example.com")) + .to.throw(TypeError, "sink credentials must be percent-encoded values"); + expect(() => httpTransport("https://events.example.com", { + headers: { "invalid header": "value" }, + })).to.throw(TypeError); + expect(() => httpTransport("https://events.example.com", { + headers: new Headers({ "x-tenant-id": "store-42" }), + })).not.to.throw(); + }); +}); + +// Model a platform object from another realm: same behavior, different instanceof chain. +function withForeignInterfacePrototype(value: T): T { + const interfacePrototype = Object.getPrototypeOf(value); + const foreignPrototype = Object.create(Object.getPrototypeOf(interfacePrototype)); + Object.defineProperties(foreignPrototype, Object.getOwnPropertyDescriptors(interfacePrototype)); + Object.setPrototypeOf(value, foreignPrototype); + return value; +} + +// A sink which records the headers of every event it accepts. +function headerCollector(): { received: IncomingMessage["headers"][]; handler: Handler } { + const received: IncomingMessage["headers"][] = []; + return { + received, + handler: (request, response) => { + received.push(request.headers); + acceptEvent(request, response); + }, + }; +} + +async function withFetch(stub: typeof globalThis.fetch, run: () => Promise): Promise { + const originalFetch = globalThis.fetch; + globalThis.fetch = stub; + + try { + await run(); + } finally { + globalThis.fetch = originalFetch; + } +} + +export async function withServer(handler: Handler, run: (url: string) => Promise): Promise { + const server = createServer((request, response) => { + Promise.resolve(handler(request, response)).catch((error) => { + response.destroy(error); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + + try { + await run(`http://127.0.0.1:${port}`); + } finally { + await new Promise((resolve, reject) => { + server.close((error) => error ? reject(error) : resolve()); + server.closeAllConnections(); + }); + } +} diff --git a/webpack.config.js b/webpack.config.js index 1894ee50..d35f024b 100644 --- a/webpack.config.js +++ b/webpack.config.js @@ -7,9 +7,7 @@ module.exports = { }, resolve: { fallback: { - util: require.resolve("util/"), - http: false, - https: false + util: require.resolve("util/") }, }, plugins: [ From a81d69668b4fcbdd5146e5eef0c210aff8dda0e8 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:21:26 +0900 Subject: [PATCH 4/6] feat!: report HTTP delivery failures and typed response bodies Reject responses that did not accept the event and distinguish HTTP status failures from requests that never produced a response. Keep an unreadable body after a successful status separate from a failed delivery, so a caller does not resend an event the receiver has already accepted. Allow transport-wide and per-send response handlers, preserving their body types through emitterFor(). Include text and discard handlers and clean up the body after a handler fails. BREAKING CHANGE: non-2xx HTTP responses reject with HTTPTransportError. Successful responses resolve with either body or bodyError. Options.headers is checked by TypeScript, and built-in HTTP transport annotations use HTTPTransportFunction and HTTPEmitterFunction. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 107 +++++++++- README.md | 109 ++++++++-- src/index.ts | 19 +- src/transport/emitter.ts | 34 +++- src/transport/http/index.ts | 240 ++++++++++++++++++++--- test/integration/emitter_factory_test.ts | 19 +- test/integration/http_transport_test.ts | 214 ++++++++++++++++++-- 7 files changed, 664 insertions(+), 78 deletions(-) diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index d33bd8ae..26c47a0e 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -103,16 +103,81 @@ new CloudEvent({ type, source, data }).emit(); ### Upgrading From 10.x to 11.0 +In the 11.0.0 release, the built-in HTTP transport was rewritten on top of the Fetch API. + #### HTTP Transport -`httpTransport()` uses Fetch and resolves with `{ response, body }`, where -`response` is the native Fetch `Response` and `body` is its text, instead of -`{ body, headers }`. Access headers through `response.headers.get(name)` and -the status through `response.status`. +`httpTransport()` used to resolve with the response for every status code, so a receiver +that rejected an event looked just like one that accepted it. + +```js +import { CloudEvent, emitterFor, httpTransport } from "cloudevents"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); + +// resolved with { body, headers } whether the receiver returned 202 or 503 +const response = await emit(new CloudEvent({ type, source, data })); +``` + +A 2xx response now resolves with the native Fetch `Response` and its body as text, and +anything else rejects with an `HTTPTransportError`. + +```js +import { + CloudEvent, + emitterFor, + HTTPTransportError, + httpTransport, +} from "cloudevents"; + +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint")); + +try { + const { response, body } = await emit(new CloudEvent({ type, source, data })); + console.log(response.status, body); +} catch (error) { + if (error instanceof HTTPTransportError) { + console.error(error.kind, error.response?.status); + } + throw error; +} +``` + +The transport takes a second argument now, for headers, how the response body +is read and, under `fetchOptions`, the other Fetch options. Its headers are Fetch header +forms, so a `string[]` value becomes a `Headers` you `append()` to. A redirect is a +failed send unless `fetchOptions: { redirect: "follow" }` says otherwise, and a body +that cannot be read after a 2xx resolves with `bodyError` in place of `body`. +```js +const headers = new Headers(); +headers.append("accept", "application/cloudevents+json"); +headers.append("accept", "application/json"); +await emit(new CloudEvent({ type, source, data }), { headers }); +``` + +#### The Emitter Singleton + +A listener registered with `Emitter.on("cloudevent", emit)` now returns a promise which +rejects for a receiver that did not accept the event. `emitEvent()` awaits those promises +by default, so the rejection reaches whoever called `emit()` on the event: ```js -const { response, body } = await emit(event); -console.log(response.status, response.headers.get("x-request-id"), body); +try { + await new CloudEvent({ type, source, data }).emit(); +} catch (error) { + console.error(error.kind, error.response?.status); +} +``` + +With `ensureDelivery` turned off, the listeners run through `EventEmitter`, which discards +what they return. A rejection there becomes an unhandled promise rejection, which a `503` +could not produce before. Leave `ensureDelivery` alone, or catch inside the listener: + +```js +Emitter.on("cloudevent", (event) => emit(event).catch(reportFailedDelivery)); + +// listeners are not awaited, so nothing else can catch what they reject with +new CloudEvent({ type, source, data }).emit(false); ``` #### Proxies, Custom CAs, mTLS and Connection Pooling @@ -165,3 +230,33 @@ project, so TypeScript accepts `dispatcher` only where those are undici's. Again DOM `RequestInit` most projects compile with, pass it as `fetchOptions: { dispatcher } as FetchRequestInit` or reach for `setGlobalDispatcher()` instead. + +#### Emitter Options + +`Options` now declares `headers` instead of leaving it to an index +signature that resolved to `unknown`. Its default header type remains the +CloudEvent header record used by custom transports. The emitter returned from +`httpTransport()` specializes it to `FetchHeadersInit`. + +TypeScript now rejects spreading custom transport options into a client +configuration that types its own `headers`: + +```ts +// merge the headers of one send over the ones the binding produced +function sendWithAxios(message: Message, options?: Options) { + const { headers, ...rest } = options ?? {}; + return axios.post(url, message.body, { + headers: { ...message.headers, ...headers } as AxiosRequestHeaders, + ...rest, + }); +} +``` + +It also rejects the generic types as annotations for the built-in transport, whose +options are narrower. Use `HTTPTransportFunction` and `HTTPEmitterFunction`, or leave +the types to inference: + +```ts +const transport: HTTPTransportFunction = httpTransport(sink); +const emit: HTTPEmitterFunction = emitterFor(transport); +``` diff --git a/README.md b/README.md index 60a6874e..b3b46903 100644 --- a/README.md +++ b/README.md @@ -67,18 +67,74 @@ async function main() { main().catch(console.error); ``` -The transport resolves with the native Fetch `Response` and its body as text -for every HTTP status. It sends each event once. Redirects are not followed -unless `fetchOptions: { redirect: "follow" }` enables them; Fetch preserves the -CloudEvent POST and its body only for `307` and `308`. +A 2xx response resolves with the native Fetch `Response` and its body, read as +text by default. Anything else rejects with an `HTTPTransportError`, whose +`kind` separates a response that was not 2xx from a request that never reached +a response. -`httpTransport()` accepts transport-wide `headers` and other Fetch options under -`fetchOptions`. Headers passed to `emit()` apply to that event only and take -precedence over transport and CloudEvent headers. URL credentials become an -authorization header unless the caller overrides it. +The body of a response that was not 2xx is on the error as `error.body`. +`error.response` cannot be read a second time. If reading the body failed, +`error.body` is absent and the error carries a `cause`. -See the [API transition guide](./API_TRANSITION_GUIDE.md) for response access and -Node.js proxy, certificate and connection settings. +```js +import { HTTPTransportError } from "cloudevents"; + +async function main() { + try { + await emit(ce); + } catch (error) { + if (error instanceof HTTPTransportError) { + console.error(error.kind, error.response?.status, error.body); + } + throw error; + } +} + +main().catch(console.error); +``` + +`httpTransport()` also takes headers for every event, a `responseHandler` to read +the body another way (or `httpDiscardResponseHandler` to skip it), and the other +Fetch options under `fetchOptions`. Headers passed to `emit()` apply to that +event only. A response handler passed with an event overrides the transport's +handler, and its return value determines the response body type. + +```js +const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { + headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` }, + responseHandler: (response) => response.json(), +})); +``` + +A 2xx response means the receiver accepted the event, so a body that cannot be +read resolves with `{ response, bodyError }` in place of `{ response, body }`. +Only one of the two is present. Check which property the result has before using +it: + +```js +async function main() { + const result = await emit(ce); + if ("body" in result) { + console.log(result.body); + } else { + console.error(result.bodyError); + } +} + +main().catch(console.error); +``` + +Redirects are not followed by default, so each one is reported as a failed send. +On Node.js the `HTTPTransportError` carries the 3xx response; a browser returns +an opaque redirect instead, so `error.response.status` is `0` and the `location` +header cannot be read. Set `fetchOptions: { redirect: "follow" }` to follow +them, keeping in mind that Fetch preserves the CloudEvent POST and its body only +for `307` and `308` - it turns a `301`, `302` or `303` into a bodyless `GET`, +which drops the event. + +The [API transition guide](./API_TRANSITION_GUIDE.md) covers the header forms +the transport accepts and where a proxy or a custom CA goes now that +`http.globalAgent` no longer applies. If you prefer to use another transport mechanism for sending events over HTTP, you can use the `HTTP` binding to create a `Message` which @@ -128,20 +184,31 @@ You may also use the `Emitter` singleton to send your `CloudEvents`. ```js const { emitterFor, httpTransport, Mode, CloudEvent, Emitter } = require("cloudevents"); -// Create a CloudEvent emitter function to send events to our receiver -const emit = emitterFor(httpTransport("https://example.com/receiver")); - -// Use the emit() function to send a CloudEvent to its endpoint when a "cloudevent" event is emitted -// (see: https://nodejs.org/api/events.html#class-eventemitter) -Emitter.on("cloudevent", emit); - -... -// In any part of the code, calling `emit()` on a `CloudEvent` instance will send the event -new CloudEvent({ type, source, data }).emit(); +async function main() { + // Create a CloudEvent emitter function to send events to our receiver + const emit = emitterFor(httpTransport("https://example.com/receiver")); + + // Use the emit() function to send a CloudEvent to its endpoint when a + // "cloudevent" event is emitted + // (see: https://nodejs.org/api/events.html#class-eventemitter) + Emitter.on("cloudevent", emit); + + // Calling emit() on a CloudEvent instance sends it through every listener + await new CloudEvent({ + type: "com.example.order.created", + source: "/stores/store-42", + data: { orderId: "order-123" }, + }).emit(); +} -// You can also have several listeners to send the event to several endpoints +main().catch(console.error); ``` +You can also have several listeners to send the event to several endpoints. +`emit()` waits for all of them, and an endpoint which refuses the event fails +the whole call. See the [API transition guide](./API_TRANSITION_GUIDE.md) for +the details. + ## CloudEvent Objects All created `CloudEvent` objects are read-only. If you need to update a property or add a new extension to an existing cloud event object, you can use the `cloneWith` method. This will return a new `CloudEvent` with any update or new properties. For example: diff --git a/src/index.ts b/src/index.ts index 3c3d5bcd..df586fc3 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,12 +8,13 @@ import { ValidationError } from "./event/validation"; import { CloudEventV1, CloudEventV1Attributes } from "./event/interfaces"; import { - Options, TransportFunction, EmitterFunction, emitterFor, Emitter, + Options, TransportFunction, EmitterFunction, HTTPEmitterFunction, emitterFor, Emitter, } from "./transport/emitter"; import { - FetchHeadersInit, FetchRequestInit, HTTPTransportFunction, HTTPTransportOptions, - HTTPTransportResponse, httpTransport, -} from "./transport/http"; + FetchHeadersInit, FetchRequestInit, HTTPResponseHandler, HTTPTransportError, + HTTPTransportErrorDetails, HTTPTransportErrorKind, + HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions, + httpDiscardResponseHandler, httpTextResponseHandler, httpTransport } from "./transport/http"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -34,6 +35,9 @@ export { // From transport emitterFor, httpTransport, + httpTextResponseHandler, + httpDiscardResponseHandler, + HTTPTransportError, Emitter, // From Constants CONSTANTS @@ -54,10 +58,15 @@ export type { // From transport TransportFunction, EmitterFunction, + HTTPEmitterFunction, Options, FetchHeadersInit, FetchRequestInit, + HTTPResponseHandler, + HTTPTransportErrorDetails, + HTTPTransportErrorKind, HTTPTransportFunction, HTTPTransportOptions, - HTTPTransportResponse + HTTPTransportResponse, + HTTPTransportSendOptions }; diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index 7ddaf7d3..f2c854a0 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -4,15 +4,20 @@ */ import { CloudEvent } from "../event/cloudevent"; -import { HTTP, Message, Mode } from "../message"; +import { Headers as CloudEventHeaders, HTTP, Message, Mode } from "../message"; import { EventEmitter } from "events"; +import type { + FetchHeadersInit, HTTPTransportFunction, HTTPTransportResponse, HTTPTransportSendOptions, +} from "./http"; /** * Options is an additional, optional dictionary of options that may * be passed to an EmitterFunction and TransportFunction * @interface */ -export interface Options { +export interface Options { + /** Headers for this send, taking precedence over the ones the binding produced */ + headers?: THeaders; [key: string]: string | Record | unknown; } @@ -27,6 +32,17 @@ export interface EmitterFunction { (event: CloudEvent, options?: TOptions): Promise; } +/** + * An emitter backed by the built-in HTTP transport. A response handler passed with one event + * determines the body type for that call; otherwise the transport's default body type is used. + */ +export interface HTTPEmitterFunction + extends EmitterFunction, Options> { + ( + event: CloudEvent, options: HTTPTransportSendOptions, + ): Promise>; +} + /** * TransportFunction is an invokable interface provided to the emitterFactory. * A TransportFunction's responsiblity is to send a JSON encoded event Message @@ -39,6 +55,20 @@ export interface TransportFunction { } const emitterDefaults: Options = { binding: HTTP, mode: Mode.BINARY }; +/** + * Creates and returns an {@linkcode HTTPEmitterFunction} using the built-in HTTP + * transport returned by `httpTransport()`. The returned function converts a + * {@linkcode CloudEvent} into a {@linkcode Message} the same way the generic overload + * below does, and resolves with what the sink sent back, or rejects with an + * `HTTPTransportError`. + * + * @param {HTTPTransportFunction} fn the built-in HTTP transport to send events with + * @param { {Binding, Mode} } options network binding and message serialization options + * @returns {HTTPEmitterFunction} an emitter which resolves with the sink's response + */ +export function emitterFor( + fn: HTTPTransportFunction, options?: Options, +): HTTPEmitterFunction; /** * Creates and returns an {@linkcode EmitterFunction} using the supplied * {@linkcode TransportFunction}. The returned {@linkcode EmitterFunction} diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 1b5e9604..404e9427 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -12,40 +12,152 @@ export type FetchRequestInit = NonNullable[1 /** The header forms accepted by the Fetch implementation in the current environment */ export type FetchHeadersInit = NonNullable; -/** Options applied to every request sent by an HTTP transport */ -export interface HTTPTransportOptions { +/** Turns a Fetch response into the body value handed back by the HTTP transport */ +export type HTTPResponseHandler = (response: Response) => Promise; + +/** + * Read a response body as text, the default behavior of the HTTP transport + * + * @param {Response} response the response whose body should be read + * @returns {Promise} the response body as text + */ +export function httpTextResponseHandler(response: Response): Promise { + return response.text(); +} + +/** + * Discard a response body when the caller only needs the response metadata + * + * @param {Response} response the response whose body should be discarded + * @returns {Promise} completion after the body has been cancelled + */ +export async function httpDiscardResponseHandler(response: Response): Promise { + await response.body?.cancel(); +} + +/** + * Options applied to every request sent by an HTTP transport + */ +export interface HTTPTransportOptions { /** HTTP headers applied to every request, per-send headers take precedence */ headers?: FetchHeadersInit; /** Fetch options applied to every request, apart from the ones this transport controls */ fetchOptions?: Omit; + /** Reads every response body; defaults to {@linkcode httpTextResponseHandler} */ + responseHandler?: HTTPResponseHandler; } -/** The native Fetch response and its body, read as text */ -export interface HTTPTransportResponse { - response: Response; - body: string; +/** + * Options which replace the response handler for one send + */ +export interface HTTPTransportSendOptions extends Options { + /** Reads this response body instead of the handler configured for the transport */ + responseHandler: HTTPResponseHandler; } -/** A transport that returns an HTTP response */ -export type HTTPTransportFunction = TransportFunction; +/** What the built-in HTTP transport hands back for a send the sink accepted */ +export type HTTPTransportResponse = + | { + /** The native Fetch response, whose body has been handled */ + response: Response; + /** What the configured response handler returned */ + body: TBody; + bodyError?: never; + } + | { + /** The native Fetch response, whose status is still a successful 2xx */ + response: Response; + body?: never; + /** What the response handler threw after the sink accepted the event */ + bodyError: unknown; + }; /** - * Send an HTTP-bound CloudEvent with Fetch. The transport sends once and resolves with - * the native response and its body as text, for every HTTP status. Redirects are not - * followed unless fetchOptions.redirect enables them. Credentials in the sink URL become - * an authorization header, overridden by transport or per-send headers. + * An HTTP transport whose default response body type can be replaced for one send + */ +export interface HTTPTransportFunction + extends TransportFunction, Options> { + (message: Message, options: HTTPTransportSendOptions): Promise>; +} + +/** The result of attempting to handle a response body */ +type HandledResponse = + | { handled: true; body: unknown } + | { handled: false; error: unknown }; + +/** + * The failure category reported by {@linkcode HTTPTransportError} + * + * - `http-status`: the sink returned a response that was not 2xx + * - `network`: the request never produced a response, e.g. DNS, connection, or TLS failure + */ +export type HTTPTransportErrorKind = "http-status" | "network"; + +/** + * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind} + */ +export type HTTPTransportErrorDetails = Pick; + +/** + * A request failure reported by the built-in HTTP transport + */ +export class HTTPTransportError extends Error { + /** The failure category */ + readonly kind: HTTPTransportErrorKind; + /** The response the sink sent, present only for `http-status` */ + readonly response?: Response; + /** What the response handler returned, present when it succeeded for `http-status` */ + readonly body?: unknown; + /** The underlying Fetch or response handler failure */ + readonly cause?: unknown; + + constructor(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails = {}) { + super(errorMessage(kind, details)); + this.name = "HTTPTransportError"; + this.kind = kind; + this.response = details.response; + this.body = details.body; + this.cause = details.cause; + } +} + +/** + * httpTransport provides a simple HTTP Transport function, which can send a CloudEvent, + * encoded as a Message to the endpoint. The returned function can be used with emitterFor() + * to provide an event emitter, for example: + * + * ```js + * const emit = emitterFor(httpTransport("http://example.com")); + * emit(myCloudEvent) + * .then(({ response, body }) => console.log(response.status, body)) + * .catch(err => console.error(err.kind, err.response?.status)); + * ``` + * + * The event is sent once, without retries. A response that is not 2xx rejects with an + * {@linkcode HTTPTransportError} holding the response the sink sent back. Redirects are + * reported as errors unless `fetchOptions.redirect` says otherwise, since Fetch keeps the + * CloudEvent POST only for 307 and 308 - see the README for the details. + * + * Every response body is passed to the configured response handler. It is read as text by + * default; a handler passed with one send takes precedence over the transport's handler. + * A handler may also return `response.body` to hand the stream to the caller, who then + * has to cancel it if it is not read to the end. + * + * Credentials in _sink_ are sent as an `authorization` header, which the `headers` option + * and the headers of one send override. * * @param {string|URL} sink the destination endpoint for the event - * @param {HTTPTransportOptions} options headers and Fetch request options - * @returns {HTTPTransportFunction} a function that sends CloudEvents to the sink + * @param {HTTPTransportOptions} options headers, Fetch options and response behavior + * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_ */ -export function httpTransport( - sink: string | URL, options: HTTPTransportOptions = {}, -): HTTPTransportFunction { +export function httpTransport( + sink: string | URL, options: HTTPTransportOptions = {}, +): HTTPTransportFunction { const url = validateHTTPURL(sink); // fetch refuses a URL with credentials const sinkHeaders = credentialsFrom(url); const transportHeaders = headersFrom(options.headers); + const transportResponseHandler = responseHandlerFrom(options.responseHandler, httpTextResponseHandler); const fetchOptions: FetchRequestInit = { ...options.fetchOptions, // fetch reads an explicitly undefined redirect as absent, so the default comes last @@ -53,17 +165,81 @@ export function httpTransport( method: "POST", }; - return async (message: Message, sendOptions?: Options): Promise => { + const send = async ( + message: Message, sendOptions?: Options, + ): Promise> => { const headers = requestHeaders( sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), ); - const response = await fetch(url, { - ...fetchOptions, - headers, - body: message.body as BodyInit, - }); - return { response, body: await response.text() }; + const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler); + let response: Response; + try { + response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + }); + } catch (cause) { + throw new HTTPTransportError("network", { cause }); + } + + const handled = await handleResponse(response, responseHandler); + if (!response.ok) { + // the status is the failure; the handler result or error goes alongside it + throw new HTTPTransportError("http-status", handled.handled + ? { response, body: handled.body } + : { response, cause: handled.error }); + } + return handled.handled + ? { response, body: handled.body } + : { response, bodyError: handled.error }; }; + + // the overloads type the per-send handler this implementation cannot + return send as HTTPTransportFunction; +} + +/** + * Check a response handler and fall back to the handler from the enclosing transport + * + * @param {unknown} value the handler supplied at this layer, if any + * @param {HTTPResponseHandler} fallback the handler inherited from the transport or default + * @returns {HTTPResponseHandler} the handler to run + */ +function responseHandlerFrom( + value: unknown, fallback: HTTPResponseHandler, +): HTTPResponseHandler { + if (value === undefined || value === null) { + return fallback; + } + if (typeof value !== "function") { + throw new TypeError("options.responseHandler must be a function"); + } + return value as HTTPResponseHandler; +} + +/** + * Run the caller's response handler. A handler that resolves may leave its body unread on + * purpose; one that throws has the rest of its body cancelled, so a failed send does not + * hold a connection open. + * + * @param {Response} response the response to handle + * @param {HTTPResponseHandler} handler the selected handler + * @returns {HandledResponse} the handler's body value or thrown error + */ +async function handleResponse( + response: Response, handler: HTTPResponseHandler, +): Promise { + try { + return { handled: true, body: await handler(response) }; + } catch (error) { + try { + await httpDiscardResponseHandler(response); + } catch { + // best effort cleanup + } + return { handled: false, error }; + } } /** @@ -204,3 +380,19 @@ function headerValueFrom(value: unknown): string | undefined { } return String(value); } + +/** + * Build the message for an {@linkcode HTTPTransportError} + * + * @param {HTTPTransportErrorKind} kind the failure category + * @param {HTTPTransportErrorDetails} details what is known about the failure + * @returns {string} the error message + */ +function errorMessage(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails): string { + switch (kind) { + case "http-status": + return `HTTP transport received a non-2xx response: ${details.response?.status}`; + default: + return "HTTP transport request failed"; + } +} diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts index e73b10e9..52313346 100644 --- a/test/integration/emitter_factory_test.ts +++ b/test/integration/emitter_factory_test.ts @@ -11,8 +11,7 @@ import request from "superagent"; import got from "got"; import CONSTANTS from "../../src/constants"; -import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } - from "../../src"; +import { CloudEvent, HTTP, Message, Mode, Options, TransportFunction, emitterFor } from "../../src"; const DEFAULT_CE_CONTENT_TYPE = CONSTANTS.DEFAULT_CE_CONTENT_TYPE; const sink = "https://cloudevents.io/"; @@ -39,7 +38,12 @@ export const fixture = new CloudEvent({ }); function axiosEmitter(message: Message, options?: Options): Promise { - return axios.post(sink, message.body, { headers: message.headers as AxiosRequestHeaders, ...options }); + // per-send headers are merged over the ones the binding produced + const { headers, ...rest } = options ?? {}; + return axios.post(sink, message.body, { + headers: { ...message.headers, ...headers } as AxiosRequestHeaders, + ...rest, + }); } function superagentEmitter(message: Message, options?: Options): Promise { @@ -66,6 +70,15 @@ function gotEmitter(message: Message, options?: Options): Promise { } describe("emitterFor() defaults", () => { + it("Keeps CloudEvent header arrays in custom transport options", async () => { + const transport: TransportFunction = async (_message, options) => { + expect(options?.headers?.["x-tenant-id"]).to.deep.equal(["store-42", "store-99"]); + }; + const emit = emitterFor(transport); + + await emit(fixture, { headers: { "x-tenant-id": ["store-42", "store-99"] } }); + }); + it("Defaults to HTTP binding, binary mode", () => { function transport(message: Message): Promise { // A binary message will have the source attribute as a header diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index cf140d43..a0c83d17 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -12,8 +12,8 @@ import { json, text } from "stream/consumers"; import { runInNewContext } from "vm"; import { - CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, - Message, Mode, emitterFor, httpTransport, + CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, HTTPTransportError, + Message, Mode, emitterFor, httpDiscardResponseHandler, httpTextResponseHandler, httpTransport, } from "../../src"; import { assertStructured, fixture } from "./emitter_factory_test"; @@ -124,6 +124,79 @@ describe("Built-in HTTP transport", () => { }); }); + it("Reports an unreadable body without failing an accepted event", async () => { + // the sink accepts the event, then closes the connection before finishing its body + await withServer((_request, response) => { + response.writeHead(202); + response.write("partial", () => response.socket?.end()); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture); + + expect(result.response.status).to.equal(202); + expect("body" in result).to.equal(false); + expect("bodyError" in result).to.equal(true); + expect(result.bodyError).not.to.equal(undefined); + }); + }); + + it("Infers bodies from transport and per-send response handlers", async () => { + await withServer((_request, response) => { + response.writeHead(202); + response.end("42"); + }, async (url) => { + const emit = emitterFor(httpTransport(url, { + responseHandler: async (response) => ({ receipt: Number(await response.text()) }), + })); + + const configured = await emit(fixture); + expectType<{ receipt: number } | undefined>(configured.body); + // @ts-expect-error the transport handler determines the default body type + expectType(configured.body); + expect(configured.body).to.deep.equal({ receipt: 42 }); + + const overridden = await emit(fixture, { responseHandler: httpTextResponseHandler }); + expectType(overridden.body); + // @ts-expect-error the per-send handler replaces the transport body type for this call + expectType<{ receipt: number } | undefined>(overridden.body); + expect(overridden.body).to.equal("42"); + }); + }); + + it("Discards a body with the exported response handler", async () => { + await withServer((_request, response) => { + response.writeHead(202); + response.end("not needed"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { responseHandler: httpDiscardResponseHandler }); + + expectType(result.body); + expect(result.body).to.equal(undefined); + expect(result.response.bodyUsed).to.equal(true); + expect("bodyError" in result).to.equal(false); + }); + }); + + it("Returns what a custom response handler throws as bodyError for 2xx", async () => { + const handlerError = new Error("receipt was malformed"); + + await withServer((_request, response) => { + response.writeHead(202); + response.end("not a receipt"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { + responseHandler: async () => { throw handlerError; }, + }); + + expect(result.response.status).to.equal(202); + expect(result.response.bodyUsed).to.equal(true); + expect("body" in result).to.equal(false); + expect(result.bodyError).to.equal(handlerError); + }); + }); + it("Applies transport headers and lets per-send headers override them", async () => { const { received: receivedHeaders, handler } = headerCollector(); @@ -284,6 +357,72 @@ describe("Built-in HTTP transport", () => { expect(receivedOptions?.redirect).to.equal("manual"); }); + for (const statusCode of [400, 503]) { + it(`Reports a ${statusCode} response without retrying it`, async () => { + let requestCount = 0; + + await withServer((_request, response) => { + requestCount++; + response.writeHead(statusCode, { "x-request-id": `request-${statusCode}` }); + response.end(`status ${statusCode}`); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(statusCode); + expect(error.response?.headers.get("x-request-id")).to.equal(`request-${statusCode}`); + expect(error.body).to.equal(`status ${statusCode}`); + expect(error.response?.bodyUsed).to.equal(true); + expect(error.cause).to.equal(undefined); + expect(requestCount).to.equal(1); + }); + }); + } + + it("Applies a custom response handler to a non-2xx body", async () => { + await withServer((_request, response) => { + response.writeHead(422, { "content-type": "application/json" }); + response.end(`{"message":"invalid event"}`); + }, async (url) => { + const emit = emitterFor(httpTransport(url, { + responseHandler: async (response) => JSON.parse(await response.text()) as { message: string }, + })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(422); + expect(error.body).to.deep.equal({ message: "invalid event" }); + expect(error.cause).to.equal(undefined); + }); + }); + + for (const statusCode of [301, 302, 303]) { + it(`Returns ${statusCode} as an error by default instead of following it`, async () => { + let requestCount = 0; + let targetRequestCount = 0; + + await withServer((request, response) => { + requestCount++; + if (request.url === "/target") { + targetRequestCount++; + response.writeHead(204); + } else { + response.writeHead(statusCode, { location: "/target" }); + } + response.end(); + }, async (url) => { + const emit = emitterFor(httpTransport(`${url}/start`)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(statusCode); + expect(requestCount).to.equal(1); + expect(targetRequestCount).to.equal(0); + }); + }); + } + // Fetch keeps POST only for 307 and 308, the others become a bodyless GET for (const { statusCode, method, keepsBody } of [ { statusCode: 301, method: "GET", keepsBody: false }, @@ -321,21 +460,26 @@ describe("Built-in HTTP transport", () => { }); } - for (const statusCode of [400, 503]) { - it(`Resolves a ${statusCode} response without retrying it`, async () => { - let requestCount = 0; - await withServer((_request, response) => { - requestCount++; - response.writeHead(statusCode); - response.end("rejected"); - }, async (url) => { - const result = await emitterFor(httpTransport(url))(fixture); - expect(result.response.status).to.equal(statusCode); - expect(result.body).to.equal("rejected"); - expect(requestCount).to.equal(1); - }); + it("Reports a non-2xx status whose body cannot be read", async () => { + // the sink starts an error body, then closes the connection before finishing it + await withServer((_request, response) => { + response.writeHead(503); + response.write("partial", () => response.socket?.end()); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("http-status"); + expect(error.response?.status).to.equal(503); + expect(error.body).to.equal(undefined); + expect(error.cause).not.to.equal(undefined); }); - } + }); + + it("Rejects a per-send response handler that is not a function", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + await rejects(emit(fixture, { responseHandler: "text" }), TypeError); + }); it("Rejects a per-send header that is not a valid header", async () => { const emit = emitterFor(httpTransport(UNREACHABLE_URL)); @@ -352,12 +496,34 @@ describe("Built-in HTTP transport", () => { }, async (url) => { const emit = emitterFor(httpTransport(url)); await emit(fixture, { - + // @ts-expect-error built-in HTTP accepts FetchHeadersInit, not CloudEventHeaders arrays headers: legacyHeaders, }); }); }); + it("Reports a network failure with its cause", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("network"); + expect(error.response).to.equal(undefined); + expect(error.cause).not.to.equal(undefined); + }); + + it("Reports a synchronous Fetch failure with its cause", async () => { + const cause = new Error("Fetch failed before returning a promise"); + + await withFetch(() => { throw cause; }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com")); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("network"); + expect(error.response).to.equal(undefined); + expect(error.cause).to.equal(cause); + }); + }); + it("Validates the sink and transport options when the transport is created", () => { expect(() => httpTransport("not a URL")).to.throw(TypeError); expect(() => httpTransport("ftp://events.example.com")).to.throw(TypeError, "unsupported protocol ftp:"); @@ -369,9 +535,23 @@ describe("Built-in HTTP transport", () => { expect(() => httpTransport("https://events.example.com", { headers: new Headers({ "x-tenant-id": "store-42" }), })).not.to.throw(); + expect(() => httpTransport("https://events.example.com", { + responseHandler: "text" as never, + })).to.throw(TypeError, "options.responseHandler must be a function"); }); }); +async function transportErrorFrom(emitted: Promise): Promise { + const error = await emitted.then(() => undefined, (reason: unknown) => reason); + expect(error).to.be.instanceOf(HTTPTransportError); + return error as HTTPTransportError; +} + +// Compile-time assertion. +function expectType(value: T): void { + void value; +} + // Model a platform object from another realm: same behavior, different instanceof chain. function withForeignInterfacePrototype(value: T): T { const interfacePrototype = Object.getPrototypeOf(value); From da558bb114c6c3393ff54df250a7bf321be5ada1 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:22:25 +0900 Subject: [PATCH 5/6] feat!: add abort signals to HTTP transport Combine transport-wide and per-send abort signals while fetching and handling a response. Preserve the first abort reason, support signals from another realm and detach subscriptions when the operation finishes. Keep the shared signal implementation in its final module so emitter wrappers can reuse it without moving transport code in a later feature. BREAKING CHANGE: Options.signal is checked as an AbortSignal. A cancelled HTTP request rejects with HTTPTransportError of kind aborted; a body failure after a successful status remains a successful delivery result. Signed-off-by: kahirokunn --- API_TRANSITION_GUIDE.md | 9 +- README.md | 20 +- src/transport/emitter.ts | 2 + src/transport/http/index.ts | 62 +++--- src/transport/signal.ts | 161 ++++++++++++++++ test/integration/http_transport_test.ts | 240 ++++++++++++++++++++++++ 6 files changed, 463 insertions(+), 31 deletions(-) create mode 100644 src/transport/signal.ts diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md index 26c47a0e..16738f85 100644 --- a/API_TRANSITION_GUIDE.md +++ b/API_TRANSITION_GUIDE.md @@ -143,11 +143,14 @@ try { } ``` -The transport takes a second argument now, for headers, how the response body +The transport takes a second argument now, for headers, a signal, how the response body is read and, under `fetchOptions`, the other Fetch options. Its headers are Fetch header forms, so a `string[]` value becomes a `Headers` you `append()` to. A redirect is a failed send unless `fetchOptions: { redirect: "follow" }` says otherwise, and a body -that cannot be read after a 2xx resolves with `bodyError` in place of `body`. +that cannot be read after a 2xx resolves with `bodyError` in place of `body`. The +signal given to `httpTransport()` lasts as long as the transport, so a per-request +deadline belongs on the signal passed with the event. + ```js const headers = new Headers(); headers.append("accept", "application/cloudevents+json"); @@ -233,7 +236,7 @@ DOM `RequestInit` most projects compile with, pass it as #### Emitter Options -`Options` now declares `headers` instead of leaving it to an index +`Options` now declares `headers` and `signal` instead of leaving them to an index signature that resolved to `unknown`. Its default header type remains the CloudEvent header record used by custom transports. The emitter returned from `httpTransport()` specializes it to `FetchHeadersInit`. diff --git a/README.md b/README.md index b3b46903..94b2a4da 100644 --- a/README.md +++ b/README.md @@ -69,8 +69,8 @@ main().catch(console.error); A 2xx response resolves with the native Fetch `Response` and its body, read as text by default. Anything else rejects with an `HTTPTransportError`, whose -`kind` separates a response that was not 2xx from a request that never reached -a response. +`kind` separates a response that was not 2xx from an abort and from a request +that never reached a response. The body of a response that was not 2xx is on the error as `error.body`. `error.response` cannot be read a second time. If reading the body failed, @@ -93,17 +93,23 @@ async function main() { main().catch(console.error); ``` -`httpTransport()` also takes headers for every event, a `responseHandler` to read -the body another way (or `httpDiscardResponseHandler` to skip it), and the other -Fetch options under `fetchOptions`. Headers passed to `emit()` apply to that -event only. A response handler passed with an event overrides the transport's -handler, and its return value determines the response body type. +`httpTransport()` also takes the headers for every event, an `AbortSignal`, a +`responseHandler` to read the body another way (or `httpDiscardResponseHandler` +to skip it), and the remaining Fetch options under `fetchOptions`. Headers +passed to `emit()` apply to that event only, and there is no timeout option: a +deadline is an `AbortSignal`. ```js const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", { headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` }, responseHandler: (response) => response.json(), })); + +async function main() { + await emit(ce, { signal: AbortSignal.timeout(10000) }); +} + +main().catch(console.error); ``` A 2xx response means the receiver accepted the event, so a body that cannot be diff --git a/src/transport/emitter.ts b/src/transport/emitter.ts index f2c854a0..601e7880 100644 --- a/src/transport/emitter.ts +++ b/src/transport/emitter.ts @@ -16,6 +16,8 @@ import type { * @interface */ export interface Options { + /** Aborts the send, for transports that support cancellation */ + signal?: AbortSignal; /** Headers for this send, taking precedence over the ones the binding produced */ headers?: THeaders; [key: string]: string | Record | unknown; diff --git a/src/transport/http/index.ts b/src/transport/http/index.ts index 404e9427..0d490646 100644 --- a/src/transport/http/index.ts +++ b/src/transport/http/index.ts @@ -5,6 +5,7 @@ import { Headers as CloudEventHeaders, Message } from "../../message"; import { Options, TransportFunction } from "../emitter"; +import { combineSignals, signalFrom } from "../signal"; /** The request options accepted by the Fetch implementation in the current environment */ export type FetchRequestInit = NonNullable[1]>; @@ -41,6 +42,11 @@ export async function httpDiscardResponseHandler(response: Response): Promise { /** HTTP headers applied to every request, per-send headers take precedence */ headers?: FetchHeadersInit; + /** + * Aborts every request sent by this transport, combined with the per-send signal. It lasts + * as long as the transport does, so set a per-request deadline on the per-send signal + */ + signal?: AbortSignal; /** Fetch options applied to every request, apart from the ones this transport controls */ fetchOptions?: Omit; /** Reads every response body; defaults to {@linkcode httpTextResponseHandler} */ @@ -89,9 +95,10 @@ type HandledResponse = * The failure category reported by {@linkcode HTTPTransportError} * * - `http-status`: the sink returned a response that was not 2xx + * - `aborted`: a signal supplied by the caller aborted the request * - `network`: the request never produced a response, e.g. DNS, connection, or TLS failure */ -export type HTTPTransportErrorKind = "http-status" | "network"; +export type HTTPTransportErrorKind = "http-status" | "aborted" | "network"; /** * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind} @@ -108,7 +115,7 @@ export class HTTPTransportError extends Error { readonly response?: Response; /** What the response handler returned, present when it succeeded for `http-status` */ readonly body?: unknown; - /** The underlying Fetch or response handler failure */ + /** The underlying failure, e.g. what fetch threw or the reason carried by a signal */ readonly cause?: unknown; constructor(kind: HTTPTransportErrorKind, details: HTTPTransportErrorDetails = {}) { @@ -141,13 +148,14 @@ export class HTTPTransportError extends Error { * Every response body is passed to the configured response handler. It is read as text by * default; a handler passed with one send takes precedence over the transport's handler. * A handler may also return `response.body` to hand the stream to the caller, who then - * has to cancel it if it is not read to the end. + * has to cancel it if it is not read to the end, since the abort signals no longer apply + * once the handler has resolved. * * Credentials in _sink_ are sent as an `authorization` header, which the `headers` option * and the headers of one send override. * * @param {string|URL} sink the destination endpoint for the event - * @param {HTTPTransportOptions} options headers, Fetch options and response behavior + * @param {HTTPTransportOptions} options headers, Fetch options, abort and response behavior * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_ */ export function httpTransport( @@ -156,6 +164,7 @@ export function httpTransport( const url = validateHTTPURL(sink); // fetch refuses a URL with credentials const sinkHeaders = credentialsFrom(url); + const transportSignal = signalFrom(options.signal); const transportHeaders = headersFrom(options.headers); const transportResponseHandler = responseHandlerFrom(options.responseHandler, httpTextResponseHandler); const fetchOptions: FetchRequestInit = { @@ -168,31 +177,40 @@ export function httpTransport( const send = async ( message: Message, sendOptions?: Options, ): Promise> => { + const sendSignal = signalFrom(sendOptions?.signal); const headers = requestHeaders( sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers), ); const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler); - let response: Response; + const combinedSignal = combineSignals(transportSignal, sendSignal); try { - response = await fetch(url, { - ...fetchOptions, - headers, - body: message.body as BodyInit, - }); - } catch (cause) { - throw new HTTPTransportError("network", { cause }); - } + let response: Response; + try { + response = await fetch(url, { + ...fetchOptions, + headers, + body: message.body as BodyInit, + signal: combinedSignal.signal, + }); + } catch (cause) { + throw combinedSignal.signal?.aborted + ? new HTTPTransportError("aborted", { cause: combinedSignal.signal.reason ?? cause }) + : new HTTPTransportError("network", { cause }); + } - const handled = await handleResponse(response, responseHandler); - if (!response.ok) { - // the status is the failure; the handler result or error goes alongside it - throw new HTTPTransportError("http-status", handled.handled + const handled = await handleResponse(response, responseHandler); + if (!response.ok) { + // the status is the failure; the handler result or error goes alongside it + throw new HTTPTransportError("http-status", handled.handled + ? { response, body: handled.body } + : { response, cause: handled.error }); + } + return handled.handled ? { response, body: handled.body } - : { response, cause: handled.error }); + : { response, bodyError: handled.error }; + } finally { + combinedSignal.dispose(); } - return handled.handled - ? { response, body: handled.body } - : { response, bodyError: handled.error }; }; // the overloads type the per-send handler this implementation cannot @@ -392,6 +410,8 @@ function errorMessage(kind: HTTPTransportErrorKind, details: HTTPTransportErrorD switch (kind) { case "http-status": return `HTTP transport received a non-2xx response: ${details.response?.status}`; + case "aborted": + return "HTTP transport request was aborted"; default: return "HTTP transport request failed"; } diff --git a/src/transport/signal.ts b/src/transport/signal.ts new file mode 100644 index 00000000..14dd50a2 --- /dev/null +++ b/src/transport/signal.ts @@ -0,0 +1,161 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +/** + * Check a signal supplied by a caller, whether it was given to a transport, to a single send, + * or to an emitter wrapper, by its platform brand rather than the interface prototype of this + * realm + * + * @param {unknown} value the signal supplied by the caller, if any + * @returns {AbortSignal|undefined} the signal, or undefined when none was supplied + */ +export function signalFrom(value: unknown): AbortSignal | undefined { + if (value === undefined || value === null) { + return undefined; + } + try { + // the getter checks the AbortSignal platform brand and accepts signals from another realm + Reflect.get(AbortSignal.prototype, "aborted", value); + } catch { + throw new TypeError("options.signal must be an AbortSignal"); + } + return value as AbortSignal; +} + +/** + * Read the conventional signal from otherwise transport-specific send options + * + * @param {unknown} options the options passed to one emitter invocation + * @returns {AbortSignal|undefined} the validated signal, when one was supplied + */ +export function signalFromOptions(options: unknown): AbortSignal | undefined { + if (typeof options !== "object" || options === null) { + return undefined; + } + return signalFrom((options as { signal?: unknown }).signal); +} + +/** An abort signal scoped to one operation, and the cleanup which ends that scope */ +export interface CombinedSignal { + signal?: AbortSignal; + dispose: () => void; +} + +/** Subscribers sharing the one platform listener attached to an abort signal */ +interface AbortFanout { + subscribers: Set<() => void>; + listener: () => void; + listening: boolean; +} + +/** Active fan-outs, weakly keyed so an otherwise unused signal can still be collected */ +const abortFanouts = new WeakMap(); + +/** The relay for an operation which has no signal to relay */ +const NO_ABORT: CombinedSignal = { signal: undefined, dispose: () => undefined }; + +/** + * Create and attach the shared platform listener for one signal + * + * @param {AbortSignal} source the signal which owns the listener + * @returns {AbortFanout} the shared subscriber collection + */ +function createAbortFanout(source: AbortSignal): AbortFanout { + const subscribers = new Set<() => void>(); + const fanout: AbortFanout = { + subscribers, + listening: true, + listener: () => { + const current = Array.from(subscribers); + subscribers.clear(); + fanout.listening = false; + abortFanouts.delete(source); + current.forEach((callback) => callback()); + }, + }; + + abortFanouts.set(source, fanout); + source.addEventListener("abort", fanout.listener, { once: true }); + return fanout; +} + +/** + * Subscribe to a signal through one shared platform listener, however many operations use it + * + * @param {AbortSignal} source the signal whose abort should be relayed + * @param {Function} subscriber one operation's abort callback + * @returns {Function} an idempotent function which removes this subscription + */ +function subscribeAbort(source: AbortSignal, subscriber: () => void): () => void { + const fanout = abortFanouts.get(source) ?? createAbortFanout(source); + + fanout.subscribers.add(subscriber); + let disposed = false; + + return (): void => { + if (disposed) { + return; + } + disposed = true; + fanout.subscribers.delete(subscriber); + + if (fanout.listening && fanout.subscribers.size === 0) { + fanout.listening = false; + source.removeEventListener("abort", fanout.listener); + abortFanouts.delete(source); + } + }; +} + +/** + * Combine abort signals behind a disposable relay, so they apply only while an operation is + * running and do not retain a long-lived source after that work ends. + * + * `AbortSignal.any()` is not used here. Node.js 24 validates its arguments with `instanceof`, + * so it rejects the signals from another realm that {@linkcode signalFrom} accepts, with + * `The "signals[0]" argument must be an instance of AbortSignal`. `addEventListener()` works + * across realms on every supported version, so the sources are relayed through + * {@linkcode subscribeAbort}, which also keeps one listener per source however many operations + * share it. + * + * @param {Array} candidates the signals which can abort the operation + * @returns {CombinedSignal} the combined signal and an idempotent cleanup function + */ +export function combineSignals(...candidates: Array): CombinedSignal { + const sources = Array.from(new Set(candidates)) + .filter((signal): signal is AbortSignal => signal !== undefined); + if (sources.length === 0) { + return NO_ABORT; + } + + const controller = new AbortController(); + const unsubscribe: Array<() => void> = []; + let disposed = false; + + const dispose = (): void => { + if (disposed) { + return; + } + disposed = true; + unsubscribe.forEach((remove) => remove()); + unsubscribe.length = 0; + }; + + const abortFrom = (source: AbortSignal): void => { + controller.abort(source.reason); + dispose(); + }; + + const alreadyAborted = sources.find((source) => source.aborted); + if (alreadyAborted) { + abortFrom(alreadyAborted); + return { signal: controller.signal, dispose }; + } + + sources.forEach((source) => { + unsubscribe.push(subscribeAbort(source, () => abortFrom(source))); + }); + return { signal: controller.signal, dispose }; +} diff --git a/test/integration/http_transport_test.ts b/test/integration/http_transport_test.ts index a0c83d17..650770f1 100644 --- a/test/integration/http_transport_test.ts +++ b/test/integration/http_transport_test.ts @@ -6,6 +6,7 @@ import "mocha"; import { rejects } from "assert"; import { expect } from "chai"; +import { getEventListeners } from "events"; import { createServer, IncomingMessage, ServerResponse } from "http"; import { AddressInfo } from "net"; import { json, text } from "stream/consumers"; @@ -30,6 +31,13 @@ const acceptEvent: Handler = (_request, response) => { response.end(); }; +// a sink which accepts the event, then finishes its body a moment later +const streamEvent: Handler = (_request, response) => { + response.writeHead(202); + response.write("stream"); + setTimeout(() => response.end("ed"), 50); +}; + describe("Built-in HTTP transport", () => { it("Sends a binary event and resolves with the response for 2xx", async () => { let receivedBody: Record | undefined; @@ -178,6 +186,23 @@ describe("Built-in HTTP transport", () => { }); }); + it("Lets a custom response handler hand its stream to the caller", async () => { + await withServer(streamEvent, async (url) => { + const controller = new AbortController(); + const emit = emitterFor(httpTransport(url)); + const result = await emit(fixture, { + signal: controller.signal, + responseHandler: async (response) => response.body, + }); + + expectType | null | undefined>(result.body); + expect(result.response.bodyUsed).to.equal(false); + // after the handler returns the stream, cancelling it is up to the caller + controller.abort(); + expect(await new Response(result.body).text()).to.equal("streamed"); + }); + }); + it("Returns what a custom response handler throws as bodyError for 2xx", async () => { const handlerError = new Error("receipt was malformed"); @@ -460,6 +485,47 @@ describe("Built-in HTTP transport", () => { }); } + it("Reports a caller timeout as an abort, with the reason it carries", async () => { + await withServer(() => undefined, async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: AbortSignal.timeout(50) })); + + expect(error.kind).to.equal("aborted"); + expect(error.response).to.equal(undefined); + expect((error.cause as Error)?.name).to.equal("TimeoutError"); + }); + }); + + it("Returns an abort during a 2xx body as bodyError", async () => { + const controller = new AbortController(); + let markHandlerStarted = (): void => undefined; + const handlerStarted = new Promise((resolve) => { + markHandlerStarted = resolve; + }); + + await withServer((_request, response) => { + response.writeHead(202); + response.write("partial"); + }, async (url) => { + const emit = emitterFor(httpTransport(url)); + const emitted = emit(fixture, { + signal: controller.signal, + responseHandler: async (response) => { + markHandlerStarted(); + return response.text(); + }, + }); + + await handlerStarted; + controller.abort(new DOMException("deadline exceeded", "TimeoutError")); + const result = await emitted; + + expect(result.response.status).to.equal(202); + expect("body" in result).to.equal(false); + expect((result.bodyError as Error).name).to.equal("TimeoutError"); + }); + }); + it("Reports a non-2xx status whose body cannot be read", async () => { // the sink starts an error body, then closes the connection before finishing it await withServer((_request, response) => { @@ -476,6 +542,177 @@ describe("Built-in HTTP transport", () => { }); }); + it("Reports an abort from the transport signal", async () => { + const controller = new AbortController(); + + await withServer(() => controller.abort(new Error("shutting down")), async (url) => { + const emit = emitterFor(httpTransport(url, { signal: controller.signal })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("aborted"); + expect(error.response).to.equal(undefined); + expect((error.cause as Error).message).to.equal("shutting down"); + }); + }); + + it("Reports an abort from a per-send signal", async () => { + const controller = new AbortController(); + + await withServer(() => controller.abort(), async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal: controller.signal })); + + expect(error.kind).to.equal("aborted"); + expect(error.cause).not.to.equal(undefined); + }); + }); + + it("Accepts an AbortSignal whose interface prototype comes from another realm", async () => { + const controller = new AbortController(); + const signal = withForeignInterfacePrototype(controller.signal); + expect(signal).not.to.be.instanceOf(AbortSignal); + + await withServer(() => controller.abort(new Error("foreign signal gave up")), async (url) => { + const emit = emitterFor(httpTransport(url)); + const error = await transportErrorFrom(emit(fixture, { signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal("foreign signal gave up"); + }); + }); + + it("Does not send when the signal is already aborted", async () => { + let requestCount = 0; + const controller = new AbortController(); + controller.abort(); + + await withServer(() => { requestCount++; }, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: controller.signal })); + const error = await transportErrorFrom(emit(fixture)); + + expect(error.kind).to.equal("aborted"); + expect(requestCount).to.equal(0); + }); + }); + + it("Uses the first reason when both signals are already aborted", async () => { + const transport = new AbortController(); + const send = new AbortController(); + transport.abort(new Error("transport stopped first")); + send.abort(new Error("send stopped second")); + + const emit = emitterFor(httpTransport(UNREACHABLE_URL, { signal: transport.signal })); + const error = await transportErrorFrom(emit(fixture, { signal: send.signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal("transport stopped first"); + }); + + it("Sends with a transport signal and a per-send signal set at once", async () => { + const transport = new AbortController(); + + await withServer(acceptEvent, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const { response } = await emit(fixture, { signal: new AbortController().signal }); + + expect(response.status).to.equal(204); + expect(transport.signal.aborted).to.equal(false); + }); + }); + + it("Reports an abort from either signal when both are set", async () => { + for (const abortedSignal of ["transport", "per-send"]) { + const transport = new AbortController(); + const send = new AbortController(); + const aborting = abortedSignal === "transport" ? transport : send; + + await withServer(() => aborting.abort(new Error(`${abortedSignal} gave up`)), async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const error = await transportErrorFrom(emit(fixture, { signal: send.signal })); + + expect(error.kind).to.equal("aborted"); + expect((error.cause as Error).message).to.equal(`${abortedSignal} gave up`); + }); + } + }); + + it("Shares one source listener between concurrent sends and removes it afterward", async () => { + const controller = new AbortController(); + let releaseRequests = (): void => undefined; + const requestsMayComplete = new Promise((resolve) => { + releaseRequests = resolve; + }); + + await withFetch(async () => { + await requestsMayComplete; + return new Response(null, { status: 204 }); + }, async () => { + const emit = emitterFor(httpTransport("https://events.example.com", { signal: controller.signal })); + const emitted = Array.from({ length: 20 }, () => emit(fixture)); + + try { + expect(getEventListeners(controller.signal, "abort")).to.have.length(1); + releaseRequests(); + await Promise.all(emitted); + expect(getEventListeners(controller.signal, "abort")).to.have.length(0); + } finally { + releaseRequests(); + await Promise.allSettled(emitted); + } + }); + }); + + it("Fans a shared transport abort out to every concurrent send", async () => { + const controller = new AbortController(); + const reason = new Error("transport shutting down"); + + await withFetch(async (_input, options) => new Promise((_resolve, reject) => { + const signal = options?.signal; + if (!signal) { + reject(new Error("expected a combined abort signal")); + return; + } + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }), async () => { + const emit = emitterFor(httpTransport("https://events.example.com", { signal: controller.signal })); + const emitted = Array.from({ length: 20 }, () => transportErrorFrom(emit(fixture))); + + try { + expect(getEventListeners(controller.signal, "abort")).to.have.length(1); + controller.abort(reason); + const errors = await Promise.all(emitted); + errors.forEach((error) => { + expect(error.kind).to.equal("aborted"); + expect(error.cause).to.equal(reason); + }); + expect(getEventListeners(controller.signal, "abort")).to.have.length(0); + } finally { + controller.abort(reason); + await Promise.allSettled(emitted); + } + }); + }); + + it("Stops applying a transport signal once the send has resolved", async () => { + const transport = new AbortController(); + + await withServer(streamEvent, async (url) => { + const emit = emitterFor(httpTransport(url, { signal: transport.signal })); + const result = await emit(fixture, { + responseHandler: async (response) => response.body, + }); + + // the transport signal no longer reaches a body the handler handed to the caller + transport.abort(); + expect(await new Response(result.body).text()).to.equal("streamed"); + }); + }); + + it("Rejects a per-send signal that is not an AbortSignal", async () => { + const emit = emitterFor(httpTransport(UNREACHABLE_URL)); + await rejects(emit(fixture, { signal: "not a signal" as unknown as AbortSignal }), TypeError); + }); + it("Rejects a per-send response handler that is not a function", async () => { const emit = emitterFor(httpTransport(UNREACHABLE_URL)); await rejects(emit(fixture, { responseHandler: "text" }), TypeError); @@ -535,6 +772,9 @@ describe("Built-in HTTP transport", () => { expect(() => httpTransport("https://events.example.com", { headers: new Headers({ "x-tenant-id": "store-42" }), })).not.to.throw(); + expect(() => httpTransport("https://events.example.com", { + signal: "not a signal" as unknown as AbortSignal, + })).to.throw(TypeError, "options.signal must be an AbortSignal"); expect(() => httpTransport("https://events.example.com", { responseHandler: "text" as never, })).to.throw(TypeError, "options.responseHandler must be a function"); From a94656d12668bbfe0e6ef76183eb3ce3a6760077 Mon Sep 17 00:00:00 2001 From: kahirokunn Date: Tue, 13 Oct 2026 13:24:42 +0900 Subject: [PATCH 6/6] feat: add configurable retries for emitters Keep the HTTP transport single-attempt and let applications opt into retrying temporary failures with a reusable emitter wrapper. Preserve the event and send options across attempts, bound backoff waits and honor Retry-After for 429 and 503 responses. Default retries recognize temporary HTTP failures, while custom transports can supply their own predicate and delay. Aborted sends are not retried, and a per-send signal can interrupt the wait before another attempt. Signed-off-by: kahirokunn --- README.md | 50 +++++ src/index.ts | 8 + src/transport/retry.ts | 240 ++++++++++++++++++++++ test/integration/retry_test.ts | 355 +++++++++++++++++++++++++++++++++ 4 files changed, 653 insertions(+) create mode 100644 src/transport/retry.ts create mode 100644 test/integration/retry_test.ts diff --git a/README.md b/README.md index 94b2a4da..83c32c5e 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,56 @@ them, keeping in mind that Fetch preserves the CloudEvent POST and its body only for `307` and `308` - it turns a `301`, `302` or `303` into a bodyless `GET`, which drops the event. +`httpTransport()` sends each event once. Applications which can accept +at-least-once delivery can add retries by wrapping the emitter. `maxAttempts` +counts the first send, so this emitter makes at most five requests: + +```js +import { CloudEvent, emitterFor, httpTransport, withRetry } from "cloudevents"; + +const emit = withRetry( + emitterFor(httpTransport("https://events.example.com/orders")), + { maxAttempts: 5 }, +); +const orderCreated = new CloudEvent({ + type: "com.example.order.created", + source: "/orders", + data: { orderId: "order-123" }, +}); + +async function main() { + await emit(orderCreated, { signal: AbortSignal.timeout(30000) }); +} + +main().catch(console.error); +``` + +The default retry policy handles failures which are commonly temporary: + +| Failure reported by the emitter | Retried by default | +| ------------------------------- | ------------------ | +| Network failure before a response | Yes | +| HTTP 408, 425, 429, 500, 502, 503 or 504 | Yes | +| An aborted send or another HTTP status | No | +| An error from a custom transport | No | + +Retries use randomized exponential backoff. For HTTP 429 and 503, a valid +`Retry-After` header takes precedence, and `maxRetryDelay` caps whatever a delay +asks for - 30 seconds by default, so a distant `Retry-After` cannot park an +emitter for hours. Pass `shouldRetry(error, context)` or +`retryDelay(error, context)` to replace either policy. The exported +`isRetryableHTTPError()` and `defaultRetryDelay()` functions let a custom policy +reuse the defaults. + +The signal passed with the event also ends a wait between attempts. The emitter +then rejects with what that signal carries, rather than with the failure which +led to the retry. + +A network failure can happen after the receiver accepted an event but before +its response arrived. Every attempt therefore uses the same CloudEvent and the +same `source` and `id`, but the receiver still needs to recognize duplicate +delivery. + The [API transition guide](./API_TRANSITION_GUIDE.md) covers the header forms the transport accepts and where a proxy or a custom CA goes now that `http.globalAgent` no longer applies. diff --git a/src/index.ts b/src/index.ts index df586fc3..b6d5dcdd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,6 +15,9 @@ import { HTTPTransportErrorDetails, HTTPTransportErrorKind, HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions, httpDiscardResponseHandler, httpTextResponseHandler, httpTransport } from "./transport/http"; +import { + RetryContext, RetryOptions, defaultRetryDelay, isRetryableHTTPError, withRetry, +} from "./transport/retry"; import { Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory, Serializer, Deserializer } from "./message"; @@ -34,6 +37,9 @@ export { MQTTMessageFactory, // From transport emitterFor, + withRetry, + defaultRetryDelay, + isRetryableHTTPError, httpTransport, httpTextResponseHandler, httpDiscardResponseHandler, @@ -59,6 +65,8 @@ export type { TransportFunction, EmitterFunction, HTTPEmitterFunction, + RetryContext, + RetryOptions, Options, FetchHeadersInit, FetchRequestInit, diff --git a/src/transport/retry.ts b/src/transport/retry.ts new file mode 100644 index 00000000..7f82562b --- /dev/null +++ b/src/transport/retry.ts @@ -0,0 +1,240 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import { CloudEvent } from "../event/cloudevent"; +import { EmitterFunction, HTTPEmitterFunction } from "./emitter"; +import { HTTPTransportError } from "./http"; +import { signalFromOptions } from "./signal"; + +/** Information about the failed send being considered for another attempt */ +export interface RetryContext { + /** The one-based number of the attempt that just failed */ + attempt: number; + /** The maximum number of attempts, including the first send */ + maxAttempts: number; +} + +/** Options applied by {@linkcode withRetry} to every event sent through its emitter */ +export interface RetryOptions { + /** Maximum number of attempts, including the first send; defaults to 5 */ + maxAttempts?: number; + /** Decides whether an error should be retried; defaults to {@linkcode isRetryableHTTPError} */ + shouldRetry?: (error: unknown, context: RetryContext) => boolean; + /** + * Returns the delay before the next attempt in milliseconds; defaults to + * {@linkcode defaultRetryDelay} + */ + retryDelay?: (error: unknown, context: RetryContext) => number; + /** + * The longest a single wait may last in milliseconds, applied to whatever `retryDelay` + * returns, so a sink cannot park an emitter with a distant `Retry-After`; defaults to 30 + * seconds. Pass `Number.POSITIVE_INFINITY` to wait for however long a delay asks + */ + maxRetryDelay?: number; +} + +const DEFAULT_MAX_ATTEMPTS = 5; +const DEFAULT_MAX_RETRY_DELAY = 30_000; +const MAX_TIMER_DELAY = 2_147_483_647; +const RETRYABLE_HTTP_STATUSES = new Set([408, 425, 429, 500, 502, 503, 504]); +const RETRY_AFTER_STATUSES = new Set([429, 503]); + +/** + * Decide whether the built-in HTTP transport reported a failure which is commonly temporary. + * Network failures and HTTP 408, 425, 429, 500, 502, 503 and 504 are retried. Aborts, other + * status codes and errors from custom transports are not. + * + * @param {unknown} error the error reported by an emitter + * @returns {boolean} whether another HTTP send may succeed + */ +export function isRetryableHTTPError(error: unknown): boolean { + if (error instanceof HTTPTransportError && error.kind === "network") { + return true; + } + const response = httpStatusResponse(error); + return response !== undefined && RETRYABLE_HTTP_STATUSES.has(response.status); +} + +/** + * Return the default delay before another send. A valid `Retry-After` header on HTTP 429 or + * 503 takes precedence. Other failures wait `300 ms * 2 ** attempt`, scaled by a random factor + * between 0.4 and 1.4: about 600 ms, 1.2 s, 2.4 s and 4.8 s before attempts two through five. + * + * What a caller waits is also bounded by `maxRetryDelay`, which {@linkcode withRetry} applies + * to whatever this function returns. + * + * @param {unknown} error the error reported by an emitter + * @param {RetryContext} context the failed attempt and configured limit + * @returns {number} delay before the next attempt in milliseconds + */ +export function defaultRetryDelay(error: unknown, context: RetryContext): number { + const retryAfter = retryAfterDelay(error); + if (retryAfter !== undefined) { + return retryAfter; + } + + const exponentialDelay = 300 * 2 ** Math.min(context.attempt, context.maxAttempts); + return Math.floor((Math.random() + 0.4) * exponentialDelay); +} + +/** + * Add retry behavior to an emitter backed by the built-in HTTP transport. Per-send response + * handlers retain the response body type they return. + * + * @param {HTTPEmitterFunction} emitter the emitter to invoke for every attempt + * @param {RetryOptions} options attempt limit, error predicate, delay policy and delay ceiling + * @returns {HTTPEmitterFunction} an emitter with the same call signatures and retry behavior + */ +export function withRetry( + emitter: HTTPEmitterFunction, options?: RetryOptions, +): HTTPEmitterFunction; +/** + * Add retry behavior to any {@linkcode EmitterFunction}. The same CloudEvent and send options + * are passed to every attempt. Errors from custom transports require a custom `shouldRetry` + * callback because the default predicate only recognizes `HTTPTransportError`. + * + * @param {EmitterFunction} emitter the emitter to invoke for every attempt + * @param {RetryOptions} options attempt limit, error predicate, delay policy and delay ceiling + * @returns {EmitterFunction} an emitter with the same result and option types + */ +export function withRetry( + emitter: EmitterFunction, options?: RetryOptions, +): EmitterFunction; +export function withRetry( + emitter: EmitterFunction, options: RetryOptions = {}, +): EmitterFunction { + if (typeof emitter !== "function") { + throw new TypeError("An EmitterFunction is required"); + } + + const maxAttempts = options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; + if (!Number.isInteger(maxAttempts) || maxAttempts < 1) { + throw new TypeError("options.maxAttempts must be a positive integer"); + } + const shouldRetry = options.shouldRetry ?? isRetryableHTTPError; + if (typeof shouldRetry !== "function") { + throw new TypeError("options.shouldRetry must be a function"); + } + + const retryDelay = options.retryDelay ?? defaultRetryDelay; + if (typeof retryDelay !== "function") { + throw new TypeError("options.retryDelay must be a function"); + } + + const maxRetryDelay = options.maxRetryDelay ?? DEFAULT_MAX_RETRY_DELAY; + if (typeof maxRetryDelay !== "number" || Number.isNaN(maxRetryDelay) || maxRetryDelay < 0) { + throw new TypeError("options.maxRetryDelay must be a non-negative number"); + } + + return async function emitWithRetry(event: CloudEvent, sendOptions?: TOptions): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await emitter(event, sendOptions); + } catch (error) { + const context = { attempt, maxAttempts }; + if (attempt >= maxAttempts || !shouldRetry(error, context)) { + throw error; + } + + const delay = retryDelay(error, context); + if (!Number.isFinite(delay) || delay < 0) { + throw new TypeError("options.retryDelay must return a finite, non-negative number"); + } + await wait(Math.min(delay, maxRetryDelay), signalFromOptions(sendOptions)); + } + } + }; +} + +/** + * Read the response a sink sent, for the failures which carry one + * + * @param {unknown} error the error which may carry an HTTP response + * @returns {Response|undefined} the response, or undefined for any other failure + */ +function httpStatusResponse(error: unknown): Response | undefined { + return error instanceof HTTPTransportError && error.kind === "http-status" + ? error.response + : undefined; +} + +/** + * Read Retry-After for a status where the built-in retry policy uses it + * + * @param {unknown} error the error which may carry an HTTP response + * @returns {number|undefined} the requested delay, or undefined when there is none + */ +function retryAfterDelay(error: unknown): number | undefined { + const response = httpStatusResponse(error); + if (response === undefined || !RETRY_AFTER_STATUSES.has(response.status)) { + return undefined; + } + + const value = response.headers.get("retry-after")?.trim(); + if (!value) { + return undefined; + } + if (/^\d+$/.test(value)) { + const milliseconds = Number(value) * 1000; + return Number.isFinite(milliseconds) ? milliseconds : undefined; + } + + const retryAt = Date.parse(value); + return Number.isNaN(retryAt) ? undefined : Math.max(0, retryAt - Date.now()); +} + +/** + * Wait without retaining an AbortSignal after the timer settles + * + * @param {number} milliseconds how long to wait + * @param {AbortSignal} signal a signal which can end the wait early + * @returns {Promise} completion when the timer expires + */ +function wait(milliseconds: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + let remaining = milliseconds; + let timeout: ReturnType | undefined; + const removeAbortListener = (): void => signal?.removeEventListener("abort", abort); + const schedule = (): void => { + const scheduled = Math.min(remaining, MAX_TIMER_DELAY); + timeout = setTimeout(() => { + remaining -= scheduled; + if (remaining > 0) { + schedule(); + } else { + removeAbortListener(); + resolve(); + } + }, scheduled); + }; + const abort = (): void => { + if (timeout !== undefined) { + clearTimeout(timeout); + } + removeAbortListener(); + reject(signal?.reason ?? abortError()); + }; + + if (signal !== undefined) { + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) { + abort(); + return; + } + } + schedule(); + }); +} + +/** + * Provide a standard-looking reason for implementations whose aborted signal has none + * + * @returns {Error} an AbortError suitable as a rejection reason + */ +function abortError(): Error { + const error = new Error("Retry was aborted"); + error.name = "AbortError"; + return error; +} diff --git a/test/integration/retry_test.ts b/test/integration/retry_test.ts new file mode 100644 index 00000000..ba2ef927 --- /dev/null +++ b/test/integration/retry_test.ts @@ -0,0 +1,355 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +import "mocha"; +import { expect } from "chai"; + +import { + EmitterFunction, HTTPTransportError, HTTPTransportResponse, RetryContext, RetryOptions, + defaultRetryDelay, emitterFor, httpTransport, isRetryableHTTPError, withRetry, +} from "../../src"; +import { fixture } from "./emitter_factory_test"; +import { withServer } from "./http_transport_test"; + +interface TestSendOptions { + label?: string; + signal?: AbortSignal; +} + +describe("withRetry()", () => { + it("Retries the same event and send options until the emitter succeeds", async () => { + const networkError = new HTTPTransportError("network", { cause: new Error("connection reset") }); + const events: unknown[] = []; + const options: unknown[] = []; + const contexts: RetryContext[] = []; + let attempts = 0; + const emitter: EmitterFunction = async (event, sendOptions) => { + attempts++; + events.push(event); + options.push(sendOptions); + if (attempts < 3) { + throw networkError; + } + return "accepted"; + }; + const emit = withRetry(emitter, { + maxAttempts: 3, + retryDelay: (_error, context) => { + contexts.push(context); + return 0; + }, + }); + const sendOptions = { label: "order-123" }; + + expect(await emit(fixture, sendOptions)).to.equal("accepted"); + expect(attempts).to.equal(3); + expect(events).to.deep.equal([fixture, fixture, fixture]); + expect(options).to.deep.equal([sendOptions, sendOptions, sendOptions]); + expect(contexts).to.deep.equal([ + { attempt: 1, maxAttempts: 3 }, + { attempt: 2, maxAttempts: 3 }, + ]); + }); + + it("Defaults to five total attempts and throws the final error unchanged", async () => { + const errors = Array.from({ length: 5 }, (_, index) => new Error(`failure ${index + 1}`)); + let attempts = 0; + const emitter: EmitterFunction = async () => { + throw errors[attempts++]; + }; + const emit = withRetry(emitter, { + shouldRetry: () => true, + retryDelay: () => 0, + }); + + expect(await errorFrom(emit(fixture))).to.equal(errors[4]); + expect(attempts).to.equal(5); + }); + + it("Allows a custom predicate to retry errors from a custom transport", async () => { + const transient = new Error("broker is busy"); + const seen: Array<{ error: unknown; context: RetryContext }> = []; + const emit = withRetry(failsOnce(transient), { + shouldRetry: (error, context) => { + seen.push({ error, context }); + return error === transient; + }, + retryDelay: () => 0, + }); + + expect(await emit(fixture)).to.equal("accepted"); + expect(seen).to.deep.equal([{ + error: transient, + context: { attempt: 1, maxAttempts: 5 }, + }]); + }); + + it("Sends once when maxAttempts is 1", async () => { + let attempts = 0; + const emitter: EmitterFunction = async () => { + attempts++; + throw new HTTPTransportError("network"); + }; + const emit = withRetry(emitter, { maxAttempts: 1, retryDelay: () => 0 }); + + expect(await errorFrom(emit(fixture))).to.be.instanceOf(HTTPTransportError); + expect(attempts).to.equal(1); + }); + + it("Retries a 503 from the built-in HTTP transport until the sink accepts the event", async () => { + const statuses = [503, 202]; + let requestCount = 0; + + await withServer((_request, response) => { + response.writeHead(statuses[requestCount++] ?? 500); + response.end(); + }, async (url) => { + const emit = withRetry(emitterFor(httpTransport(url)), { retryDelay: () => 0 }); + const { response } = await emit(fixture); + + expect(response.status).to.equal(202); + expect(requestCount).to.equal(2); + }); + }); + + it("Does not retry a 400 from the built-in HTTP transport", async () => { + let requestCount = 0; + + await withServer((_request, response) => { + requestCount++; + response.writeHead(400); + response.end(); + }, async (url) => { + const emit = withRetry(emitterFor(httpTransport(url)), { retryDelay: () => 0 }); + const error = await errorFrom(emit(fixture)); + + expect(error).to.be.instanceOf(HTTPTransportError); + expect((error as HTTPTransportError).response?.status).to.equal(400); + expect(requestCount).to.equal(1); + }); + }); + + it("Stops during backoff when the per-send signal aborts", async () => { + const controller = new AbortController(); + const reason = new Error("delivery deadline reached"); + let attempts = 0; + const emitter: EmitterFunction = async () => { + attempts++; + throw new HTTPTransportError("network"); + }; + const emit = withRetry(emitter, { retryDelay: () => 10_000 }); + + const emitted = emit(fixture, { signal: controller.signal }); + setTimeout(() => controller.abort(reason), 10); + + expect(await errorFrom(emitted)).to.equal(reason); + expect(attempts).to.equal(1); + }); + + it("Splits delays which exceed the JavaScript timer limit", async () => { + const maximumTimerDelay = 2_147_483_647; + + const scheduled = await recordedTimerDelays(async () => { + const emit = withRetry(failsOnce(new HTTPTransportError("network")), { + retryDelay: () => maximumTimerDelay + 1, + // the delay ceiling would otherwise keep this within a single timer + maxRetryDelay: Number.POSITIVE_INFINITY, + }); + expect(await emit(fixture)).to.equal("accepted"); + }); + + expect(scheduled).to.deep.equal([maximumTimerDelay, 1]); + }); + + it("Bounds what a sink asks for with maxRetryDelay", async () => { + // a sink which asks for a day before the next attempt + const asksForADay = httpStatusError(503, "86400"); + + const byDefault = await recordedTimerDelays(async () => { + expect(await withRetry(failsOnce(asksForADay))(fixture)).to.equal("accepted"); + }); + const configured = await recordedTimerDelays(async () => { + expect(await withRetry(failsOnce(asksForADay), { maxRetryDelay: 5_000 })(fixture)) + .to.equal("accepted"); + }); + + expect(byDefault).to.deep.equal([30_000]); + expect(configured).to.deep.equal([5_000]); + }); + + it("Validates retry configuration and callback results", async () => { + const emitter: EmitterFunction = async () => { + throw new HTTPTransportError("network"); + }; + + expect(() => withRetry(emitter, { maxAttempts: 0 })).to.throw( + TypeError, "options.maxAttempts must be a positive integer", + ); + expect(() => withRetry(emitter, { + shouldRetry: "yes" as unknown as RetryOptions["shouldRetry"], + })).to.throw(TypeError, "options.shouldRetry must be a function"); + expect(() => withRetry(emitter, { + retryDelay: 100 as unknown as RetryOptions["retryDelay"], + })).to.throw(TypeError, "options.retryDelay must be a function"); + expect(() => withRetry(emitter, { maxRetryDelay: -1 })).to.throw( + TypeError, "options.maxRetryDelay must be a non-negative number", + ); + expect(() => withRetry(emitter, { maxRetryDelay: Number.POSITIVE_INFINITY })).not.to.throw(); + + const emit = withRetry(emitter, { retryDelay: () => Number.POSITIVE_INFINITY }); + expect(await errorFrom(emit(fixture))).to.be.instanceOf(TypeError); + }); + + it("Keeps the built-in HTTP emitter's per-send response type", () => { + const emit = withRetry(emitterFor(httpTransport("https://events.example.com/orders"))); + const inferred = () => emit(fixture, { + responseHandler: async () => ({ accepted: true }), + }); + const typed: () => Promise> = inferred; + + expect(typed).to.be.a("function"); + }); +}); + +describe("isRetryableHTTPError()", () => { + it("Retries network failures and temporary HTTP statuses", () => { + expect(isRetryableHTTPError(new HTTPTransportError("network"))).to.equal(true); + + for (const status of [408, 425, 429, 500, 502, 503, 504]) { + expect(isRetryableHTTPError(httpStatusError(status)), `${status}`).to.equal(true); + } + }); + + it("Does not retry aborts, permanent statuses or unknown errors", () => { + expect(isRetryableHTTPError(new HTTPTransportError("aborted"))).to.equal(false); + expect(isRetryableHTTPError(httpStatusError(400))).to.equal(false); + expect(isRetryableHTTPError(httpStatusError(501))).to.equal(false); + expect(isRetryableHTTPError(new Error("custom transport failed"))).to.equal(false); + }); +}); + +describe("defaultRetryDelay()", () => { + it("Uses the exponential schedule with randomized jitter", () => { + withRandom(0.5, () => { + expect(defaultRetryDelay(new Error("temporary"), context(1))).to.equal(540); + expect(defaultRetryDelay(new Error("temporary"), context(2))).to.equal(1080); + expect(defaultRetryDelay(new Error("temporary"), context(3))).to.equal(2160); + expect(defaultRetryDelay(new Error("temporary"), context(4))).to.equal(4320); + }); + }); + + it("Caps the exponential schedule at maxAttempts", () => { + withRandom(0.5, () => { + expect(defaultRetryDelay(new Error("temporary"), { + attempt: 6, + maxAttempts: 5, + })).to.equal(8640); + }); + }); + + it("Uses Retry-After seconds for 429 and 503", () => { + expect(defaultRetryDelay(httpStatusError(429, "3"), context(1))).to.equal(3000); + expect(defaultRetryDelay(httpStatusError(503, "0"), context(2))).to.equal(0); + }); + + it("Uses a future Retry-After HTTP date", () => { + const now = Date.UTC(2026, 7, 17, 12, 0, 0); + withNow(now, () => { + const retryAt = new Date(now + 5000).toUTCString(); + expect(defaultRetryDelay(httpStatusError(503, retryAt), context(1))).to.equal(5000); + }); + }); + + it("Falls back to exponential delay for invalid or inapplicable Retry-After headers", () => { + withRandom(0.5, () => { + expect(defaultRetryDelay(httpStatusError(429, "later"), context(1))).to.equal(540); + expect(defaultRetryDelay(httpStatusError(500, "3"), context(1))).to.equal(540); + }); + }); + + it("Allows a custom retryDelay to replace Retry-After handling", async () => { + const error = httpStatusError(503, "120"); + let delayError: unknown; + const emit = withRetry(failsOnce(error), { + retryDelay: (received) => { + delayError = received; + return 0; + }, + }); + + expect(await emit(fixture)).to.equal("accepted"); + expect(delayError).to.equal(error); + }); +}); + +// Record what withRetry() asks a timer for, without waiting for it. +async function recordedTimerDelays(run: () => Promise): Promise { + const scheduled: number[] = []; + const originalSetTimeout = globalThis.setTimeout; + globalThis.setTimeout = ((callback: (...args: unknown[]) => void, milliseconds?: number) => { + scheduled.push(milliseconds ?? 0); + return originalSetTimeout(callback, 0); + }) as typeof setTimeout; + + try { + await run(); + } finally { + globalThis.setTimeout = originalSetTimeout; + } + return scheduled; +} + +// Run with Math.random() pinned, so the jittered delays are exact. +function withRandom(value: number, run: () => T): T { + const originalRandom = Math.random; + Math.random = () => value; + try { + return run(); + } finally { + Math.random = originalRandom; + } +} + +// Run with Date.now() pinned, so a Retry-After date resolves to an exact delay. +function withNow(now: number, run: () => T): T { + const originalNow = Date.now; + Date.now = () => now; + try { + return run(); + } finally { + Date.now = originalNow; + } +} + +// An emitter which throws once, then accepts the event. +function failsOnce(error: unknown): EmitterFunction { + let attempts = 0; + return async () => { + if (attempts++ === 0) { + throw error; + } + return "accepted"; + }; +} + +function context(attempt: number): RetryContext { + return { attempt, maxAttempts: 5 }; +} + +function httpStatusError(status: number, retryAfter?: string): HTTPTransportError { + const headers = retryAfter === undefined ? undefined : { "retry-after": retryAfter }; + return new HTTPTransportError("http-status", { + response: new Response(null, { status, headers }), + }); +} + +async function errorFrom(promise: Promise): Promise { + try { + await promise; + } catch (error) { + return error; + } + throw new Error("Expected the promise to reject"); +}