Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions CHANGES.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,18 @@ To be released.
fanout before the upgrade also keep their original proof-covered JSON when
replayed. Incoming signatures that still use `identity/v1` continue to
verify. [[#664]]
- Fixed automatic queue worker startup failures becoming unhandled promise
rejections. When `manuallyStartQueue` was left `false` and a `MessageQueue`
implementation could not start listening, such as `WorkersMessageQueue` on
Cloudflare Workers whose `listen()` throws a `TypeError`, the rejection from
the implicit startup in inbox handling and in `Context.sendActivity()` was
discarded. It is now logged as a warning once per `Federation` instance,
and a synchronous throw from one queue no longer prevents the remaining
queues from starting. The failed worker is not retried, so restart the
process after fixing the queue. `Federation.startQueue()` still rejects so
callers that start workers deliberately see the error, and Cloudflare
Workers deployments should set `manuallyStartQueue: true` to avoid the
warning entirely. [[#1381], [#1390]]
- Fixed inbox handling that treated valid date-times without seconds, such
as `2024-01-01T10:00Z` or `2024-01-01T10:00.5Z`, as permanently malformed.
Activity Streams 2.0 allows seconds to be omitted, so activities with such
Expand All @@ -34,6 +46,8 @@ To be released.
[#664]: https://github.com/fedify-dev/fedify/issues/664
[#1339]: https://github.com/fedify-dev/fedify/issues/1339
[#1347]: https://github.com/fedify-dev/fedify/pull/1347
[#1381]: https://github.com/fedify-dev/fedify/issues/1381
[#1390]: https://github.com/fedify-dev/fedify/pull/1390

### @fedify/vocab

Expand Down
17 changes: 17 additions & 0 deletions changes.d/fedify/queue-startup-failures.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
---
links:
'#1381': https://github.com/fedify-dev/fedify/issues/1381
'#1390': https://github.com/fedify-dev/fedify/pull/1390
---
- Fixed automatic queue worker startup failures becoming unhandled promise
rejections. When `manuallyStartQueue` was left `false` and a `MessageQueue`
implementation could not start listening, such as `WorkersMessageQueue` on
Cloudflare Workers whose `listen()` throws a `TypeError`, the rejection from
the implicit startup in inbox handling and in `Context.sendActivity()` was
discarded. It is now logged as a warning once per `Federation` instance,
and a synchronous throw from one queue no longer prevents the remaining
queues from starting. The failed worker is not retried, so restart the
process after fixing the queue. `Federation.startQueue()` still rejects so
callers that start workers deliberately see the error, and Cloudflare
Workers deployments should set `manuallyStartQueue: true` to avoid the
warning entirely. [[#1381], [#1390]]
16 changes: 13 additions & 3 deletions docs/manual/deploy.md
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,9 @@ export default {
const federation = await builder.build({
kv: new WorkersKvStore(env.KV_NAMESPACE),
queue: new WorkersMessageQueue(env.QUEUE),
// Cloudflare Queues cannot be polled, so the queue() handler below
// consumes the messages instead of Federation.startQueue():
manuallyStartQueue: true,
// Other options...
});

Expand All @@ -371,10 +374,16 @@ export default {

Cloudflare Queues don't provide polling-based APIs, so the `WorkersMessageQueue`
cannot implement a traditional `~MessageQueue.listen()` method. Instead, you
must manually connect queue handlers:
must manually connect queue handlers.

Because `WorkersMessageQueue.listen()` throws a `TypeError`, set
`~FederationOptions.manuallyStartQueue` to `true` as well. Otherwise Fedify
tries to start the queue workers automatically the first time an activity is
received or sent, and logs a warning through [LogTape](./log.md) that the
workers could not be started. The `queue()` handler below is what actually
processes the messages, so the option only prevents that unsuccessful attempt:

~~~~ typescript twoslash
// @noErrors: 2345
import { createFederationBuilder, type Message } from "@fedify/fedify";
import { Person } from "@fedify/vocab";
import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers";
Expand All @@ -394,13 +403,14 @@ export default {
const federation = await builder.build({
kv: new WorkersKvStore(env.KV_NAMESPACE),
queue: new WorkersMessageQueue(env.QUEUE),
manuallyStartQueue: true,
});

for (const message of batch.messages) {
try {
await federation.processQueuedTask(
message.body as unknown as Message,
env,
message.body as unknown as Message,
);
message.ack();
} catch (error) {
Expand Down
15 changes: 14 additions & 1 deletion docs/manual/federation.md
Original file line number Diff line number Diff line change
Expand Up @@ -194,13 +194,26 @@ Further details are explained in the [*Message queue* section](./mq.md).
Whether to start the task queue manually or automatically.

If `true`, the task queue will not start automatically and you need to
manually start it by calling the `Federation.startQueue()` method.
manually start it by calling the `Federation.startQueue()` method. On runtimes
that cannot start workers in the request process, such as Cloudflare Workers,
`Federation.processQueuedTask()` consumes the messages instead, so you still set
this option to `true`.

If `false`, the task queue will start automatically as soon as the first
task is enqueued.

By default, the queue starts automatically.

> [!NOTE]
> Automatic startup is best-effort. If a queue cannot start listening
> (for example, when `MessageQueue.listen()` throws), Fedify logs a warning
> through [LogTape](./log.md) once per `Federation` instance instead of
> producing an unhandled promise rejection. The failed worker is not retried by
> later requests, and the application keeps running without it, so restart the
> process or recreate the federation after fixing the queue. Explicitly calling
> `Federation.startQueue()` reports the error to the caller instead of logging
> it.

> [!TIP]
> This option is useful when you want to separately deploy the web server
> and the task queue worker. In this case, you can start the task queue
Expand Down
14 changes: 14 additions & 0 deletions docs/manual/mq.md
Original file line number Diff line number Diff line change
Expand Up @@ -477,6 +477,9 @@ export default {
kv: undefined as unknown as KvStore,
// ---cut-end---
queue: new WorkersMessageQueue(env.QUEUE_BINDING),
// Cloudflare Queues cannot be polled, so the queue() handler below
// consumes the messages instead of Federation.startQueue():
manuallyStartQueue: true,
});
// Omit the rest of the code for brevity
},
Expand All @@ -491,6 +494,10 @@ export default {
kv: undefined as unknown as KvStore,
// ---cut-end---
queue: new WorkersMessageQueue(env.QUEUE_BINDING),
// Inbox listeners run inside processQueuedTask() and may call
// ctx.sendActivity(), which would otherwise start the queue workers
// automatically:
manuallyStartQueue: true,
});
for (const msg of batch.messages) {
await federation.processQueuedTask(
Expand Down Expand Up @@ -521,6 +528,13 @@ export default {
> method, you need to call `Federation.processQueuedTask()` method to manually
> process the messages. The `queue()` method is the only way to consume
> messages from the queue in Cloudflare Workers.
>
> Set `~FederationOptions.manuallyStartQueue` to `true` for the same reason.
> Otherwise Fedify tries to start the queue workers automatically the first time
> an activity is received or sent, and since `listen()` throws, it logs a
> warning through [LogTape](./log.md) that the workers could not be started.
> The workers are not started on this runtime either way; the option prevents
> that unsuccessful attempt and the warning.

[Cloudflare Workers]: https://workers.cloudflare.com/
[Cloudflare Queues]: https://developers.cloudflare.com/queues/
Expand Down
7 changes: 7 additions & 0 deletions examples/cloudflare-workers/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,9 @@ export default {
const federation: Federation<ContextData> = await builder.build({
kv: new WorkersKvStore(env.FEDIFY_KV),
queue: new WorkersMessageQueue(env.FEDIFY_QUEUE),
// Cloudflare Queues cannot be polled, so the queue() handler below
// consumes the messages instead of Federation.startQueue():
manuallyStartQueue: true,
});
const url = new URL(req.url);
if (url.pathname === "/") return await listFollowers(req, env);
Expand All @@ -163,6 +166,10 @@ export default {
const federation: Federation<ContextData> = await builder.build({
kv: new WorkersKvStore(env.FEDIFY_KV),
queue: new WorkersMessageQueue(env.FEDIFY_QUEUE),
// Inbox listeners run inside processQueuedTask() and may call
// ctx.sendActivity(), which would otherwise start the queue workers
// automatically:
manuallyStartQueue: true,
});
for (const message of batch.messages) {
await federation.processQueuedTask(
Expand Down
43 changes: 27 additions & 16 deletions packages/cfworkers/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -16,35 +16,44 @@ implementations for [Cloudflare Workers]:
- [`WorkersMessageQueue`]

~~~~ typescript
import type { Federation } from "@fedify/fedify";
import { createFederation, type Message } from "@fedify/fedify";
import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers";

type Env = {
KV_BINDING: KVNamespace<string>;
QUEUE_BINDING: Queue;
};

export default {
async fetch(request, env, ctx) {
const federation = createFederation({
async fetch(request: Request, env: Env): Promise<Response> {
const federation = createFederation<Env>({
kv: new WorkersKvStore(env.KV_BINDING),
queue: new WorkersMessageQueue(env.QUEUE_BINDING),
// Cloudflare Queues cannot be polled, so the queue() handler below
// consumes the messages instead of Federation.startQueue():
manuallyStartQueue: true,
// ... other options
});
return federation.handle(request, { contextData: env });

return await federation.fetch(request, { contextData: env });
},
async queue(batch, env, ctx) {
const federation = createFederation({

async queue(batch: MessageBatch<unknown>, env: Env): Promise<void> {
const federation = createFederation<Env>({
kv: new WorkersKvStore(env.KV_BINDING),
queue: new WorkersMessageQueue(env.QUEUE_BINDING),
// Inbox listeners run inside processQueuedTask() and may call
// ctx.sendActivity(), which would otherwise start the queue workers
// automatically:
manuallyStartQueue: true,
// ... other options
});

for (const message of batch.messages) {
await federation.processQueuedTask(message.body);
await federation.processQueuedTask(env, message.body as Message);
}
}
} satisfies ExportedHandler<{
KV_BINDING: KVNamespace<string>;
QUEUE_BINDING: Queue;
}>;
},
} satisfies ExportedHandler<Env>;
~~~~

[JSR badge]: https://jsr.io/badges/@fedify/cfworkers
Expand Down Expand Up @@ -99,7 +108,9 @@ management.
> API when new messages are available in the queue. Inside the `queue()`
> method, you need to call `Federation.processQueuedTask()` method to manually
> process the messages. The `queue()` method is the only way to consume
> messages from the queue in Cloudflare Workers.
> messages from the queue in Cloudflare Workers. Set `manuallyStartQueue` to
> `true` as well; otherwise Fedify tries to start the workers automatically,
> and since `listen()` throws, it logs a warning that they could not be started.

[Cloudflare Queues]: https://developers.cloudflare.com/queues/

Expand Down
6 changes: 6 additions & 0 deletions packages/fedify/src/federation/federation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -814,6 +814,12 @@ export interface FederationOptions<TContextData> {
*
* By default, the queue starts automatically.
*
* Automatic startup is best-effort: if a queue cannot start listening,
* a warning is logged once per federation instance and the failed worker is
* not retried, while an explicit {@link Federation.startQueue} call rejects.
* Runtimes that consume the queue elsewhere, such as Cloudflare Workers with
* {@link Federation.processQueuedTask}, should set this to `true`.
*
* @since 0.12.0
*/
manuallyStartQueue?: boolean;
Expand Down
Loading
Loading