Add OpenTelemetry observability to custom background tasks#812
Add OpenTelemetry observability to custom background tasks#8122chanhaeng wants to merge 5 commits into
Conversation
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository UI Review profile: ASSERTIVE Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a custom background task API to Fedify, allowing developers to define, enqueue, and process arbitrary background jobs with type-safe payload validation via Standard Schema. The implementation supports robust serialization of complex types and Activity Vocabulary objects using devalue, customizable retry policies, queue routing, best-effort or native deduplication, and OpenTelemetry instrumentation. Feedback on the changes highlights a compatibility issue with Node.js 20 due to the use of Array.fromAsync in codec.ts, suggesting standard for...of loops instead, and recommends implementing a recursion depth limit during deserialization to prevent potential Denial of Service (DoS) attacks from deeply nested payloads.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| classReviver( | ||
| isInstanceOf(Array), | ||
| (): unknown[] => [], | ||
| async (revive, node, arr) => { | ||
| for (const item of await Array.fromAsync(node, revive)) arr.push(item); | ||
| }, | ||
| ), |
There was a problem hiding this comment.
Using Array.fromAsync introduces compatibility issues with Node.js 20 (which is an active LTS version of Node.js) as it was introduced in ES2024 and is only natively supported in Node.js 22+. To maintain compatibility across all target environments (including Node.js 20), replace it with a standard for...of loop with await.
classReviver(
isInstanceOf(Array),
(): unknown[] => [],
async (revive, node, arr) => {
for (const item of node) {
arr.push(await revive(item));
}
},
),| isInstanceOf(Set), | ||
| () => new Set<unknown>(), | ||
| async (revive, node, set) => { | ||
| for (const v of await Array.fromAsync(node, revive)) set.add(v); | ||
| }, | ||
| ), |
There was a problem hiding this comment.
Using Array.fromAsync introduces compatibility issues with Node.js 20 (which is an active LTS version of Node.js) as it was introduced in ES2024 and is only natively supported in Node.js 22+. To maintain compatibility across all target environments (including Node.js 20), replace it with a standard for...of loop with await.
| isInstanceOf(Set), | |
| () => new Set<unknown>(), | |
| async (revive, node, set) => { | |
| for (const v of await Array.fromAsync(node, revive)) set.add(v); | |
| }, | |
| ), | |
| classReviver( | |
| isInstanceOf(Set), | |
| () => new Set<unknown>(), | |
| async (revive, node, set) => { | |
| for (const v of node) { | |
| set.add(await revive(v)); | |
| } | |
| }, | |
| ), |
| #revive = (seen: Seen): Revive => { | ||
| const inner: Revive = async (node) => { | ||
| if (node === null || typeof node !== "object") return node; | ||
| if (seen.has(node)) return seen.get(node); | ||
| for (const reviver of this.#classRevivers) { | ||
| const out = reviver(seen, inner, node); | ||
| if (out !== undefined) return await out; | ||
| } | ||
| // devalue can handle non-container objects. | ||
| return node; | ||
| }; | ||
| return inner; | ||
| }; |
There was a problem hiding this comment.
To prevent stack overflow or resource exhaustion from maliciously crafted deep JSON payloads, implement a depth limit in recursive traversal functions. The limit should be high enough for legitimate data (e.g., 1000) but low enough to prevent DoS attacks.
#revive = (seen: Seen, maxDepth = 1000): Revive => {
const inner = async (node: unknown, depth = 0): Promise<unknown> => {
if (node === null || typeof node !== "object") return node;
if (seen.has(node)) return seen.get(node);
if (depth >= maxDepth) {
throw new TypeError("Maximum depth limit exceeded during deserialization.");
}
const nextRevive = (n: unknown) => inner(n, depth + 1);
for (const reviver of this.#classRevivers) {
const out = reviver(seen, nextRevive, node);
if (out !== undefined) return await out;
}
// devalue can handle non-container objects.
return node;
};
return inner;
};References
- To prevent stack overflow from maliciously crafted deep JSON, implement a depth limit in recursive traversal functions. The limit should be high enough for legitimate data but low enough to prevent DoS attacks.
Codecov Report❌ Patch coverage is
... and 2 files with indirect coverage changes 🚀 New features to boost your workflow:
|
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.
fedify-dev#798
Assisted-by: Claude Code:claude-opus-4-8
The #enqueueTasks and #encodeTaskMessage methods made ContextImpl oversized, so move the handle validation, deduplication planning, payload encoding, and queue dispatch into a new tasks/enqueue.ts module. ContextImpl now delegates to enqueueTasks(), passing only the small slice of itself (federation, codec, origin, data) the pipeline needs. Pull the shared task-test helpers (the schema factory, stock schemas, base federation options, and the recording MockQueue) into a new testing/mq-tasks.ts module, and split the enqueue-specific cases out of tasks.test.ts into enqueue.test.ts. Teach the fixture-usage check to expand glob patterns in its allowlist so the whole testing/ directory is covered by a single entry instead of one path per file. Assisted-by: Claude Code:claude-opus-4-8
Two branches both touched the task testing utilities and diverged: one split MockQueue and the shared schemas/options out into mq-tasks.ts, while the other kept evolving them in tasks.ts. After rebasing the common edits, consolidate everything back into a single tasks.ts and drop the now-redundant mq-tasks.ts. Assisted-by: Claude Code:claude-opus-4-8
The key-value deduplication path reserved a marker before dispatching to the queue but never undid it when the dispatch failed. A transient backend failure therefore left the marker behind, so the retry was silently deduplicated against a task that had never reached the queue. The cas claim now stores a unique token instead of a bare `true`, and a failed dispatch conditionally clears it (cas succeeds only while the stored value is still our token). The conditional clear keeps a stale rollback from deleting a marker that another concurrent enqueue has already re-claimed. A rollback that itself fails is logged and swallowed so the original enqueue error still reaches the caller. The enqueueMany requirement for deduplicated multi-item batches now keys on whether deduplication is actually applied—a native queue or the cas fallback—rather than on nativeDeduplication alone. Under the "open" fallback (no native dedup, no cas) no marker is taken, so the batch fans out without deduplication instead of throwing. ParallelMessageQueue likewise rejects a deduplicated batch when the wrapped queue lacks enqueueMany, since fanning out cannot carry one key atomically. fedify-dev#798 Assisted-by: Claude Code:claude-opus-4-8
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. fedify-dev#799 Assisted-by: Claude Code:claude-opus-4-8
1 participant
Resolves #799, the third and final sub-issue of #206 (custom background tasks). Once this lands, #206 is fully resolved.
Background
The core task API (#797/#803) shipped task dispatch behavior and structured logging, but the task worker carries no span and no metrics: of the message variants handled in
processQueuedTask, every other branch (fanout/outbox/inbox) is dispatched with instrumentation, buttask.This PR closes that gap by layering task-specific telemetry onto the decision points the core already established. It reuses the queue-task metric pattern introduced in #759 and mirrors the existing
http_signatures.failure_reasonenum in metrics.ts. It changes no drop/retry behavior: telemetry is observed, never enforced.What changes
Span
Each dequeued task now runs inside a
fedify.taskconsumer span. The name is namespaced underfedify.rather thanactivitypub.because tasks are not part of ActivityPub, paralleling the existingactivitypub.inbox/outbox/fanoutspans. The span:fedify.task.nameandfedify.task.attempt(the zero-based attempt number).fedify.task.failure_reasonand sets its status toERRORon a terminal failure, so trace backends surface failed tasks without re-deriving the reason from logs.Failure attribution
#listenTaskMessagenow returns the failure reason (orundefinedon success) so the span/metric wrapper can attribute it. To distinguish a deserialization failure from a validation failure, the former combinedcodec.decode(...)call is split into its existingdeserializethenvalidatephases. This is behavior-preserving—decodeis literallyvalidate(schema, await deserialize(raw))—andTaskCodecgains a thin instancevalidate()wrapper so the dispatch site can split the two phases without importing the class.The four bounded
fedify.task.failure_reasonvalues map one-to-one to the worker's dispatch decision points:deserialization— the wire payload could not be deserialized.validation— the deserialized payload failed schema validation.unknown_task— the task name has no registered handler.handler— the registered handler threw.A worker shutdown is the one exception: an interrupted attempt is reported as an
abortedoutcome with nofedify.task.failure_reason, never as ahandlerfailure.Metrics surface
Tasks reuse the
fedify.queue.task.*metric family under a newtaskrole:QueueTaskRolegains"task".QueueTaskCommonAttributesgainstaskName, emitted asfedify.task.name.QueueTaskFailureReasontype, mirroringHttpSignatureMetricFailureReason.recordQueueTaskOutcome()gains an optional trailingfailureReasonparameter (non-breaking); it is emitted asfedify.task.failure_reasononly on afailedresult.recordQueueTaskEnqueuedrecordsrole: "task"at both the enqueue site (after a genuine dispatch, never on a dedup skip or a failed enqueue) and the retry re-enqueue site.fedify.queue.backendreports the resolved queue—the one actually used after routing, which may be the outbox queue under the fallback mode—so the metric stays accurate regardless of routing.Cardinality
Bounded by construction: task names are a registered, known-at-startup set (never derived from message content), and
failure_reasonis a four-value bounded enum. Combined cardinality istaskName × |failure_reason| × queue.backend, within OTel attribute safety. The process-localin_flightUpDownCounter omitsfedify.task.nameso its series stays drained.Out of scope
taskName(would risk unbounded cardinality).QueueTaskFailureReasonset—explicitly open to later refinement as long as it stays a small bounded set.Tests
packages/fedify/src/federation/tasks/tasks.test.ts gains a telemetry block with one assertion per acceptance criterion, using
TestSpanExporter/createTestTracerProvider/createTestMeterProviderfrom@fedify/fixture. Coverage:fedify.taskspan exists withfedify.task.nameandfedify.task.attempt.fedify.task.failure_reason.fedify.queue.backendreflects the resolved queue, including the outbox fallback.recordQueueTaskEnqueued/recordQueueTaskOutcomecarryrole: "task".Verified across Deno, Node.js, and Bun.
Documentation
fedify.taskspan row, thetaskvalue added to thefedify.queue.roleenumeration, a widenedfailed-result definition covering acked task drops, and thefedify.task.name/fedify.task.attempt/fedify.task.failure_reasonattribute rows.AI disclosure
Assisted-by: Claude Code:claude-opus-4-8