diff --git a/API_TRANSITION_GUIDE.md b/API_TRANSITION_GUIDE.md
index 4370a732..16738f85 100644
--- a/API_TRANSITION_GUIDE.md
+++ b/API_TRANSITION_GUIDE.md
@@ -100,3 +100,166 @@ 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
+
+In the 11.0.0 release, the built-in HTTP transport was rewritten on top of the Fetch API.
+
+#### HTTP Transport
+
+`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, 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`. 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");
+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
+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
+
+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.
+
+#### Emitter Options
+
+`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`.
+
+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 b40b8de3..50c00c9b 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,166 @@ 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);
+```
+
+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 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,
+`error.body` is absent and the error carries a `cause`.
+
+```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 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. Wrap an emitter with `withTimeout()`
+when every send should have the same time limit:
+
+```js
+import { emitterFor, httpTransport, withTimeout } from "cloudevents";
+
+const emit = withTimeout(
+  emitterFor(httpTransport("https://my.receiver.com/endpoint", {
+    headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` },
+    responseHandler: (response) => response.json(),
+  })),
+  10000,
+);
+
+async function main() {
+  await emit(ce);
+}
+
+main().catch(console.error);
+```
+
+For a deadline which varies by event, pass `AbortSignal.timeout()` to that
+`emit()` call instead. If both are present, the caller's signal and the timeout
+can each abort the send. The timeout passed to `withTimeout()` must be a whole
+number from 0 to 2,147,483,647 milliseconds.
+
+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.
+
+`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, withTimeout,
+} from "cloudevents";
+
+const emit = withRetry(
+  withTimeout(
+    emitterFor(httpTransport("https://events.example.com/orders")),
+    10000,
+  ),
+  { maxAttempts: 5 },
+);
+const orderCreated = new CloudEvent({
+  type: "com.example.order.created",
+  source: "/orders",
+  data: { orderId: "order-123" },
+});
+
+async function main() {
+  await emit(orderCreated);
+}
+
+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 |
+
+The wrapper order decides which work the timeout covers:
+
+| Goal | Composition | What the timeout covers |
+| ---- | ----------- | ----------------------- |
+| Limit each attempt to 10 seconds | `withRetry(withTimeout(base, 10000))` | One fresh timeout per attempt; retry backoff is excluded |
+| Limit the whole delivery to 30 seconds | `withTimeout(withRetry(base), 30000)` | Every attempt and retry backoff share one timeout |
+
+A timed-out HTTP send is reported as an abort, which the default policy does
+not retry. When an attempt fails earlier with a retryable network or HTTP error,
+the next attempt receives a fresh timeout in the first composition above.
+
+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.
+
 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
@@ -109,20 +266,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/package-lock.json b/package-lock.json
index bbffeea7..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",
@@ -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"
       },
@@ -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",
@@ -1891,6 +1894,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 +8873,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 +9736,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": {
@@ -9824,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",
@@ -10402,9 +10401,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..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",
@@ -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"
   },
diff --git a/src/index.ts b/src/index.ts
index ab0148db..f1e111be 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -7,8 +7,18 @@ 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, HTTPEmitterFunction, emitterFor, Emitter,
+} from "./transport/emitter";
+import {
+  FetchHeadersInit, FetchRequestInit, HTTPResponseHandler, HTTPTransportError,
+  HTTPTransportErrorDetails, HTTPTransportErrorKind,
+  HTTPTransportFunction, HTTPTransportOptions, HTTPTransportResponse, HTTPTransportSendOptions,
+  httpDiscardResponseHandler, httpTextResponseHandler, httpTransport } from "./transport/http";
+import {
+  RetryContext, RetryOptions, defaultRetryDelay, isRetryableHTTPError, withRetry,
+} from "./transport/retry";
+import { withTimeout } from "./transport/timeout";
 import {
   Headers, Mode, Binding, HTTP, Kafka, KafkaEvent, KafkaMessage, Message, MQTT, MQTTMessage, MQTTMessageFactory,
   Serializer, Deserializer } from "./message";
@@ -26,8 +36,16 @@ export {
   Kafka,
   MQTT,
   MQTTMessageFactory,
+  // From transport
   emitterFor,
+  withRetry,
+  withTimeout,
+  defaultRetryDelay,
+  isRetryableHTTPError,
   httpTransport,
+  httpTextResponseHandler,
+  httpDiscardResponseHandler,
+  HTTPTransportError,
   Emitter,
   // From Constants
   CONSTANTS
@@ -48,5 +66,17 @@ export type {
   // From transport
   TransportFunction,
   EmitterFunction,
-  Options
+  HTTPEmitterFunction,
+  RetryContext,
+  RetryOptions,
+  Options,
+  FetchHeadersInit,
+  FetchRequestInit,
+  HTTPResponseHandler,
+  HTTPTransportErrorDetails,
+  HTTPTransportErrorKind,
+  HTTPTransportFunction,
+  HTTPTransportOptions,
+  HTTPTransportResponse,
+  HTTPTransportSendOptions
 };
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..601e7880 100644
--- a/src/transport/emitter.ts
+++ b/src/transport/emitter.ts
@@ -4,15 +4,22 @@
 */
 
 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<THeaders = CloudEventHeaders> {
+  /** 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<string, unknown> | unknown;
 }
 
@@ -20,23 +27,50 @@ 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 {
-  <T>(event: CloudEvent<T>, options?: Options): Promise<unknown>;
+export interface EmitterFunction<TResult = unknown, TOptions = Options> {
+  <T>(event: CloudEvent<T>, options?: TOptions): Promise<TResult>;
+}
+
+/**
+ * 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<TDefaultBody = string>
+  extends EmitterFunction<HTTPTransportResponse<TDefaultBody>, Options<FetchHeadersInit>> {
+  <TEvent, TBody>(
+    event: CloudEvent<TEvent>, options: HTTPTransportSendOptions<TBody>,
+  ): Promise<HTTPTransportResponse<TBody>>;
 }
 
 /**
  * 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<unknown>;
+export interface TransportFunction<TResult = unknown, TOptions = Options> {
+  (message: Message, options?: TOptions): Promise<TResult>;
 }
 
 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<TBody>(
+  fn: HTTPTransportFunction<TBody>, options?: Options,
+): HTTPEmitterFunction<TBody>;
 /**
  * Creates and returns an {@linkcode EmitterFunction} using the supplied
  * {@linkcode TransportFunction}. The returned {@linkcode EmitterFunction}
@@ -51,19 +85,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<TResult = unknown, TOptions = Options>(
+  fn: TransportFunction<TResult, TOptions>, options?: Options,
+): EmitterFunction<TResult, TOptions>;
+export function emitterFor<TResult = unknown, TOptions = Options>(
+  fn: TransportFunction<TResult, TOptions>, options: Options = emitterDefaults,
+): EmitterFunction<TResult, TOptions> {
   if (!fn) {
     throw new TypeError("A TransportFunction is required");
   }
   const { binding, mode }: any = { ...emitterDefaults, ...options };
-  return function emit<T>(event: CloudEvent<T>, opts?: Options): Promise<unknown> {
-    opts = opts || {};
+  return function emit<T>(event: CloudEvent<T>, opts?: TOptions): Promise<TResult> {
+    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..0d490646 100644
--- a/src/transport/http/index.ts
+++ b/src/transport/http/index.ts
@@ -3,61 +3,416 @@
  SPDX-License-Identifier: Apache-2.0
 */
 
-import { Socket } from "net";
-import http, { OutgoingHttpHeaders } from "http";
-import https, { RequestOptions } from "https";
+import { Headers as CloudEventHeaders, Message } from "../../message";
+import { Options, TransportFunction } from "../emitter";
+import { combineSignals, signalFrom } from "../signal";
 
-import { Message, Options } from "../..";
-import { TransportFunction } from "../emitter";
+/** The request options accepted by the Fetch implementation in the current environment */
+export type FetchRequestInit = NonNullable<Parameters<typeof globalThis.fetch>[1]>;
+
+/** The header forms accepted by the Fetch implementation in the current environment */
+export type FetchHeadersInit = NonNullable<FetchRequestInit["headers"]>;
+
+/** Turns a Fetch response into the body value handed back by the HTTP transport */
+export type HTTPResponseHandler<TBody> = (response: Response) => Promise<TBody>;
+
+/**
+ * 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<string>} the response body as text
+ */
+export function httpTextResponseHandler(response: Response): Promise<string> {
+  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<void>} completion after the body has been cancelled
+ */
+export async function httpDiscardResponseHandler(response: Response): Promise<void> {
+  await response.body?.cancel();
+}
+
+/**
+ * Options applied to every request sent by an HTTP transport
+ */
+export interface HTTPTransportOptions<TBody = string> {
+  /** 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<FetchRequestInit, "method" | "headers" | "body" | "signal">;
+  /** Reads every response body; defaults to {@linkcode httpTextResponseHandler} */
+  responseHandler?: HTTPResponseHandler<TBody>;
+}
+
+/**
+ * Options which replace the response handler for one send
+ */
+export interface HTTPTransportSendOptions<TBody> extends Options<FetchHeadersInit> {
+  /** Reads this response body instead of the handler configured for the transport */
+  responseHandler: HTTPResponseHandler<TBody>;
+}
+
+/** What the built-in HTTP transport hands back for a send the sink accepted */
+export type HTTPTransportResponse<TBody = string> =
+  | {
+    /** 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;
+  };
+
+/**
+ * An HTTP transport whose default response body type can be replaced for one send
+ */
+export interface HTTPTransportFunction<TDefaultBody = string>
+  extends TransportFunction<HTTPTransportResponse<TDefaultBody>, Options<FetchHeadersInit>> {
+  <TBody>(message: Message, options: HTTPTransportSendOptions<TBody>): Promise<HTTPTransportResponse<TBody>>;
+}
+
+/** 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
+ * - `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" | "aborted" | "network";
+
+/**
+ * What an {@linkcode HTTPTransportError} reports alongside its {@linkcode HTTPTransportErrorKind}
+ */
+export type HTTPTransportErrorDetails = Pick<HTTPTransportError, "response" | "body" | "cause">;
+
+/**
+ * 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 failure, e.g. what fetch threw or the reason carried by a signal */
+  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:
- * 
- * const emitter = emitterFor(httpTransport("http://example.com"));
- * emitter.emit(myCloudEvent)
- *    .then(resp => console.log(resp));
- * 
+ *
+ * ```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, 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
- * @returns {TransportFunction} a function which can be used to send CloudEvents to _sink_
+ * @param {HTTPTransportOptions} options headers, Fetch options, abort and response behavior
+ * @returns {HTTPTransportFunction} a function that sends CloudEvents to _sink_
  */
-export function httpTransport(sink: string | URL): TransportFunction {
-  const url = new URL(sink);
-  let base: any;
-  if (url.protocol === "https:") {
-    base = https;
-  } else if (url.protocol === "http:") {
-    base = http;
-  } else {
-    throw new TypeError(`unsupported protocol ${url.protocol}`);
-  }
-  return function(message: Message, options?: Options): Promise<unknown> {
-    return new Promise((resolve, reject) => {
-      options = { ...options };
+export function httpTransport<TBody = string>(
+  sink: string | URL, options: HTTPTransportOptions<TBody> = {},
+): HTTPTransportFunction<TBody> {
+  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 = {
+    ...options.fetchOptions,
+    // fetch reads an explicitly undefined redirect as absent, so the default comes last
+    redirect: options.fetchOptions?.redirect ?? "manual",
+    method: "POST",
+  };
 
-      // TODO: Callers should be able to set any Node.js RequestOptions
-      const opts: RequestOptions = {
-        method: "POST",
-        headers: {...message.headers, ...options.headers as OutgoingHttpHeaders},
-      };
+  const send = async (
+    message: Message, sendOptions?: Options<FetchHeadersInit>,
+  ): Promise<HTTPTransportResponse<unknown>> => {
+    const sendSignal = signalFrom(sendOptions?.signal);
+    const headers = requestHeaders(
+      sinkHeaders, message.headers, transportHeaders, headersFrom(sendOptions?.headers),
+    );
+    const responseHandler = responseHandlerFrom(sendOptions?.responseHandler, transportResponseHandler);
+    const combinedSignal = combineSignals(transportSignal, sendSignal);
+    try {
+      let response: Response;
       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); });
+        response = await fetch(url, {
+          ...fetchOptions,
+          headers,
+          body: message.body as BodyInit,
+          signal: combinedSignal.signal,
         });
-        req.on("error", reject);
-        req.write(message.body);
-        req.end();
-      } catch (err) {
-        reject(err);
+      } 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
+          ? { response, body: handled.body }
+          : { response, cause: handled.error });
       }
-    });
+      return handled.handled
+        ? { response, body: handled.body }
+        : { response, bodyError: handled.error };
+    } finally {
+      combinedSignal.dispose();
+    }
   };
+
+  // the overloads type the per-send handler this implementation cannot
+  return send as HTTPTransportFunction<TBody>;
+}
+
+/**
+ * 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<unknown>,
+): HTTPResponseHandler<unknown> {
+  if (value === undefined || value === null) {
+    return fallback;
+  }
+  if (typeof value !== "function") {
+    throw new TypeError("options.responseHandler must be a function");
+  }
+  return value as HTTPResponseHandler<unknown>;
+}
+
+/**
+ * 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<unknown>,
+): Promise<HandledResponse> {
+  try {
+    return { handled: true, body: await handler(response) };
+  } catch (error) {
+    try {
+      await httpDiscardResponseHandler(response);
+    } catch {
+      // best effort cleanup
+    }
+    return { handled: false, error };
+  }
+}
+
+/**
+ * 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 {URL} the parsed sink
+ */
+function validateHTTPURL(sink: string | URL): URL {
+  const url = new URL(sink);
+  if (url.protocol !== "http:" && url.protocol !== "https:") {
+    throw new TypeError(`unsupported protocol ${url.protocol}`);
+  }
+  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<string, unknown>)) {
+    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);
+}
+
+/**
+ * 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}`;
+    case "aborted":
+      return "HTTP transport request was aborted";
+    default:
+      return "HTTP transport request failed";
+  }
 }
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<TBody>(
+  emitter: HTTPEmitterFunction<TBody>, options?: RetryOptions,
+): HTTPEmitterFunction<TBody>;
+/**
+ * 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<TResult, TOptions>(
+  emitter: EmitterFunction<TResult, TOptions>, options?: RetryOptions,
+): EmitterFunction<TResult, TOptions>;
+export function withRetry<TResult, TOptions>(
+  emitter: EmitterFunction<TResult, TOptions>, options: RetryOptions = {},
+): EmitterFunction<TResult, TOptions> {
+  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<T>(event: CloudEvent<T>, sendOptions?: TOptions): Promise<TResult> {
+    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<void>} completion when the timer expires
+ */
+function wait(milliseconds: number, signal?: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    let remaining = milliseconds;
+    let timeout: ReturnType<typeof setTimeout> | 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/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<AbortSignal, AbortFanout>();
+
+/** 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<AbortSignal|undefined>} candidates the signals which can abort the operation
+ * @returns {CombinedSignal} the combined signal and an idempotent cleanup function
+ */
+export function combineSignals(...candidates: Array<AbortSignal | undefined>): 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/src/transport/timeout.ts b/src/transport/timeout.ts
new file mode 100644
index 00000000..a84540c4
--- /dev/null
+++ b/src/transport/timeout.ts
@@ -0,0 +1,97 @@
+/*
+ Copyright 2021 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+import { CloudEvent } from "../event/cloudevent";
+import { EmitterFunction, HTTPEmitterFunction } from "./emitter";
+import { combineSignals, signalFromOptions } from "./signal";
+
+const MAX_TIMEOUT_MS = 2_147_483_647;
+
+/** A timeout signal and the cleanup which cancels its pending timer */
+interface TimeoutSignal {
+  signal: AbortSignal;
+  dispose: () => void;
+}
+
+/**
+ * Give each invocation of a built-in HTTP emitter its own timeout. Per-send response handlers
+ * retain the response body type they return.
+ *
+ * @param {HTTPEmitterFunction} emitter the emitter to invoke with a timeout signal
+ * @param {number} timeoutMs the maximum duration of one invocation in milliseconds, from 0 to 2147483647
+ * @returns {HTTPEmitterFunction} an emitter with the same call signatures and a per-send timeout
+ */
+export function withTimeout<TBody>(
+  emitter: HTTPEmitterFunction<TBody>, timeoutMs: number,
+): HTTPEmitterFunction<TBody>;
+/**
+ * Give each invocation of an {@linkcode EmitterFunction} its own timeout. The wrapped emitter
+ * must honor `options.signal`. When a caller also supplies a signal, either that signal or the
+ * timeout can abort the invocation. Other send options are preserved.
+ *
+ * Wrapping `withTimeout` inside `withRetry` starts a fresh timeout for every attempt. Wrapping
+ * `withRetry` inside `withTimeout` applies one timeout across all attempts and backoff waits.
+ *
+ * @param {EmitterFunction} emitter the emitter to invoke with a timeout signal
+ * @param {number} timeoutMs the maximum duration of one invocation in milliseconds, from 0 to 2147483647
+ * @returns {EmitterFunction} an emitter with the same result and option types
+ */
+export function withTimeout<TResult, TOptions extends { signal?: AbortSignal }>(
+  emitter: EmitterFunction<TResult, TOptions>, timeoutMs: number,
+): EmitterFunction<TResult, TOptions>;
+export function withTimeout<TResult, TOptions extends { signal?: AbortSignal }>(
+  emitter: EmitterFunction<TResult, TOptions>, timeoutMs: number,
+): EmitterFunction<TResult, TOptions> {
+  if (typeof emitter !== "function") {
+    throw new TypeError("An EmitterFunction is required");
+  }
+  if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 0) {
+    throw new TypeError("timeoutMs must be a non-negative safe integer");
+  }
+  if (timeoutMs > MAX_TIMEOUT_MS) {
+    throw new RangeError(`timeoutMs cannot be greater than ${MAX_TIMEOUT_MS}`);
+  }
+
+  return async function emitWithTimeout<T>(
+    event: CloudEvent<T>, sendOptions?: TOptions,
+  ): Promise<TResult> {
+    const callerSignal = signalFromOptions(sendOptions);
+    const timeout = timeoutSignal(timeoutMs);
+    const combinedSignal = combineSignals(callerSignal, timeout.signal);
+
+    try {
+      return await emitter(event, {
+        ...sendOptions,
+        signal: combinedSignal.signal,
+      } as TOptions);
+    } finally {
+      combinedSignal.dispose();
+      timeout.dispose();
+    }
+  };
+}
+
+/**
+ * Create a timeout signal and a cleanup which cancels its timer
+ *
+ * @param {number} milliseconds the duration before the signal aborts
+ * @returns {TimeoutSignal} the timeout signal and an idempotent timer cleanup function
+ */
+function timeoutSignal(milliseconds: number): TimeoutSignal {
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(timeoutError()), milliseconds);
+  const dispose = (): void => clearTimeout(timer);
+
+  return { signal: controller.signal, dispose };
+}
+
+/**
+ * Create the standard reason carried by an AbortSignal timeout
+ *
+ * @returns {DOMException} a TimeoutError suitable as an abort reason
+ */
+function timeoutError(): DOMException {
+  return new DOMException("The operation was aborted due to timeout", "TimeoutError");
+}
diff --git a/test/integration/emitter_factory_test.ts b/test/integration/emitter_factory_test.ts
index 568f6ccc..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, httpTransport }
-  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<unknown> {
-  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<unknown> {
@@ -66,6 +70,15 @@ function gotEmitter(message: Message, options?: Options): Promise<unknown> {
 }
 
 describe("emitterFor() defaults", () => {
+  it("Keeps CloudEvent header arrays in custom transport options", async () => {
+    const transport: TransportFunction<void> = 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<unknown> {
       // A binary message will have the source attribute as a header
@@ -101,8 +114,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 +129,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..650770f1
--- /dev/null
+++ b/test/integration/http_transport_test.ts
@@ -0,0 +1,844 @@
+/*
+ Copyright 2021 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+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";
+import { runInNewContext } from "vm";
+
+import {
+  CONSTANTS, FetchHeadersInit, FetchRequestInit, Headers as CloudEventHeaders, HTTPTransportError,
+  Message, Mode, emitterFor, httpDiscardResponseHandler, httpTextResponseHandler, httpTransport,
+} from "../../src";
+import { assertStructured, fixture } from "./emitter_factory_test";
+
+type Handler = (request: IncomingMessage, response: ServerResponse) => void | Promise<void>;
+
+// 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();
+};
+
+// 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<string, string> | undefined;
+    let receivedHeaders: IncomingMessage["headers"] | undefined;
+
+    await withServer(async (request, response) => {
+      receivedHeaders = request.headers;
+      receivedBody = await json(request) as Record<string, string>;
+      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<string, unknown> | undefined;
+
+    await withServer(async (request, response) => {
+      received = {
+        ...await json(request) as Record<string, unknown>,
+        ...request.headers,
+      };
+      acceptEvent(request, response);
+    }, async (url) => {
+      const emit = emitterFor(httpTransport(url), { mode: Mode.STRUCTURED });
+      await emit(fixture);
+      assertStructured(received as Record<string, Record<string, string>>);
+    });
+  });
+
+  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("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<string | undefined>(configured.body);
+      expect(configured.body).to.deep.equal({ receipt: 42 });
+
+      const overridden = await emit(fixture, { responseHandler: httpTextResponseHandler });
+      expectType<string | undefined>(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<void | undefined>(result.body);
+      expect(result.body).to.equal(undefined);
+      expect(result.response.bodyUsed).to.equal(true);
+      expect("bodyError" in result).to.equal(false);
+    });
+  });
+
+  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<ReadableStream<Uint8Array> | 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");
+
+    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();
+
+    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");
+  });
+
+  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 },
+    { 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("");
+      });
+    });
+  }
+
+  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<void>((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) => {
+      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("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<void>((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<Response>((_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);
+  });
+
+  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, {
+        // @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:");
+    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();
+    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");
+  });
+});
+
+async function transportErrorFrom(emitted: Promise<unknown>): Promise<HTTPTransportError> {
+  const error = await emitted.then(() => undefined, (reason: unknown) => reason);
+  expect(error).to.be.instanceOf(HTTPTransportError);
+  return error as HTTPTransportError;
+}
+
+// Compile-time assertion.
+function expectType<T>(value: T): void {
+  void value;
+}
+
+// Model a platform object from another realm: same behavior, different instanceof chain.
+function withForeignInterfacePrototype<T extends AbortSignal | Headers>(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<void>): Promise<void> {
+  const originalFetch = globalThis.fetch;
+  globalThis.fetch = stub;
+
+  try {
+    await run();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+}
+
+export async function withServer(handler: Handler, run: (url: string) => Promise<void>): Promise<void> {
+  const server = createServer((request, response) => {
+    Promise.resolve(handler(request, response)).catch((error) => {
+      response.destroy(error);
+    });
+  });
+  await new Promise<void>((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<void>((resolve, reject) => {
+      server.close((error) => error ? reject(error) : resolve());
+      server.closeAllConnections();
+    });
+  }
+}
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<string, TestSendOptions> = 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<void> = 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<void> = 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<void, TestSendOptions> = 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<void> = 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<HTTPTransportResponse<{ accepted: boolean }>> = 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<void>): Promise<number[]> {
+  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<T>(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<T>(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<string> {
+  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<unknown>): Promise<unknown> {
+  try {
+    await promise;
+  } catch (error) {
+    return error;
+  }
+  throw new Error("Expected the promise to reject");
+}
diff --git a/test/integration/timeout_test.ts b/test/integration/timeout_test.ts
new file mode 100644
index 00000000..953fd805
--- /dev/null
+++ b/test/integration/timeout_test.ts
@@ -0,0 +1,118 @@
+/*
+ Copyright 2021 The CloudEvents Authors
+ SPDX-License-Identifier: Apache-2.0
+*/
+
+import "mocha";
+import { expect } from "chai";
+
+import {
+  EmitterFunction, HTTPTransportError, HTTPTransportResponse, emitterFor, httpTransport,
+  withRetry, withTimeout,
+} from "../../src";
+import { fixture } from "./emitter_factory_test";
+import { withServer } from "./http_transport_test";
+
+interface TestSendOptions {
+  label?: string;
+  signal?: AbortSignal;
+}
+
+describe("withTimeout()", () => {
+  it("Aborts a built-in HTTP send after its timeout", async () => {
+    await withServer(() => undefined, async (url) => {
+      const emit = withTimeout(emitterFor(httpTransport(url)), 50);
+      const error = await errorFrom(emit(fixture));
+
+      expect(error).to.be.instanceOf(HTTPTransportError);
+      expect((error as HTTPTransportError).kind).to.equal("aborted");
+      expect(((error as HTTPTransportError).cause as Error)?.name).to.equal("TimeoutError");
+    });
+  });
+
+  it("Combines a caller's signal with the timeout and preserves other send options", async () => {
+    const controller = new AbortController();
+    const reason = new Error("caller stopped delivery");
+    let received: TestSendOptions | undefined;
+    const emitter: EmitterFunction<void, TestSendOptions> = async (_event, options) => {
+      received = options;
+      const signal = options?.signal;
+      if (!signal) {
+        throw new Error("expected a timeout signal");
+      }
+      await new Promise<void>((_resolve, reject) => {
+        signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+      });
+    };
+    const emit = withTimeout(emitter, 10_000);
+    const sendOptions = { label: "order-123", signal: controller.signal };
+
+    const emitted = emit(fixture, sendOptions);
+    controller.abort(reason);
+
+    expect(await errorFrom(emitted)).to.equal(reason);
+    expect(received?.label).to.equal("order-123");
+    expect(received?.signal).not.to.equal(controller.signal);
+    expect(sendOptions.signal).to.equal(controller.signal);
+  });
+
+  it("Starts a fresh timeout for every retry attempt when wrapped inside withRetry", async () => {
+    const signals: Array<AbortSignal | undefined> = [];
+    let attempts = 0;
+    const emitter: EmitterFunction<string, TestSendOptions> = async (_event, options) => {
+      signals.push(options?.signal);
+      if (attempts++ === 0) {
+        throw new HTTPTransportError("network");
+      }
+      return "accepted";
+    };
+    const emit = withRetry(withTimeout(emitter, 10_000), { retryDelay: () => 0 });
+
+    expect(await emit(fixture)).to.equal("accepted");
+    expect(signals).to.have.length(2);
+    expect(signals[0]).not.to.equal(undefined);
+    expect(signals[1]).not.to.equal(signals[0]);
+  });
+
+  it("Validates its emitter and timeout", () => {
+    const emitter: EmitterFunction<void> = async () => undefined;
+
+    expect(() => withTimeout(undefined as unknown as EmitterFunction<void>, 100)).to.throw(
+      TypeError, "An EmitterFunction is required",
+    );
+    for (const timeout of [-1, 1.5, Number.POSITIVE_INFINITY, Number.NaN]) {
+      expect(() => withTimeout(emitter, timeout), `${timeout}`).to.throw(
+        TypeError, "timeoutMs must be a non-negative safe integer",
+      );
+    }
+    for (const timeout of [2_147_483_648, Number.MAX_SAFE_INTEGER]) {
+      expect(() => withTimeout(emitter, timeout), `${timeout}`).to.throw(
+        RangeError, "timeoutMs cannot be greater than 2147483647",
+      );
+    }
+    expect(() => withTimeout(emitter, 0)).not.to.throw();
+    expect(() => withTimeout(emitter, 2_147_483_647)).not.to.throw();
+  });
+
+  it("Keeps the built-in HTTP emitter's per-send response type", () => {
+    const emit = withTimeout(
+      emitterFor(httpTransport("https://events.example.com/orders")),
+      1000,
+    );
+    const inferred = () => emit(fixture, {
+      responseHandler: async () => ({ accepted: true }),
+    });
+    const typed: () => Promise<HTTPTransportResponse<{ accepted: boolean }>> = inferred;
+
+    expect(typed).to.be.a("function");
+  });
+});
+
+async function errorFrom(promise: Promise<unknown>): Promise<unknown> {
+  try {
+    await promise;
+  } catch (error) {
+    return error;
+  }
+  throw new Error("Expected the promise to reject");
+}
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: [