feat: add configurable retries for emitters

API_TRANSITION_GUIDE.md

@@ -102,3 +102,164 @@ 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, 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`. Pass
+`headers` and `signal` with the event for settings that apply to one message.
+
+```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 RequestInit` 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()` accepts Fetch header forms instead.
+
+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,
+  });
+}
+```
+
+Leave the built-in transport and emitter types to inference so their message
+options and response body types are preserved:
+
+```ts
+const transport = httpTransport(sink);
+const emit = emitterFor(transport);
+```

README.md

@@ -51,7 +51,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
 import { CloudEvent, emitterFor, httpTransport } from "cloudevents";
@@ -62,10 +63,151 @@ 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, a
+`responseHandler` to read the body another way, and the remaining Fetch options
+under `fetchOptions`. The options passed to `emit()` apply to that Message only:
+`headers` and `signal`. Unsupported transport or message options reject with a
+`TypeError`. To discard a response body, use
+`responseHandler: async (response) => { await response.body?.cancel(); }`.
+
+Existing `ce-*` and `content-type` headers cannot be changed. New CloudEvents
+headers must be valid and, in structured mode, match the body.
+
+Other headers, such as `authorization`, `traceparent` and `baggage`, can be added
+or overridden. `host` and `content-length` are controlled by Fetch and are
+rejected. Invalid headers reject with a `TypeError` before the request is sent.
+
+There is no timeout option: a deadline is an `AbortSignal` passed with the
+message.
+
+```js
+const emit = emitterFor(httpTransport("https://my.receiver.com/endpoint", {
+  headers: { authorization: `Bearer ${process.env.RECEIVER_TOKEN}` },
+  responseHandler: (response) => response.json(),
+}));
+
+async function main() {
+  await emit(ce, { signal: AbortSignal.timeout(10000) });
+}
+
+main().catch(console.error);
+```
+
+A 2xx response means the receiver accepted the event, so a body that cannot be
+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 } from "cloudevents";
+
+const emit = withRetry(
+  emitterFor(httpTransport("https://events.example.com/orders")),
+  { maxAttempts: 5 },
+);
+const orderCreated = new CloudEvent({
+  type: "com.example.order.created",
+  source: "/orders",
+  data: { orderId: "order-123" },
+});
+
+async function main() {
+  await emit(orderCreated, { signal: AbortSignal.timeout(30000) });
+}
+
+main().catch(console.error);
+```
+
+The default retry policy handles failures which are commonly temporary:
+
+| Failure reported by the emitter | Retried by default |
+| ------------------------------- | ------------------ |
+| Network failure before a response | Yes |
+| HTTP 408, 425, 429, 500, 502, 503 or 504 | Yes |
+| An aborted send or another HTTP status | No |
+| An error from a custom transport | No |
+
+Retries use randomized exponential backoff. For HTTP 429 and 503, a valid
+`Retry-After` header takes precedence, and `maxRetryDelay` caps whatever a delay
+asks for - 30 seconds by default, so a distant `Retry-After` cannot park an
+emitter for hours. Pass `shouldRetry(error, context)` or
+`retryDelay(error, context)` to replace either policy. The exported
+`isRetryableHTTPError()` and `defaultRetryDelay()` functions let a custom policy
+reuse the defaults.
+
+The signal passed with the event also ends a wait between attempts. The emitter
+then rejects with what that signal carries, rather than with the failure which
+led to the retry.
+
+A network failure can happen after the receiver accepted an event but before
+its response arrived. Every attempt therefore uses the same CloudEvent and the
+same `source` and `id`, but the receiver still needs to recognize duplicate
+delivery.
+
+The [API transition guide](./API_TRANSITION_GUIDE.md) covers the header forms
+the transport accepts and where a proxy or a custom CA goes now that
+`http.globalAgent` no longer applies.
+
 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
@@ -121,20 +263,31 @@ import {
   Mode,
 } from "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:

src/event/cloudevent.ts

@@ -62,10 +62,10 @@ export class CloudEvent<T = undefined> implements CloudEventV1<T> {
     delete properties.time;
 
     this.type = properties.type as string;
-    delete (properties as any).type;
+    delete properties.type;
 
     this.source = properties.source as string;
-    delete (properties as any).source;
+    delete properties.source;
 
     this.specversion = (properties.specversion) || V1;
     delete properties.specversion;