Keep WorkersMessageQueue sends alive with a waitUntil option #1386

Motivation

Cloudflare Workers can cancel a promise that is neither awaited nor passed to ExecutionContext.waitUntil() when the invocation ends. WorkersMessageQueue.enqueue() and enqueueMany() already await Queue.send() and Queue.sendBatch(), so Fedify's own enqueues are covered. Application code that calls the queue directly has no supported way to keep the queue send alive. A promise dropped before the enqueue reaches the adapter is a separate problem, which #1383 addresses.

The external report in iwasa-kosui/monorepo#533 works around this with a queue wrapper that registers every enqueue() and enqueueMany() promise with ctx.waitUntil(). The wrapper is correct, but every application has to write and test it.

Proposed API

Add an optional callback to WorkersMessageQueueOptions in packages/cfworkers/src/mod.ts:

export interface WorkersMessageQueueOptions {
  // ... existing options ...

  /**
   * Keeps `Queue.send()` and `Queue.sendBatch()` alive after the current
   * invocation ends.  Pass `ctx.waitUntil.bind(ctx)` or `waitUntil` from
   * `cloudflare:workers`.  Only this adapter's sends are covered.
   * @since 2.0.x
   */
  readonly waitUntil?: (promise: Promise<void>) => void;
}

The implementation registers each send while still awaiting the original promise:

#keepAlive(promise: Promise<unknown>): void {
  if (this.#waitUntil == null) return;
  const settled = promise.then(() => undefined, () => undefined);
  try {
    this.#waitUntil(settled);
  } catch {
    // A lifetime hook must not fail the send.
  }
}

async enqueue(message, options) {
  const sending = this.#queue.send(wrapped, { ... });
  this.#keepAlive(sending);
  await sending;
}

enqueueMany() calls #keepAlive(sending) for each sendBatch() in its flush() helper. The original promise is still awaited, so a caller sees the same rejection as before. The promise passed to waitUntil() never rejects, matching the core callback proposed in #1383.

Usage:

const queue = new WorkersMessageQueue(env.QUEUE, {
  waitUntil: ctx.waitUntil.bind(ctx),
});

When the queue is constructed where cloudflare:workers is available, the imported function can be passed directly:

import { waitUntil } from "cloudflare:workers";
const queue = new WorkersMessageQueue(env.QUEUE, { waitUntil });

Behavior

• The callback is invoked synchronously inside enqueue() or enqueueMany(), before the method returns.
• It receives a derived Promise<void> that resolves when the send settles and never rejects.
• The original send promise is unchanged, so await queue.enqueue() still throws on failure.
• A callback that throws is ignored; the send is not affected.
• When the option is omitted, behavior is identical to today.

What this does not cover

This is a queue adapter safety net, not a general request-lifetime API. It does not keep alive work that is canceled before queue.send() is reached, such as key export, JSON-LD compaction, or fanout message construction. It does not cover direct HTTP delivery with no queue, the sibling-delivery drain in #1380, the automatic queue startup rejection in #1381, or FEP-ef61 portable forwarding after its deadline. #1383 remains the general solution.

Why this is a 2.0-maintenance exception

Adding an optional property is a public API change, even though it is backward compatible. 2.0-maintenance normally takes only bug fixes. This issue is a deliberate exception, not a precedent. The external workaround is already deployed, the change is small and isolated to @fedify/cfworkers, and the option name and semantics match the core waitUntil callback planned for 2.5.0. Waiting for 2.5.0 would leave direct WorkersMessageQueue users without a supported path in the meantime; the exception gives them one without changing the behavior of existing code.

Scope

• packages/cfworkers/src/mod.ts: WorkersMessageQueueOptions, the constructor, enqueue() and enqueueMany().
• packages/cfworkers/src/mod.test.ts or the package's workerd tests.
• docs/manual/mq.md and the Cloudflare Workers example.
• Use the next 2.0.x patch version for @since. Forward-port the same option to main and 2.5.0 without renaming it.

Tests

• A fake Queue whose send() resolves after a delay: the callback is called synchronously and the registered promise resolves after the send.
• A failing send(): enqueue() rejects, the registered promise resolves, and no unhandled rejection is reported.
• enqueueMany() with enough messages to produce multiple batches: every sendBatch() is registered.
• No waitUntil option: the fake queue sees the same calls and the adapter behaves as before.