Add custom background tasks in worker#923
Merged
Merged
Conversation
Generalize Fedify's enqueue-and-process-later pattern, previously limited to outgoing activity delivery, to arbitrary application-defined background jobs. `Federation` and `FederationBuilder` gain `defineTask()` (via the new `TaskRegistry` interface), and `Context` gains `enqueueTask()`/`enqueueTaskMany()`. Each task carries a Standard Schema that infers the payload type and validates it both at enqueue time and at dequeue time, guarding against schema drift across deployments. Payloads are serialized with devalue so that `Date`, `Map`, `Set`, `URL`, `bigint`, circular references, and Activity Vocabulary objects round-trip faithfully across every message queue backend. Failed handlers retry with exponential backoff by default, configurable per task or federation-wide, and tasks can be isolated onto a dedicated queue or fall back to the outbox queue. The payload codec is implemented twice on purpose: `codec.ts` as a class (`TaskCodec`) and `codec-fn.ts` as standalone utility functions, each with its own tests. Only the class is wired into the runtime; the functional variant is kept temporarily so the team can compare the two styles and decide which reads better before one is removed. #206 #797 Assisted-by: Claude Code:claude-opus-4-8
Split the monolithic `install` task into `install:deno` and `install:pnpm`, with `codegen` as an explicit dependency, so each runtime's setup can be run on its own. `test:deno` now depends on `install:deno` instead of `prepare`, since Deno runs the TypeScript sources directly and does not need the build step. Update AGENTS.md to match: document `mise run prepare`/`prepare-each` for building, `check-each` and `test-each` for scoping work to specific packages, and add a section directing agents to consult `mise tasks`. Assisted-by: Claude Code:claude-opus-4-8
`isPlainObject` in the task codec only accepted objects whose prototype is exactly `Object.prototype`, so an object made with `Object.create(null)` was treated as a non-plain leaf. Any vocab object nested inside such an object was therefore left as its parked holder instead of being revived, even though devalue round-trips null-prototype objects without throwing. Accept a `null` prototype as well, and add a regression test that round-trips a vocab object nested in an `Object.create(null)` object. #803 (comment) Assisted-by: Claude Code:claude-opus-4-8
When a custom task handler throws and the queue does not own retries, the error path computes the elapsed time from `message.started` to feed the retry policy. `message.started` is normally a valid ISO instant set at enqueue time, but a corrupted or drifted queue could hand back an invalid string, in which case `Temporal.Instant.from()` threw out of the error-handling block. That masked the original handler error and aborted the retry, silently dropping the task. Wrap the parse in a try-catch, fall back to a zero elapsed time, and log the offending value. A regression test drives a message with a malformed `started` through a throwing handler and asserts the retry is still enqueued. #803 (comment) Assisted-by: Claude Code:claude-opus-4-8
`FederationBuilderImpl.taskDefinitions` was a plain object, so the duplicate check `name in this.taskDefinitions` and the lookups `this.taskDefinitions[taskName]` consulted the prototype chain. Task names are arbitrary user-supplied strings, so a name such as "constructor", "toString", or "__proto__" was wrongly reported as already defined and resolved to an inherited method on lookup. Switch the registry to a `Map`, which is immune to prototype keys by construction and avoids the clone footgun where a later spread or `Object.assign` would silently reintroduce the prototype. Sibling registries stay plain objects since they are keyed by controlled values (type-id URLs). Add a regression test covering names that collide with `Object.prototype`. #803 (comment) Assisted-by: Claude Code:claude-opus-4-8
`#revive` mapped every node through all five class revivers, allocating five promises per node and resolving them with `Array.fromAsync` before picking the first truthy result. The class filters are mutually exclusive, so it now finds the single matching reviver and runs only that one, cutting the per-node work to a single promise. This keeps the existing behaviour (cycles, repeated references, and Map/Set/Array/plain-object/null-prototype containers all still round-trip, as the codec tests assert) and folds the rationale for two declined suggestions into a comment: the walked tree is devalue's throwaway parse output, so there is no external identity to preserve and nothing to clone lazily; and a recursion-depth cap is moot because this pass recurses with `await` (unwinding the stack each level) while devalue's own recursive `stringify`/`parse` is the binding limit on nesting and would overflow first. #803 (comment) #803 (comment) Assisted-by: Claude Code:claude-opus-4-8
A task may route to its own queue via `defineTask(name, { queue })`, and
`resolveTaskQueue()` enqueues its messages there, but
`_startQueueInternal()` only listened on the four federation-wide queues
(inbox, outbox, fanout, task). A task queue that was none of those got
no worker, so its messages were never processed even while
`startQueue()` was running.
Collect the distinct dedicated queue instances from the task registry
and start a worker for each, treating them as part of the "task"
selector. Dedupe against the standard queues and against task queues
already started on an earlier call so no instance is listened on twice,
and let a deployment whose only queues are per-task ones still start:
the early return no longer bails out when a dedicated task queue exists.
#803 (comment)
Assisted-by: Claude Code:claude-opus-4-8
@dahlia, the maintainer picks `TaskCodec` because it carries the loader state on the instance at [a comment](#803 (comment)). Therefore remove the *codec-fn.ts*. `TaskCodecLoaders` moved to *codec.ts* because `TaskCodec` use it.
The task payload schema validates on both sides of the queue: at enqueue time and again at dequeue time. The wire therefore carries the validated *output*, which the same schema must re-accept as input, so transforming schemas (e.g., Zod's .transform()) whose output differs in shape from their input cannot round-trip. This constraint was neither documented nor tested; state it in the manual and the schema option's JSDoc, and pin it with a regression test. #803 Assisted-by: Claude Code:claude-fable-5
resolveTaskQueue() returns the fallback queue even for a task name with no registered definition, so enqueuing a handle created by a different federation instance silently succeeded and the worker later dropped the message with only a warning. The task API's contract is to fail fast at the enqueue call site (it already validates the payload there), so check the registry before resolving a queue and throw a TypeError instead. #803 Assisted-by: Claude Code:claude-fable-5
Small follow-ups from review:
- Document that tasks must be defined before startQueue() (or the
first request); workers for dedicated per-task queues are only
registered when the queue machinery starts, so a queue defined
later never gets a worker.
- Return early from the enqueue path when no payloads are given,
instead of reaching enqueueMany()/Promise.all with an empty
batch, whose backend behavior is undefined.
- Rename #enqueueSingular to #encodeTaskMessage; it encodes and
builds a TaskMessage but does not enqueue anything.
- Fix a comment typo in the codec.
#803
Assisted-by: Claude Code:claude-fable-5
The revival dispatch pulled init/set out of a heterogeneous tuple list, losing the correlation between each tuple's filter and its init/set node type, which forced two @ts-ignore suppressions at the call site. Such suppressions hide any future error on those lines, so they are unfit for a permanent implementation. Each entry is now built by a generic classReviver() factory whose single type parameter ties the filter to its init/set, letting the compiler check the calls it previously could not. Also bind the recursive reviver to one inner closure per decode pass instead of allocating a fresh closure on every dispatch. #803 (comment) Assisted-by: Claude Code:claude-fable-5
The __contextData phantom field binds a TaskDefinition handle to its federation's context data type, but as a string-keyed property it leaked into user-facing docs and IDE completions despite its @internal tag. Replace it with a module-private unique symbol key: no value exists at runtime, the marker disappears from completions, and cross-federation handle rejection still type-checks, now guarded by a regression test. Also replace the tasks barrel's wildcard re-export of task.ts with explicit named exports of the six types its consumers actually use, so nothing new falls through the barrel unnoticed. #803 (comment) Assisted-by: Claude Code:claude-fable-5
Automated reviewers keep proposing a fixed recursion depth cap (~100) in TaskCodec's #revive to guard against stack overflow from deeply nested payloads. The concern does not apply: the revive traversal suspends at an await on every level, so nesting depth consumes heap (promise chains) rather than native stack, and a structure deep enough to threaten the stack would fail inside devalue.parse() before #revive ever ran. A cap would only reject legitimate payloads. Add a regression test that round-trips a payload nested 1,000 levels deep—an order of magnitude above any proposed cap—through alternating objects and arrays down to a vocab leaf, so introducing such a cap now fails the suite. #803 (comment) Assisted-by: Claude Code:claude-fable-5
The enqueue guard only checked that the handle's task name existed in the local registry, so once two federation instances defined the same task name, a handle from the other instance slipped through: the local context encoded the payload under the schema carried by the foreign handle while the worker decoded it under the local definition's schema. A payload the local schema would have rejected at enqueue thus landed in the queue anyway, only to be dropped at decode time—defeating the fail-fast purpose the guard exists for. defineTask() now stores the exact handle object it returns alongside the internal definition, and enqueueTask()/enqueueTaskMany() compare that handle by identity. Handles still work on every federation built from the same builder, since build() shares the stored definitions. The cross-federation regression test now covers the same-name case in addition to the undefined-name case. #803 (comment) Assisted-by: Claude Code:claude-fable-5
MockContext.enqueueTask() invoked the handler with the raw input, while production enqueueTask() validates the payload against the task schema and hands the validated output to the handler. Tests written against @fedify/testing therefore accepted payloads that production rejects at enqueue, and observed the raw input rather than the coerced or normalized value a transforming schema produces—masking integration bugs the mock exists to surface. The mock now runs the registered schema's Standard Schema validator before invoking the handler, throwing the same TypeError production throws on failure and passing the validated output through. enqueueTaskMany() inherits this since it delegates to enqueueTask(). Added tests covering a rejected payload, a coercing schema whose validated output reaches the handler, and per-item validation in the batch path. #803 (comment) Assisted-by: Claude Code:claude-fable-5
The @standard-schema/spec import is shared by the fedify and testing packages, so it belongs at the workspace level rather than being declared per package. The root deno.json already lists it and workspace members inherit the root import map, making the copy in the fedify package's deno.json redundant; drop it. The pnpm side already sources the version from the catalog in pnpm-workspace.yaml, with each package.json referencing it as "catalog:". Assisted-by: Claude Code:claude-fable-5
Every other enqueue path (inbox, outbox, fanout, forwarding) calls _startQueueInternal() right before enqueuing unless manuallyStartQueue is set, but #enqueueTasks did not. An application that only uses the custom task API never sends an activity, so with the default configuration its first enqueueTask() accepted the message while no worker ever listened: tasks piled up in the queue unprocessed until startQueue() was called explicitly or an activity happened to be sent. Add the same guard to #enqueueTasks, plus a regression test asserting that the first enqueue starts the task worker exactly once and that a second enqueue does not start another listener. #803 (comment) Assisted-by: Claude Code:claude-fable-5
Production's enqueueTaskMany() validates and encodes every payload with Promise.all() before enqueuing anything, so a batch with one invalid item rejects with no effect. The mock looped enqueueTask() per item instead, invoking handlers for earlier payloads before a later one failed validation—tests could observe a partial processing state that cannot occur in production. Split the definition lookup and the schema validation out of enqueueTask() into helpers, and make enqueueTaskMany() validate the whole batch up front, running handlers only once every payload has passed. The existing batch-validation test now pins that no handler runs at all when the batch rejects. #803 (comment) #803 (comment) Assisted-by: Claude Code:claude-fable-5
Production compares the registered handle by identity (14313a1), so passing a handle from another federation instance throws even when both instances define the same task name. The mock looked definitions up by name only, and defineTask() did not keep the handle it returned, so an identity check was impossible: tests could pass with a handle the real federation rejects. Store the returned handle with the definition and require the enqueued handle to be that very object, with the same error message production uses. A regression test defines the same task name on two mock federations and asserts the foreign handle is rejected without running any handler. #803 (comment) Assisted-by: Claude Code:claude-fable-5
The custom background task APIs added on this branch were annotated with @SInCE 2.3.0, but the release that will include them is not yet decided. Replace those tags with the placeholder 2.x.x so the documentation does not promise a specific version prematurely. Affected APIs: Context.enqueueTask and enqueueTaskMany, the taskRetryPolicy and taskQueueResolution federation options, the task queue option, TaskMessage, and the task definition types (TaskHandler, TaskDefinitionOptions, TaskDefinition, TaskRegistry, and TaskEnqueueOptions). Assisted-by: Claude Code:claude-opus-4-8
Local review of the custom background task PR flagged several
documentation-level problems; no runtime behavior is affected:
- The tasks manual claimed the API ships in Fedify 2.3.0, while the
new APIs' JSDoc had already moved to `@since 2.x.x` because the
containing release is undecided. Align the manual with the JSDoc.
- The manual also claimed a per-task queue defined after the queue
machinery starts "never gets a worker." Without
`manuallyStartQueue`, the next request or enqueue starts the
worker, so soften the claim to match the implementation.
- A comment in codec.test.ts described an instance-level `#seen` map
that does not exist—each `deserialize()` call builds its own
per-decode map—and the first test's title claimed a fresh instance
per operation while the tests share one module-level codec.
Correct both.
- Fix grammar errors in the new *AGENTS.md* paragraph about
`mise tasks`.
#803
Assisted-by: Claude Code:claude-fable-5
Assisted-by: Codex:gpt-5-5
The custom task API's producer side—Context.enqueueTask() and
enqueueTaskMany()—had no direct coverage at the middleware layer. Add
tests that drive a real ContextImpl against a recording queue and assert
on what it enqueues:
- enqueueTask() builds a well-formed task message (type, taskName,
baseUrl, attempt, UUID id, parseable started instant, trace context)
and round-trips a vocab payload through the codec as JSON-LD.
- enqueueTaskMany() routes a multi-item batch through enqueueMany(),
preserving order and forwarding delay/orderingKey, while a
single-item batch uses enqueue() instead.
- When the queue lacks enqueueMany(), the batch falls back to
concurrent single enqueues—verified with a rendezvous queue that
blocks until both are in flight—still preserving order and options.
- An invalid payload anywhere in the batch rejects with a schema
TypeError and enqueues nothing.
To avoid duplicating fixtures, the MockQueue and Standard Schema test
helpers that tasks.test.ts defined inline move to testing/tasks.ts
(re-exported by testing/mod.ts); both suites now import the single
implementation, and the fixture-usage allowlist covers the new file.
#803
Assisted-by: Claude Code:claude-opus-4-8
The array reviver in TaskCodec restored elements with `arr.push(...await Array.fromAsync(node, revive))`. Spreading the revived elements into a single call hits the engine's argument-count limit, so a large enough array throws `RangeError: Maximum call stack size exceeded` during decode. Since the worker drops decode failures without retry, an otherwise-valid payload that enqueued fine is silently lost on the dequeue side. Replace the spread with a per-item loop append, matching the existing Map and Set revivers, so revival no longer depends on the array length. Add a regression test that round-trips a 200,000-element array. #803 (comment) Assisted-by: Claude Code:claude-opus-4-8
Three documentation points raised on the task API review, plus a
regression test backing the Temporal claim:
- The payload codec round-trips devalue's built-in Temporal types
with no extra code, but the supported-payload list omitted them.
List `Temporal` (with `Temporal.Instant` / `Temporal.Duration`
examples) and add a serialize/deserialize round-trip test so the
documented support stays covered.
- The vocab import example used the compatibility path
`@fedify/fedify/vocab`. Switch it to `@fedify/vocab`, matching the
surrounding docs and the current package boundary so copied code
does not bind to a path slated for removal.
- Task payloads now cross durable queue storage and can hold arbitrary
application data. Add a trust-boundary security note to the queue
isolation section: treat the backend and payloads as internal
trusted storage, pass identifiers the worker resolves rather than
long-lived secrets, and use a dedicated task queue with
`taskQueueResolution: "strict"` when isolation is required.
#803 (comment)
#803 (comment)
#803 (comment)
Assisted-by: Claude Code:claude-opus-4-8
Context.enqueueTask() and enqueueTaskMany() now accept a
deduplicationKey requesting at-most-once enqueue for tasks that share
it (new TaskEnqueueOptions.deduplicationKey).
Resolution follows the queue and key-value store capabilities:
- A queue declaring the new MessageQueue.nativeDeduplication owns the
check; the key is forwarded through the new
MessageQueueEnqueueOptions.deduplicationKey.
- Otherwise Fedify applies a best-effort guard through the optional
KvStore.cas primitive under a new taskDeduplication key prefix,
tunable with the new FederationOptions.taskDeduplicationTtl and
taskDeduplicationFallback options.
For enqueueTaskMany(), a single key governs the whole batch. A native
queue that does not implement enqueueMany() cannot express batch-level
at-most-once with a per-message key, so such a multi-item enqueue is
rejected with a TypeError instead of silently leaking duplicates.
Configuration errors that are decidable without a payload (a native
queue lacking enqueueMany, or a closed fallback without cas) are
checked before payloads are validated and encoded, so they reject
before any user schema runs or any key is reserved.
#798
Assisted-by: Claude Code:claude-opus-4-8
2chanhaeng
force-pushed
the
feat/custom-worker
branch
from
31e2786 to
246bd0b
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/manual/opentelemetry.md`:
- Around line 368-374: The queue metric descriptions are still using the old
queue wording, so update the definitions in metrics.ts for
fedify.queue.task.enqueued and fedify.queue.task.failed to match the task-based
descriptions shown in the docs. Locate the metric strings in the metrics setup
(the same area that defines
fedify.queue.task.started/completed/duration/in_flight) and make the wording
consistent with the new semantics so the contract stays aligned.
In `@packages/fedify/src/federation/tasks/tasks.test.ts`:
- Around line 1623-1747: Add coverage for the nativeRetrial branch in
`tasks.test.ts` by testing a non-abort handler failure in the existing
`#listenTaskMessage`-related task processing flow. Create a new `t.step` beside
the current abort cases using `MockQueue({ nativeRetrial: true })`, define a
task whose handler throws a regular error, and assert the error is propagated
while no retry is enqueued. Verify `fedify.queue.task.failed` is recorded with
`fedify.task.failure_reason` set to `handler`, and that the `fedify.task` span
is marked with error status, since this path bypasses `retryPolicy` and should
be classified as a failed handler error rather than retried.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 19da1ebd-f27e-4ed4-8371-94a34568babd
📒 Files selected for processing (19)
AGENTS.mdCHANGES.mddocs/manual/opentelemetry.mddocs/manual/tasks.mdmise.tomlpackages/cli/src/startup.test.tspackages/create/src/package.test.tspackages/fedify/src/federation/metrics.tspackages/fedify/src/federation/middleware.test.tspackages/fedify/src/federation/middleware.tspackages/fedify/src/federation/tasks/codec.test.tspackages/fedify/src/federation/tasks/codec.tspackages/fedify/src/federation/tasks/enqueue.test.tspackages/fedify/src/federation/tasks/enqueue.tspackages/fedify/src/federation/tasks/error.tspackages/fedify/src/federation/tasks/mod.tspackages/fedify/src/federation/tasks/tasks.test.tspackages/fedify/src/federation/tasks/types.tspackages/init/src/package.test.ts
Comment on lines
+1623
to
+1747
| await t.step( | ||
| "records an abort as aborted, without a failure reason or error status", | ||
| async () => { | ||
| const queue = new MockQueue({ nativeRetrial: true }); | ||
| const { federation, recorder, exporter } = instrument({ | ||
| ...baseOptions, | ||
| queue: { task: queue }, | ||
| }); | ||
| federation.defineTask("aborts", { | ||
| schema: stringSchema, | ||
| handler: () => { | ||
| throw globalThis.Object.assign(new Error("shutting down"), { | ||
| name: "AbortError", | ||
| }); | ||
| }, | ||
| }); | ||
| const message = await makeTaskMessage("aborts", "payload"); | ||
| await rejects( | ||
| () => federation.processQueuedTask(undefined, message), | ||
| { name: "AbortError" }, | ||
| ); | ||
|
|
||
| const span = exporter.getSpans("fedify.task")[0]; | ||
| ok(span != null); | ||
| strictEqual(span.attributes["fedify.task.failure_reason"], undefined); | ||
| strictEqual(span.status.code, SpanStatusCode.UNSET); | ||
| strictEqual( | ||
| recorder.getMeasurements("fedify.queue.task.failed").length, | ||
| 0, | ||
| ); | ||
| const durations = recorder.getMeasurements("fedify.queue.task.duration"); | ||
| strictEqual(durations.length, 1); | ||
| strictEqual( | ||
| durations[0].attributes["fedify.queue.task.result"], | ||
| "aborted", | ||
| ); | ||
| }, | ||
| ); | ||
|
|
||
| await t.step( | ||
| "on a non-native queue an aborted handler that gives up is aborted", | ||
| async () => { | ||
| const queue = new MockQueue(); | ||
| const { federation, recorder, exporter } = instrument({ | ||
| ...baseOptions, | ||
| queue: { task: queue }, | ||
| }); | ||
| federation.defineTask("aborts-give-up", { | ||
| schema: stringSchema, | ||
| handler: () => { | ||
| throw globalThis.Object.assign(new Error("shutting down"), { | ||
| name: "AbortError", | ||
| }); | ||
| }, | ||
| retryPolicy: () => null, | ||
| }); | ||
| await federation.processQueuedTask( | ||
| undefined, | ||
| await makeTaskMessage("aborts-give-up", "payload"), | ||
| ); | ||
|
|
||
| strictEqual(queue.enqueued.length, 0); | ||
| const span = exporter.getSpans("fedify.task")[0]; | ||
| strictEqual(span.attributes["fedify.task.failure_reason"], undefined); | ||
| strictEqual(span.status.code, SpanStatusCode.UNSET); | ||
| strictEqual( | ||
| recorder.getMeasurements("fedify.queue.task.failed").length, | ||
| 0, | ||
| ); | ||
| strictEqual( | ||
| recorder.getMeasurements("fedify.queue.task.completed").length, | ||
| 0, | ||
| ); | ||
| const durations = recorder.getMeasurements("fedify.queue.task.duration"); | ||
| strictEqual(durations.length, 1); | ||
| strictEqual( | ||
| durations[0].attributes["fedify.queue.task.result"], | ||
| "aborted", | ||
| ); | ||
| }, | ||
| ); | ||
|
|
||
| await t.step( | ||
| "on a non-native queue an aborted handler is retried, not failed", | ||
| async () => { | ||
| const queue = new MockQueue(); // nativeRetrial: false | ||
| const { federation, recorder, exporter } = instrument({ | ||
| ...baseOptions, | ||
| queue: { task: queue }, | ||
| }); | ||
| federation.defineTask("aborts-soft", { | ||
| schema: stringSchema, | ||
| handler: () => { | ||
| throw globalThis.Object.assign(new Error("shutting down"), { | ||
| name: "AbortError", | ||
| }); | ||
| }, | ||
| retryPolicy: () => Temporal.Duration.from({ milliseconds: 1 }), | ||
| }); | ||
| await federation.processQueuedTask( | ||
| undefined, | ||
| await makeTaskMessage("aborts-soft", "payload"), | ||
| ); | ||
|
|
||
| strictEqual(queue.enqueued.length, 1); // retried, behavior unchanged | ||
| strictEqual(queue.enqueued[0].message.attempt, 1); | ||
| // No `handler` failure leaks onto the span or any failure metric… | ||
| const span = exporter.getSpans("fedify.task")[0]; | ||
| strictEqual(span.attributes["fedify.task.failure_reason"], undefined); | ||
| strictEqual(span.status.code, SpanStatusCode.UNSET); | ||
| strictEqual( | ||
| recorder.getMeasurements("fedify.queue.task.failed").length, | ||
| 0, | ||
| ); | ||
| // …and the swallowed-into-retry attempt records `completed`, matching the | ||
| // inbox/outbox internal-retry convention. | ||
| const completed = recorder.getMeasurements("fedify.queue.task.completed"); | ||
| strictEqual(completed.length, 1); | ||
| strictEqual( | ||
| completed[0].attributes["fedify.task.failure_reason"], | ||
| undefined, | ||
| ); | ||
| }, | ||
| ); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
Missing coverage: nativeRetrial task queue with a non-abort handler error.
The abort tests exercise nativeRetrial: true only with an AbortError. The adjacent branch in #listenTaskMessage (if (queue?.nativeRetrial) throw error;) also fires for ordinary handler errors, skipping retryPolicy entirely and relying on classifyTaskError to record failed/handler. That path isn't covered here.
🧪 Suggested additional test
await t.step(
"on a native-retrial queue a non-abort handler error is failed, not retried",
async () => {
const queue = new MockQueue({ nativeRetrial: true });
const { federation, recorder, exporter } = instrument({
...baseOptions,
queue: { task: queue },
});
federation.defineTask("explodes-native", {
schema: stringSchema,
handler: () => {
throw new Error("boom");
},
});
await rejects(
() =>
federation.processQueuedTask(
undefined,
await makeTaskMessage("explodes-native", "payload"),
),
{ message: "boom" },
);
strictEqual(queue.enqueued.length, 0); // backend owns retries
const failed = recorder.getMeasurements("fedify.queue.task.failed");
strictEqual(failed.length, 1);
strictEqual(failed[0].attributes["fedify.task.failure_reason"], "handler");
const span = exporter.getSpans("fedify.task")[0];
strictEqual(span.status.code, SpanStatusCode.ERROR);
},
);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/fedify/src/federation/tasks/tasks.test.ts` around lines 1623 - 1747,
Add coverage for the nativeRetrial branch in `tasks.test.ts` by testing a
non-abort handler failure in the existing `#listenTaskMessage`-related task
processing flow. Create a new `t.step` beside the current abort cases using
`MockQueue({ nativeRetrial: true })`, define a task whose handler throws a
regular error, and assert the error is propagated while no retry is enqueued.
Verify `fedify.queue.task.failed` is recorded with `fedify.task.failure_reason`
set to `handler`, and that the `fedify.task` span is marked with error status,
since this path bypasses `retryPolicy` and should be classified as a failed
handler error rather than retried.
Layer task-specific telemetry onto the custom background task dispatch path, reusing the queue-task metric pattern and mirroring the existing `http_signatures.failure_reason` enum in metrics.ts. Each dequeued task now runs in a `fedify.task` span that inherits the enqueue site's trace context and carries `fedify.task.name`, `fedify.task.attempt`, and, on a terminal failure, `fedify.task.failure_reason`. The `fedify.queue.task.*` metrics report task runs under the new `"task"` role with the task name and, on failure, a bounded `fedify.task.failure_reason`. To tell the failure reasons apart, `#listenTaskMessage` splits the former `decode()` call into its deserialize and validate phases and returns the decision point that failed: `deserialization`, `validation`, `unknown_task`, or `handler`. A swallowed abort is reported as a graceful interruption, not a failure. The reported `fedify.queue.backend` reflects the resolved queue so it stays accurate under the outbox fallback. Public surface: `QueueTaskRole` gains `"task"`, `QueueTaskCommonAttributes` gains `taskName`, and a new `QueueTaskFailureReason` type plus an optional trailing `failureReason` parameter on `recordQueueTaskOutcome()` carry the reason. `TaskCodec` exposes an instance `validate()` wrapper so the dispatch site can split decoding without importing the class. #799 Assisted-by: Claude Code:claude-opus-4-8
Deno executes the TypeScript sources directly, so `test:deno` spent most of its time on a `build` it never needed: with no dist/ output present, the whole Deno suite passes except the npm packaging regression tests added for #655, which assert that the built package.json entry points of `@fedify/cli`, `@fedify/create`, and `@fedify/init` exist. Those checks guard the npm artifacts, not the Deno runtime, and still run under `test:node` and `test:bun`, which build first—so skip them under Deno and drop the `build` dependency from `test:deno`. `@fedify/lint`'s oxlint integration test already skips itself when *dist/oxlint.js* is absent. Update AGENTS.md to match: document `mise run build`/`prepare-each` for building, `check-each` and `test-each` for scoping work to specific packages, recommend the now build-free `test:deno` as the default test loop during development, and add a section directing agents to consult `mise tasks`. Assisted-by: Claude Code:claude-opus-4-8 Assisted-by: Claude Code:claude-fable-5
The fedify.task span and the fedify.queue.task.* metrics promised that
fedify.task.failure_reason is set only on a terminal failure, but the
task worker attributed a handler failure to every thrown attempt, even
when the retry policy had just scheduled a re-enqueue. A transient
error that later succeeded thus produced a failed measurement and an
ERROR span, inflating failure alerts and diverging from the
inbox/outbox convention where an attempt folded into a retry records
result=completed.
To fix this at the decision point instead of patching each call site,
the task listener now returns a dispatch result that distinguishes the
outcome:
- A handler error folded into a scheduled retry records completed.
- Only a terminal give-up records failed with the handler reason.
- An aborted attempt that the retry policy abandons records aborted
rather than completed, so tasks dropped during a graceful shutdown
are no longer invisible to telemetry.
TaskCodec.decode() now reports which phase failed (deserialization or
validation) instead of throwing, so the worker no longer re-composes
the codec's deserialize/validate pipeline inline to tell the two
apart; the redundant instance validate() wrapper is removed.
The retry-path test now pins the decided semantics (no failed
measurement, span status UNSET), a new test covers the abandoned-abort
case, and the manual's failed/aborted definitions are updated to match.
#812 (comment)
#812 (comment)
Assisted-by: Claude Code:claude-fable-5
Assisted-by: Claude Code:claude-opus-4-8
Assisted-by: Codex:gpt-5.5
The fedify.queue.task.enqueued counter was recorded only after the whole dispatch resolved, so on a queue without enqueueMany a batch fanned out through Promise.all could lose measurements: one rejected enqueue aborted the batch before any recording, leaving messages that had already reached the backend uncounted and skewing the enqueued-versus-completed reconciliation. The fan-out path now uses Promise.allSettled and records each message right after its individual enqueue succeeds, mirroring how the outbox delivery path counts partial successes, and rethrows the first rejection afterwards. The single-message and enqueueMany paths record the whole batch with one counter add carrying the batch size, so recordQueueTaskEnqueued() gains an optional count parameter instead of being called once per identical-attribute message. FederationImpl also gains a metrics getter wrapping the memoized getFederationMetrics() lookup, replacing the scattered per-call-site invocations. #812 (review) Assisted-by: Claude Code:claude-fable-5 Assisted-by: Codex:gpt-5.5
Version errors are found by `Claude Code:fable-5` #812 (comment)
Addresses maintainer review feedback on the custom-task
observability additions. All four changes clarify existing
semantics; none change telemetry behavior.
- The `fedify.queue.task.failed` metric summary said tasks are
counted as failed "because processing threw," which excluded the
dispatch-stage drops (`deserialization`, `validation`,
`unknown_task`) that are recorded as failed without the handler
throwing. The summary now covers both a processing throw and a
dispatch drop.
- The `fedify.queue.task.enqueued` counter is emitted at the enqueue
site, not during the worker run, so grouping it under "task run
measurement" was misleading. The docs now state that `enqueued`
is recorded at the enqueue site while `started`, `completed`,
`failed`, and `duration` are recorded during the worker run.
- A short operational example now precedes the OpenTelemetry
cross-reference, showing how to split schema drift from handler
give-ups via `fedify.task.failure_reason` and how to spot retry
re-enqueues via `fedify.queue.task.attempt > 0`.
- The `completed` outcome returned after scheduling a retry now
carries a comment explaining it means "the attempt was folded
into a retry," not "the handler succeeded," to keep a future
change from regressing the terminal-failure-only task telemetry.
#812 (comment)
#812 (comment)
#812 (comment)
#812 (comment)
Assisted-by: Claude Code:claude-opus-4-8
When a custom task's handler failed and a retry was scheduled, but
re-enqueuing the retry message to the queue backend itself threw, the
worker boundary recorded `fedify.task.failure_reason` as `handler` and
re-threw the enqueue error, shadowing the original handler error. An
infrastructure failure was thus indistinguishable from a handler fault,
and the documented failure reasons map to dispatch decision points, none
of which is a re-enqueue failure.
Because the failure-reason set is a small bounded set that is open to
refinement, this adds a distinct `retry_enqueue` value rather than
folding the case into `handler`. One extra value keeps the metric
cardinality bounded (`taskName x |failure_reason| x backend`).
- `QueueTaskFailureReason` gains `retry_enqueue`.
- The retry re-enqueue is now wrapped narrowly; on failure the worker
logs it and re-throws a `TaskRetryEnqueueError` that preserves the
original handler error as its `cause`. The throw keeps the message
nacked rather than dropped, and the worker boundary records
`retry_enqueue` with the handler error surfaced on the span.
- `TaskRetryEnqueueError` and `QueueTaskDispatchResult` move out of
middleware.ts into dedicated `tasks/error.ts` and `tasks/types.ts`
modules, re-exported through the task barrel.
- The tasks and OpenTelemetry manuals document the new value, and a
worker test covers the re-enqueue-failure path.
Assisted-by: Claude Code:claude-opus-4-8
The fanout, outbox, inbox, and task branches of FederationImpl.processQueuedTask() each repeated the same consumer-span and metric boilerplate: open a CONSUMER span, scope the extracted trace context, record the started/outcome/duration metrics, pair the in-flight increment with its decrement, set the span status, and end the span. The task branch had also drifted from the other three, recording its outcome outside the finally block that guarantees it. Collapse the four copies onto two helpers. #runWorkerSpan() opens the span and scopes the trace context; #instrumentWorkerBody() wraps a worker body with the shared boundary telemetry. Each branch now supplies only its span name, span attributes, common metric attributes, body, and error classifier. The outcome record lives in one finally block again, so a "failed" outcome that arrives through the body's return value (a dropped or given-up task) is instrumented the same way as one that arrives through a thrown error. Unify the former WorkerSpanOutcome type into QueueTaskDispatchResult (failureReason and error are now optional on the failed variant), so the task dispatch result and the worker-boundary outcome share one type, and move the classifyAbortableError and classifyTaskError classifiers into tasks/error.ts. #812 Assisted-by: Claude Code:claude-opus-4-8
Running only `mise run test:deno` by itself is not a problem. However, when running `mise run test`, `test:deno` and `build` would run simultaneously, often causing errors. To prevent this, the `wait_for` attributes are added in the `test:deno` task with `build` item.
The test drove the first marker's expiry through a real 1 ms taskDeduplicationTtl, but that TTL equally applied to the marker the second enqueue re-claimed. Every assertion after the re-claim therefore had to run within 1 ms of wall clock before MemoryKvStore's lazy expiry check judged the marker expired. Local runs finish those microtask hops in time, but under CI load—especially `mise run test:bun`, which runs every package's tests in parallel on a Temporal polyfill—the window was routinely blown, failing the test intermittently. Simulate the first marker's expiry deterministically by deleting the key instead, and stretch the TTL to one minute so the second marker comfortably outlives the assertions. This removes the wall-clock dependence entirely and also drops the 20 ms delay. Assisted-by: Claude Code:claude-fable-5
2chanhaeng
force-pushed
the
feat/custom-worker
branch
from
246bd0b to
6eb234c
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/manual/opentelemetry.md`:
- Around line 1010-1011: Update the `fedify.queue.role` row in the OpenTelemetry
summary table to include `task` in the documented enum so it matches the
task-telemetry section. Keep the existing identifiers and wording around
`fedify.queue.role` and `fedify.queue.backend`, and expand the role list to
cover `inbox`, `outbox`, `fanout`, `shared`, and `task`.
In `@packages/fedify/src/federation/tasks/types.ts`:
- Around line 3-9: `QueueTaskDispatchResult` is a public exported type but lacks
JSDoc, so add a doc comment above it in `tasks/types.ts` describing each
`outcome` case and the meaning of `failureReason` and `error`. Use the existing
documentation style from `QueueTaskFailureReason` as a guide, and ensure the
comment clearly explains the semantics of the `completed`, `aborted`, and
`failed` variants.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 5d71d9fd-c9d6-44d6-882c-f40ec1d351d4
📒 Files selected for processing (19)
AGENTS.mdCHANGES.mddocs/manual/opentelemetry.mddocs/manual/tasks.mdmise.tomlpackages/cli/src/startup.test.tspackages/create/src/package.test.tspackages/fedify/src/federation/metrics.tspackages/fedify/src/federation/middleware.test.tspackages/fedify/src/federation/middleware.tspackages/fedify/src/federation/tasks/codec.test.tspackages/fedify/src/federation/tasks/codec.tspackages/fedify/src/federation/tasks/enqueue.test.tspackages/fedify/src/federation/tasks/enqueue.tspackages/fedify/src/federation/tasks/error.tspackages/fedify/src/federation/tasks/mod.tspackages/fedify/src/federation/tasks/tasks.test.tspackages/fedify/src/federation/tasks/types.tspackages/init/src/package.test.ts
Comment on lines
+1010
to
+1011
| | `fedify.queue.role` | string | The Fedify queue role: `inbox`, `outbox`, `fanout`, or `shared` for queue depth rows where one queue backs multiple roles. | `"outbox"` | | ||
| | `fedify.queue.backend` | string | The queue implementation's constructor name (best-effort backend identifier). | `"RedisMessageQueue"` | |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add task to the fedify.queue.role enum here.
The task-telemetry section above already documents fedify.queue.role = task, so this summary table is now inconsistent.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/manual/opentelemetry.md` around lines 1010 - 1011, Update the
`fedify.queue.role` row in the OpenTelemetry summary table to include `task` in
the documented enum so it matches the task-telemetry section. Keep the existing
identifiers and wording around `fedify.queue.role` and `fedify.queue.backend`,
and expand the role list to cover `inbox`, `outbox`, `fanout`, `shared`, and
`task`.
Comment on lines
+3
to
+9
| export type QueueTaskDispatchResult = | ||
| | { readonly outcome: "completed" | "aborted" } | ||
| | { | ||
| readonly outcome: "failed"; | ||
| readonly failureReason?: QueueTaskFailureReason; | ||
| readonly error?: unknown; | ||
| }; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Missing JSDoc for exported type.
QueueTaskDispatchResult is exported but undocumented, unlike the adjacent QueueTaskFailureReason in metrics.ts which carries a full doc comment explaining each variant. Consider documenting the outcome/failureReason/error semantics here too.
As per coding guidelines, **/*.{ts,tsx}: "Include JSDoc comments for public APIs and update documentation when public APIs change."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/fedify/src/federation/tasks/types.ts` around lines 3 - 9,
`QueueTaskDispatchResult` is a public exported type but lacks JSDoc, so add a
doc comment above it in `tasks/types.ts` describing each `outcome` case and the
meaning of `failureReason` and `error`. Use the existing documentation style
from `QueueTaskFailureReason` as a guide, and ensure the comment clearly
explains the semantics of the `completed`, `aborted`, and `failed` variants.
Source: Coding guidelines
dahlia
requested changes
Jul 7, 2026
…d the final PR number in changelog
dahlia
requested changes
Jul 7, 2026
By [JSR #1473](jsr-io/jsr#1473), the docs building task broken. For that, @dahlia updates [`markdown-it-jsr-ref`](dahlia/markdown-it-jsr-ref#1). Assisted-by: Claude Code:claude-opus-4-8
dahlia
approved these changes
Jul 7, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Resolves #206.
Summary
This PR merges the
feat/custom-workerbranch, where the custom worker work has been carried out, intomainnow that the work is complete. In short, it adds custom background task support to Fedify. For the detailed work history, see each sub-issue and PR:defineTask/enqueueTaskAPIdeduplicationKey)Most of the design discussion and review already took place in the sub-issues and PRs above. The work on the
feat/custom-workerbranch was carried out independently ofmain. This PR was opened to mergefeat/custom-workerintomainwhile preventing conflicts. Please focus the review on whether the two branches have been merged together without problems.Changes
See CHANGES.md and each PR above for the detailed change list:
defineTask/enqueueTaskAPI #797 → Add custom background task API (defineTask/enqueueTask) #803): define schema-validated tasks withdefineTask(), enqueue them from anyContextwithenqueueTask()/enqueueTaskMany(), and run them on the queue worker with retry support.TaskEnqueueOptions.deduplicationKeyenables at-most-once enqueue, backed byMessageQueue.nativeDeduplicationor a best-effortKvStore.casfallback.fedify.taskspan, and thefedify.queue.task.*metrics gain ataskrole with a boundedfailure_reasonenum.Rider changes
Two small tooling changes ride along (reviewed within the sub-PRs):
test:denonow useswait_for = ["build"]instead ofdepends = ["build"]—Deno executes TypeScript directly so the task doesn't need a build, but it must not race one that is already running in parallel.test:deno {PATH} --filter,check-each, and a pointer tomise tasks).AI disclosure
This PR body was drafted in English by AI based on the user's instructions and input, and was finalized after human review and adjustment. Individual commits carry their own
Assisted-bytrailers, as disclosed in #803, #806, and #812:Assisted-by: Claude Code:claude-opus-4-8
Assisted-by: Claude Code:claude-fable-5
Assisted-by: Codex:gpt-5.5