From 193bbd5a5ea36d1ad44c8efa84243abb301a2c63 Mon Sep 17 00:00:00 2001 From: u-zzn Date: Thu, 1 Oct 2026 15:02:01 +0900 Subject: [PATCH 01/10] Align LogTape context with activity spans LogTape's withContext() was only refreshed at the two outer boundaries of federation processing: Federation.fetch()'s own HTTP request span, and the queue worker's consumer span opened by #runWorkerSpan()/processQueuedTask(). It was never refreshed when a deeper, operation-specific span was opened for an individual delivery (activitypub.send_activity) or inbox request (activitypub.inbox), so logs emitted inside those spans still carried the enclosing worker's or request's traceId/spanId instead of their own -- the same IDs a TraceActivityRecord derived from that span would carry. Scope the LogTape context to those two span-creation sites using the same withContext({traceId, spanId}, ...) pattern already used at the outer boundaries. AsyncLocalStorage-backed withContext() restores the outer context automatically once the inner callback settles, so no manual save/restore was needed. Add regression tests that exercise real sendActivity(), handleInbox(), and processQueuedTask() calls and compare the logs they actually emit against real OpenTelemetry spans and, where one exists, a real TraceActivityRecord extracted via FedifySpanExporter -- including restoration after success and after a thrown error, isolation between genuinely concurrent operations, and preservation of existing context properties such as requestId and messageId. A delivery that never succeeds has no persisted record at all, since FedifySpanExporter only derives one from the "sent" event that sendActivityInternal() adds on success; that failure case is verified against the delivery span itself instead, with the record's absence asserted explicitly. See https://github.com/fedify-dev/fedify/issues/1030 Assisted-by: Claude Code:claude-sonnet-5 --- .../align-log-context-with-activity-spans.md | 8 + .../fedify/src/federation/handler.test.ts | 316 ++++++++++++++++++ packages/fedify/src/federation/handler.ts | 13 +- .../fedify/src/federation/middleware.test.ts | 128 ++++++- packages/fedify/src/federation/send.test.ts | 309 +++++++++++++++++ packages/fedify/src/federation/send.ts | 13 +- 6 files changed, 782 insertions(+), 5 deletions(-) create mode 100644 changes.d/fedify/align-log-context-with-activity-spans.md diff --git a/changes.d/fedify/align-log-context-with-activity-spans.md b/changes.d/fedify/align-log-context-with-activity-spans.md new file mode 100644 index 000000000..46c6dc526 --- /dev/null +++ b/changes.d/fedify/align-log-context-with-activity-spans.md @@ -0,0 +1,8 @@ + - Fixed log entries emitted during an individual outbound activity delivery + or inbound activity processing carrying the enclosing HTTP request's or + queue worker's `traceId`/`spanId` instead of the delivery or processing + operation's own. Warning and error logs now match the `traceId`/`spanId` + of the `TraceActivityRecord` that operation produces (as exposed by + `@fedify/fedify/otel`), including when the operation runs inside a + background queue worker, which makes it possible to correlate a failure + log with the specific activity it belongs to. [[#1030] by u-zzn] diff --git a/packages/fedify/src/federation/handler.test.ts b/packages/fedify/src/federation/handler.test.ts index 4d33cac6f..106cd3547 100644 --- a/packages/fedify/src/federation/handler.test.ts +++ b/packages/fedify/src/federation/handler.test.ts @@ -21,8 +21,18 @@ import { assertEquals, assertGreaterOrEqual, assertInstanceOf, + assertNotEquals, assertRejects, } from "@std/assert"; +import { + configure, + getLogger, + type LogRecord, + reset, + withContext, +} from "@logtape/logtape"; +import { AsyncLocalStorage } from "node:async_hooks"; +import { FedifySpanExporter } from "../otel/exporter.ts"; import { parseAcceptSignature } from "../sig/accept.ts"; import { signRequest } from "../sig/http.ts"; import { generateCryptoKeyPair } from "../sig/key.ts"; @@ -4938,6 +4948,312 @@ test("handleInbox() records OpenTelemetry span events", async () => { ); }); +test("handleInbox() aligns the LogTape context with its own span", async (t) => { + async function buildSignedRequest(activityId: string, noteId: string) { + const activity = new Create({ + id: new URL(activityId), + actor: new URL("https://example.com/users/someone"), + object: new Note({ + id: new URL(noteId), + content: "Hello, world!", + }), + }); + const request = new Request("https://example.com/users/someone/inbox", { + method: "POST", + headers: { "Content-Type": "application/activity+json" }, + body: JSON.stringify(await activity.toJsonLd()), + }); + return await signRequest( + request, + rsaPrivateKey3, + new URL("https://example.com/users/someone#main-key"), + ); + } + + const actorDispatcher: ActorDispatcher = (ctx, identifier) => { + if (identifier !== "someone") return null; + return new Person({ + id: ctx.getActorUri(identifier), + name: "Someone", + inbox: new URL("https://example.com/users/someone/inbox"), + publicKey: rsaPublicKey2, + }); + }; + + async function callHandleInbox( + federation: ReturnType>, + kv: MemoryKvStore, + tracerProvider: ReturnType[0], + listeners: ActivityListenerSet>, + signed: Request, + ) { + const context = createRequestContext({ + federation, + request: signed, + url: new URL(signed.url), + data: undefined, + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + getActorUri(identifier: string) { + return new URL(`https://example.com/users/${identifier}`); + }, + }); + return await handleInbox(signed, { + recipient: "someone", + context, + inboxContextFactory(_activity) { + return createInboxContext({ ...context, clone: undefined }); + }, + kv, + kvPrefixes: { + activityIdempotence: ["activityIdempotence"], + publicKey: ["publicKey"], + acceptSignatureNonce: ["acceptSignatureNonce"], + }, + actorDispatcher, + inboxListeners: listeners, + inboxErrorHandler: undefined, + onNotFound: (_request) => new Response("Not found", { status: 404 }), + signatureTimeWindow: false, + skipSignatureVerification: true, + tracerProvider, + }); + } + + const records: LogRecord[] = []; + await reset(); + try { + await configure({ + sinks: { buffer: (record: LogRecord) => records.push(record) }, + filters: {}, + loggers: [ + { category: [], sinks: ["buffer"], lowestLevel: "debug" }, + { category: ["logtape", "meta"], sinks: [] }, + ], + contextLocalStorage: new AsyncLocalStorage(), + }); + + // A sentinel outer context standing in for whatever span/context an + // enclosing HTTP request (e.g. Federation.fetch()'s own withContext() at + // the "request.method" span) would already have established. Using an + // arbitrary, unmistakable value -- rather than just asserting + // "not equal to the inner span's ID" -- proves the outer context is + // restored to its *actual* prior value, not merely cleared. + const outerContext = { + requestId: "outer-request", + traceId: "outer0000000000000000000000sentinel", + spanId: "outer00000sentinel", + }; + + await t.step( + "a listener failure log matches the inbox's own TraceActivityRecord, " + + "and the outer context is restored afterward", + async () => { + const [tracerProvider, exporter] = createTestTracerProvider(); + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + const listeners = new ActivityListenerSet>(); + listeners.add(Create, () => { + throw new Error("listener boom"); + }); + + const signed = await buildSignedRequest( + "https://example.com/activity/listener-failure", + "https://example.com/note/listener-failure", + ); + + await withContext(outerContext, async () => { + const response = await callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signed, + ); + // The listener's error is caught and logged internally, so + // handleInbox() still resolves (with a 500) instead of rejecting. + assertEquals(response.status, 500); + getLogger(["fedify", "federation", "handler.test"]).info( + "after handled failure", + ); + }); + + const span = exporter.getSpan("activitypub.inbox"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); + + // The HTTP Signature verification event is recorded on this span + // *before* routeActivity() dispatches to the listener, so a + // TraceActivityRecord is persisted for it regardless of whether the + // listener later throws -- unlike a delivery that never succeeds. + // Feed the real captured span through the real exporter so the + // comparison is against the actual persisted record, not a + // hand-matched ID. + const fedifyExporter = new FedifySpanExporter(new MemoryKvStore()); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + const [record] = await fedifyExporter.getActivitiesByTraceId(traceId); + assert(record != null); + assertEquals(record.direction, "inbound"); + assertEquals(record.spanId, spanId); + + // Logged by routeActivity()'s own "activitypub.dispatch_inbox_listener" + // child span, which doesn't scope its own LogTape context -- it must + // inherit the "activitypub.inbox" span's context set by handleInbox(). + const failureLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + assert(failureLog != null); + assertEquals(failureLog.properties.spanId, record.spanId); + assertEquals(failureLog.properties.traceId, record.traceId); + assertEquals(failureLog.properties.requestId, "outer-request"); + + const afterLog = records.find((r) => + r.rawMessage === "after handled failure" + ); + assert(afterLog != null); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + + exporter.clear(); + records.length = 0; + }, + ); + + await t.step( + "outer context is restored after successful inbox processing", + async () => { + const [tracerProvider, exporter] = createTestTracerProvider(); + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + const listeners = new ActivityListenerSet>(); + let received: Activity | null = null; + listeners.add(Create, (_ctx, activity) => { + received = activity; + }); + + const signed = await buildSignedRequest( + "https://example.com/activity/listener-success", + "https://example.com/note/listener-success", + ); + + await withContext(outerContext, async () => { + const response = await callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signed, + ); + assertEquals(response.status, 202); + getLogger(["fedify", "federation", "handler.test"]).info( + "after success", + ); + }); + assert(received != null); + + const span = exporter.getSpan("activitypub.inbox"); + assert(span != null); + const { spanId } = span.spanContext(); + + const afterLog = records.find((r) => r.rawMessage === "after success"); + assert(afterLog != null); + assertNotEquals(afterLog.properties.spanId, spanId); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + + exporter.clear(); + records.length = 0; + }, + ); + + await t.step( + "concurrent inbox requests do not leak context into each other", + async () => { + const [tracerProvider, exporter] = createTestTracerProvider(); + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + + let arrived = 0; + let release: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const listeners = new ActivityListenerSet>(); + listeners.add(Create, async () => { + arrived++; + if (arrived >= 2) release(); + await gate; + throw new Error("listener boom"); + }); + + const [signedA, signedB] = await Promise.all([ + buildSignedRequest( + "https://example.com/activity/concurrent-a", + "https://example.com/note/concurrent-a", + ), + buildSignedRequest( + "https://example.com/activity/concurrent-b", + "https://example.com/note/concurrent-b", + ), + ]); + + const [responseA, responseB] = await Promise.all([ + callHandleInbox(federation, kv, tracerProvider, listeners, signedA), + callHandleInbox(federation, kv, tracerProvider, listeners, signedB), + ]); + assertEquals(responseA.status, 500); + assertEquals(responseB.status, 500); + + const spans = exporter.getSpans("activitypub.inbox"); + assertEquals(spans.length, 2); + const spanA = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/concurrent-a" + ); + const spanB = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/concurrent-b" + ); + assert(spanA != null && spanB != null); + assertNotEquals( + spanA.spanContext().spanId, + spanB.spanContext().spanId, + ); + + const logA = records.find((r) => + r.properties.activityId === + "https://example.com/activity/concurrent-a" && + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + const logB = records.find((r) => + r.properties.activityId === + "https://example.com/activity/concurrent-b" && + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + assert(logA != null && logB != null); + assertEquals(logA.properties.spanId, spanA.spanContext().spanId); + assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertNotEquals(logA.properties.spanId, logB.properties.spanId); + + exporter.clear(); + records.length = 0; + }, + ); + } finally { + await reset(); + } +}); + test("handleInbox() records fedify.queue.task.enqueued when queued", async () => { const [meterProvider, recorder] = createTestMeterProvider(); const kv = new MemoryKvStore(); diff --git a/packages/fedify/src/federation/handler.ts b/packages/fedify/src/federation/handler.ts index 14ecec4ee..cced9e4d8 100644 --- a/packages/fedify/src/federation/handler.ts +++ b/packages/fedify/src/federation/handler.ts @@ -32,7 +32,7 @@ import { readBoundedText, } from "../utils/body.ts"; import jsonld from "@fedify/vocab-runtime/jsonld"; -import { getLogger } from "@logtape/logtape"; +import { getLogger, withContext } from "@logtape/logtape"; import type { MeterProvider, Span, @@ -1825,8 +1825,17 @@ export async function handleInbox( if (options.recipient != null) { span.setAttribute("fedify.inbox.recipient", options.recipient); } + const spanContext = span.spanContext(); try { - return await handleInboxInternal(request, options, span); + // Scope the LogTape context to this inbox span so that logs emitted + // while it's active carry the same traceId/spanId as the + // TraceActivityRecord derived from this span, rather than the + // enclosing HTTP request's. AsyncLocalStorage-backed withContext() + // restores the outer context once this callback settles. + return await withContext( + { traceId: spanContext.traceId, spanId: spanContext.spanId }, + () => handleInboxInternal(request, options, span), + ); } catch (e) { options.observation!.hasException = true; span.setStatus({ code: SpanStatusCode.ERROR, message: String(e) }); diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index 4ff0de460..370c85394 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -15,7 +15,12 @@ import { Person, } from "@fedify/vocab"; import { FetchError, getDocumentLoader, UrlError } from "@fedify/vocab-runtime"; -import { configure, type LogRecord, reset } from "@logtape/logtape"; +import { + configure, + type LogRecord, + reset, + withContext, +} from "@logtape/logtape"; import { metrics, SpanStatusCode } from "@opentelemetry/api"; import { DataPointType, @@ -42,6 +47,7 @@ import { strictEqual, throws, } from "node:assert/strict"; +import { AsyncLocalStorage } from "node:async_hooks"; import dns from "node:dns/promises"; import createFixture from "../../../fixture/src/fixtures/example.com/create.json" with { type: "json", @@ -8427,6 +8433,126 @@ test("FederationImpl.processQueuedTask() permanent failure", async (t) => { fetchMock.hardReset(); }); +test( + "FederationImpl.processQueuedTask() aligns the LogTape context with " + + "the delivery span inside the outbox worker", + async () => { + await withLogtapeLock(async () => { + fetchMock.spyGlobal(); + fetchMock.post("https://example.com/inbox-queue-failing", { + status: 500, + body: "Internal Server Error", + }); + + const records: LogRecord[] = []; + await reset(); + try { + await configure({ + sinks: { buffer: (record: LogRecord) => records.push(record) }, + filters: {}, + loggers: [ + { category: [], sinks: ["buffer"], lowestLevel: "debug" }, + { category: ["logtape", "meta"], sinks: [] }, + ], + contextLocalStorage: new AsyncLocalStorage(), + }); + + const [tracerProvider, exporter] = createTestTracerProvider(); + const queuedMessages: Message[] = []; + const queue: MessageQueue = { + enqueue(message, _options) { + queuedMessages.push(message); + return Promise.resolve(); + }, + listen(_handler, _options) { + return Promise.resolve(); + }, + }; + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + queue, + // This test's delivery target is a mocked, unresolvable host. + allowPrivateAddress: true, + tracerProvider, + }); + + const message = { + type: "outbox", + id: crypto.randomUUID(), + baseUrl: "https://example.com", + keys: [], + activity: { + "@context": "https://www.w3.org/ns/activitystreams", + type: "Create", + id: "https://example.com/activity/queue-failing", + actor: "https://example.com/users/alice", + object: { type: "Note", content: "test" }, + }, + activityType: "https://www.w3.org/ns/activitystreams#Create", + inbox: "https://example.com/inbox-queue-failing", + sharedInbox: false, + started: new Date().toISOString(), + attempt: 0, + headers: {}, + traceContext: {}, + } satisfies OutboxMessage; + + // Simulates logging context an application might have already + // established (e.g. a batch ID) around the queue callback, on top + // of the messageId that processQueuedTask() itself sets. + await withContext({ batchId: "outer-batch" }, async () => { + await federation.processQueuedTask(undefined, message); + }); + + // Retried once, same as the existing "500 does not permanently + // fail" behavior -- this fix must not change that. + assertEquals(queuedMessages.length, 1); + + const deliverySpan = exporter.getSpan("activitypub.send_activity"); + assert(deliverySpan != null); + const deliveryCtx = deliverySpan.spanContext(); + + const workerSpan = exporter.getSpan("activitypub.outbox"); + assert(workerSpan != null); + const workerCtx = workerSpan.spanContext(); + + assertNotEquals(deliveryCtx.spanId, workerCtx.spanId); + + // Logged by sendActivityInternal() while the delivery span (a + // child of the outbox worker span) is active. + const deliveryLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to send activity {activityId} to {inbox} ({status}", + ) + ); + assert(deliveryLog != null); + assertEquals(deliveryLog.properties.spanId, deliveryCtx.spanId); + assertEquals(deliveryLog.properties.traceId, deliveryCtx.traceId); + assertEquals(deliveryLog.properties.messageId, message.id); + assertEquals(deliveryLog.properties.batchId, "outer-batch"); + + // Logged by the outbox worker's own retry handling, after + // sendActivity()'s span has already ended -- it must keep the + // worker's own span, not the delivery attempt's. + const retryLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to send activity {activityId} to {inbox} (attempt", + ) + ); + assert(retryLog != null); + assertEquals(retryLog.properties.spanId, workerCtx.spanId); + assertEquals(retryLog.properties.traceId, workerCtx.traceId); + assertNotEquals(retryLog.properties.spanId, deliveryCtx.spanId); + assertEquals(retryLog.properties.messageId, message.id); + assertEquals(retryLog.properties.batchId, "outer-batch"); + } finally { + await reset(); + fetchMock.hardReset(); + } + }); + }, +); + test("FederationImpl.processQueuedTask() circuit breaker", async (t) => { fetchMock.spyGlobal(); diff --git a/packages/fedify/src/federation/send.test.ts b/packages/fedify/src/federation/send.test.ts index 1889754e1..222a6dcc5 100644 --- a/packages/fedify/src/federation/send.test.ts +++ b/packages/fedify/src/federation/send.test.ts @@ -23,6 +23,13 @@ import { assertNotEquals, assertRejects, } from "@std/assert"; +import { + configure, + getLogger, + type LogRecord, + reset, + withContext, +} from "@logtape/logtape"; import { AggregationTemporality, InMemoryMetricExporter, @@ -30,7 +37,9 @@ import { PeriodicExportingMetricReader, } from "@opentelemetry/sdk-metrics"; import fetchMock from "fetch-mock"; +import { AsyncLocalStorage } from "node:async_hooks"; import dns from "node:dns/promises"; +import { FedifySpanExporter } from "../otel/exporter.ts"; import { verifyRequest } from "../sig/http.ts"; import { doesActorOwnKey } from "../sig/owner.ts"; import { @@ -39,6 +48,7 @@ import { rsaPrivateKey2, rsaPublicKey2, } from "../testing/keys.ts"; +import { MemoryKvStore } from "./kv.ts"; import { extractInboxes, sendActivity, SendActivityError } from "./send.ts"; @@ -564,6 +574,305 @@ test("sendActivity() records OpenTelemetry span events", async (t) => { }); }); +function createActivity(id: string) { + return { + "@context": "https://www.w3.org/ns/activitystreams", + type: "Create", + id, + actor: "https://example.com/person", + }; +} + +function deliveryParams( + activityId: string, + inbox: string, + tracerProvider: ReturnType[0], +) { + return { + activity: createActivity(activityId), + activityId, + activityType: "https://www.w3.org/ns/activitystreams#Create", + keys: [{ + keyId: new URL("https://example.com/person#key"), + privateKey: rsaPrivateKey2, + }], + inbox: new URL(inbox), + tracerProvider, + }; +} + +test("sendActivity() aligns the LogTape context with its own span", async (t) => { + const [tracerProvider, exporter] = createTestTracerProvider(); + fetchMock.spyGlobal(); + + const records: LogRecord[] = []; + await reset(); + try { + await configure({ + sinks: { buffer: (record: LogRecord) => records.push(record) }, + filters: {}, + loggers: [ + { category: [], sinks: ["buffer"], lowestLevel: "debug" }, + { category: ["logtape", "meta"], sinks: [] }, + ], + contextLocalStorage: new AsyncLocalStorage(), + }); + + // A sentinel outer context standing in for whatever span/context an + // enclosing queue worker (e.g. #runWorkerSpan's own withContext()) would + // already have established. Using an arbitrary, unmistakable value -- + // rather than just asserting "not equal to the inner span's ID" -- + // proves the outer context is restored to its *actual* prior value, not + // merely cleared. + const outerContext = { + requestId: "outer-request", + traceId: "outer0000000000000000000000sentinel", + spanId: "outer00000sentinel", + }; + + await t.step( + "a warning logged during a successful send matches the resulting " + + "TraceActivityRecord", + async () => { + fetchMock.post("https://example.com/inbox-warn", { status: 202 }); + + await withContext({ requestId: "outer-request-warn" }, async () => { + // A non-RSA-only key set makes sendActivityInternal() log a + // warning ("No supported key found to sign...") but still + // complete the delivery, which is exactly the "warning beside a + // successful activity record" case #902 wants to surface. + await sendActivity({ + ...deliveryParams( + "https://example.com/activity/warn", + "https://example.com/inbox-warn", + tracerProvider, + ), + keys: [{ + keyId: ed25519Multikey.id!, + privateKey: ed25519PrivateKey, + }], + }); + }); + + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); + + // Feed the real captured span through the real exporter so the + // comparison is against the actual TraceActivityRecord that would be + // persisted for a debugger to read, not a hand-matched ID. + const kv = new MemoryKvStore(); + const fedifyExporter = new FedifySpanExporter(kv); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + const [record] = await fedifyExporter.getActivitiesByTraceId(traceId); + assert(record != null); + assertEquals(record.direction, "outbound"); + assertEquals(record.spanId, spanId); + + const warnLog = records.find((r) => + String(r.rawMessage).startsWith("No supported key found to sign") + ); + assert(warnLog != null); + assertEquals(warnLog.properties.spanId, record.spanId); + assertEquals(warnLog.properties.traceId, record.traceId); + // Properties set by an enclosing withContext() must survive the + // inner one scoped to the delivery span. + assertEquals(warnLog.properties.requestId, "outer-request-warn"); + + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + }, + ); + + await t.step( + "a failed delivery's error log matches its own span, and the outer " + + "context is restored once the rejection propagates", + async () => { + fetchMock.post("https://example.com/inbox-failing", { + status: 500, + body: "Internal Server Error", + }); + + await withContext(outerContext, async () => { + await assertRejects( + () => + sendActivity( + deliveryParams( + "https://example.com/activity/failing", + "https://example.com/inbox-failing", + tracerProvider, + ), + ), + SendActivityError, + ); + // Logged right after the rejected promise settles, i.e. once + // sendActivity()'s own span has already ended. + getLogger(["fedify", "federation", "send.test"]).info( + "after failure", + ); + }); + + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); + + // A delivery that never succeeds never gets a persisted + // TraceActivityRecord (FedifySpanExporter only extracts one from the + // "activitypub.activity.sent" event, which is added on success only). + // The fix must still scope the failure log to this span's own ID + // -- the same ID a record derived from it would carry -- rather than + // the enclosing worker's. + const kv = new MemoryKvStore(); + const fedifyExporter = new FedifySpanExporter(kv); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + assertEquals(await fedifyExporter.getActivitiesByTraceId(traceId), []); + + const failureLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to send activity {activityId} to {inbox} ({status}", + ) + ); + assert(failureLog != null); + assertEquals(failureLog.properties.spanId, spanId); + assertEquals(failureLog.properties.traceId, traceId); + // Properties set by an enclosing withContext() must survive the + // inner one scoped to the delivery span. + assertEquals(failureLog.properties.requestId, outerContext.requestId); + + const afterLog = records.find((r) => r.rawMessage === "after failure"); + assert(afterLog != null); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + }, + ); + + await t.step( + "outer context is restored after a successful send", + async () => { + fetchMock.post("https://example.com/inbox-ok", { status: 202 }); + + await withContext(outerContext, async () => { + await sendActivity( + deliveryParams( + "https://example.com/activity/ok", + "https://example.com/inbox-ok", + tracerProvider, + ), + ); + getLogger(["fedify", "federation", "send.test"]).info( + "after success", + ); + }); + + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { spanId } = span.spanContext(); + + const afterLog = records.find((r) => r.rawMessage === "after success"); + assert(afterLog != null); + assertNotEquals(afterLog.properties.spanId, spanId); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + }, + ); + + await t.step( + "concurrent deliveries do not leak context into each other", + async () => { + let arrived = 0; + let release: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + // Each handler blocks until both requests have actually arrived, so + // the two sendActivity() calls are genuinely in flight at once + // instead of merely racing a timer. + const handler = () => async () => { + arrived++; + if (arrived >= 2) release(); + await gate; + return { status: 500, body: "Internal Server Error" }; + }; + fetchMock.post("https://example.com/inbox-concurrent-a", handler()); + fetchMock.post("https://example.com/inbox-concurrent-b", handler()); + + const [resultA, resultB] = await Promise.allSettled([ + sendActivity( + deliveryParams( + "https://example.com/activity/a", + "https://example.com/inbox-concurrent-a", + tracerProvider, + ), + ), + sendActivity( + deliveryParams( + "https://example.com/activity/b", + "https://example.com/inbox-concurrent-b", + tracerProvider, + ), + ), + ]); + assertEquals(resultA.status, "rejected"); + assertEquals(resultB.status, "rejected"); + + const spans = exporter.getSpans("activitypub.send_activity"); + assertEquals(spans.length, 2); + const spanA = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/a" + ); + const spanB = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/b" + ); + assert(spanA != null && spanB != null); + assertNotEquals( + spanA.spanContext().spanId, + spanB.spanContext().spanId, + ); + + const logA = records.find((r) => + r.properties.activityId === "https://example.com/activity/a" && + String(r.rawMessage).startsWith("Failed to send activity") + ); + const logB = records.find((r) => + r.properties.activityId === "https://example.com/activity/b" && + String(r.rawMessage).startsWith("Failed to send activity") + ); + assert(logA != null && logB != null); + assertEquals(logA.properties.spanId, spanA.spanContext().spanId); + assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertNotEquals(logA.properties.spanId, logB.properties.spanId); + + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + }, + ); + } finally { + await reset(); + fetchMock.hardReset(); + } +}); + test("sendActivity() records OpenTelemetry delivery metrics", async (t) => { const [meterProvider, recorder] = createTestMeterProvider(); fetchMock.spyGlobal(); diff --git a/packages/fedify/src/federation/send.ts b/packages/fedify/src/federation/send.ts index 656b50014..8c20e0f8a 100644 --- a/packages/fedify/src/federation/send.ts +++ b/packages/fedify/src/federation/send.ts @@ -1,6 +1,6 @@ import type { Recipient } from "@fedify/vocab"; import { FetchError, UrlError, validatePublicUrl } from "@fedify/vocab-runtime"; -import { getLogger } from "@logtape/logtape"; +import { getLogger, withContext } from "@logtape/logtape"; import { type Attributes, type MeterProvider, @@ -294,8 +294,17 @@ export function sendActivity( if (options.activityType != null) { span.setAttribute("activitypub.activity.type", options.activityType); } + const spanContext = span.spanContext(); try { - await sendActivityInternal({ ...options, tracerProvider }, span); + // Scope the LogTape context to this delivery attempt's own span so + // that logs emitted while it's active carry the same traceId/spanId + // as the TraceActivityRecord derived from this span, rather than the + // enclosing queue worker's. AsyncLocalStorage-backed withContext() + // restores the outer context once this callback settles. + await withContext( + { traceId: spanContext.traceId, spanId: spanContext.spanId }, + () => sendActivityInternal({ ...options, tracerProvider }, span), + ); } catch (e) { span.setStatus({ code: SpanStatusCode.ERROR, message: String(e) }); throw e; From 3394960398500b1248ad99966aaaada10c73a816 Mon Sep 17 00:00:00 2001 From: u-zzn Date: Fri, 2 Oct 2026 14:37:26 +0900 Subject: [PATCH 02/10] Record the pull request number in the changelog fragment https://github.com/fedify-dev/fedify/pull/1205 Assisted-by: Claude Code:claude-sonnet-5 --- CHANGES.md | 13 +++++++++++++ .../fedify/align-log-context-with-activity-spans.md | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 4d3a4f422..2b449a8db 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -8,6 +8,19 @@ Version 2.5.0 To be released. +### @fedify/fedify + + - Fixed log entries emitted during an individual outbound activity delivery + or inbound activity processing carrying the enclosing HTTP request's or + queue worker's `traceId`/`spanId` instead of the delivery or processing + operation's own. Warning and error logs now match the `traceId`/`spanId` + of the `TraceActivityRecord` that operation produces (as exposed by + `@fedify/fedify/otel`), including when the operation runs inside a + background queue worker, which makes it possible to correlate a failure + log with the specific activity it belongs to. [[#1030] by u-zzn\] + +[#1030]: https://github.com/fedify-dev/fedify/issues/1030 + ### @fedify/cli - Changed the `fedify nodeinfo` favicon selector to pick a usable bitmap diff --git a/changes.d/fedify/align-log-context-with-activity-spans.md b/changes.d/fedify/align-log-context-with-activity-spans.md index 46c6dc526..23c4b3bc1 100644 --- a/changes.d/fedify/align-log-context-with-activity-spans.md +++ b/changes.d/fedify/align-log-context-with-activity-spans.md @@ -5,4 +5,4 @@ of the `TraceActivityRecord` that operation produces (as exposed by `@fedify/fedify/otel`), including when the operation runs inside a background queue worker, which makes it possible to correlate a failure - log with the specific activity it belongs to. [[#1030] by u-zzn] + log with the specific activity it belongs to. [[#1030], [#1205] by u-zzn] From 552021558d71dd17d21d267f922a8fbe0fd85da8 Mon Sep 17 00:00:00 2001 From: u-zzn Date: Fri, 2 Oct 2026 14:58:17 +0900 Subject: [PATCH 03/10] Remove explanatory comments added for this change These comments restated what the surrounding code and test names already make clear, and the design rationale they carried (event timing for TraceActivityRecord creation, outer-context restoration semantics) is already covered in the pull request description. Assisted-by: Claude Code:claude-sonnet-5 --- .../fedify/src/federation/handler.test.ts | 18 ------------ packages/fedify/src/federation/handler.ts | 5 ---- .../fedify/src/federation/middleware.test.ts | 11 -------- packages/fedify/src/federation/send.test.ts | 28 ------------------- packages/fedify/src/federation/send.ts | 5 ---- 5 files changed, 67 deletions(-) diff --git a/packages/fedify/src/federation/handler.test.ts b/packages/fedify/src/federation/handler.test.ts index 106cd3547..06045e244 100644 --- a/packages/fedify/src/federation/handler.test.ts +++ b/packages/fedify/src/federation/handler.test.ts @@ -5033,12 +5033,6 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => contextLocalStorage: new AsyncLocalStorage(), }); - // A sentinel outer context standing in for whatever span/context an - // enclosing HTTP request (e.g. Federation.fetch()'s own withContext() at - // the "request.method" span) would already have established. Using an - // arbitrary, unmistakable value -- rather than just asserting - // "not equal to the inner span's ID" -- proves the outer context is - // restored to its *actual* prior value, not merely cleared. const outerContext = { requestId: "outer-request", traceId: "outer0000000000000000000000sentinel", @@ -5070,8 +5064,6 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => listeners, signed, ); - // The listener's error is caught and logged internally, so - // handleInbox() still resolves (with a 500) instead of rejecting. assertEquals(response.status, 500); getLogger(["fedify", "federation", "handler.test"]).info( "after handled failure", @@ -5082,13 +5074,6 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => assert(span != null); const { traceId, spanId } = span.spanContext(); - // The HTTP Signature verification event is recorded on this span - // *before* routeActivity() dispatches to the listener, so a - // TraceActivityRecord is persisted for it regardless of whether the - // listener later throws -- unlike a delivery that never succeeds. - // Feed the real captured span through the real exporter so the - // comparison is against the actual persisted record, not a - // hand-matched ID. const fedifyExporter = new FedifySpanExporter(new MemoryKvStore()); await new Promise((resolve) => { fedifyExporter.export([span], () => resolve()); @@ -5098,9 +5083,6 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => assertEquals(record.direction, "inbound"); assertEquals(record.spanId, spanId); - // Logged by routeActivity()'s own "activitypub.dispatch_inbox_listener" - // child span, which doesn't scope its own LogTape context -- it must - // inherit the "activitypub.inbox" span's context set by handleInbox(). const failureLog = records.find((r) => String(r.rawMessage).startsWith( "Failed to process the incoming activity", diff --git a/packages/fedify/src/federation/handler.ts b/packages/fedify/src/federation/handler.ts index cced9e4d8..a5e4c3b8b 100644 --- a/packages/fedify/src/federation/handler.ts +++ b/packages/fedify/src/federation/handler.ts @@ -1827,11 +1827,6 @@ export async function handleInbox( } const spanContext = span.spanContext(); try { - // Scope the LogTape context to this inbox span so that logs emitted - // while it's active carry the same traceId/spanId as the - // TraceActivityRecord derived from this span, rather than the - // enclosing HTTP request's. AsyncLocalStorage-backed withContext() - // restores the outer context once this callback settles. return await withContext( { traceId: spanContext.traceId, spanId: spanContext.spanId }, () => handleInboxInternal(request, options, span), diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index 370c85394..be8ec6777 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -8471,7 +8471,6 @@ test( const federation = new FederationImpl({ kv: new MemoryKvStore(), queue, - // This test's delivery target is a mocked, unresolvable host. allowPrivateAddress: true, tracerProvider, }); @@ -8497,15 +8496,10 @@ test( traceContext: {}, } satisfies OutboxMessage; - // Simulates logging context an application might have already - // established (e.g. a batch ID) around the queue callback, on top - // of the messageId that processQueuedTask() itself sets. await withContext({ batchId: "outer-batch" }, async () => { await federation.processQueuedTask(undefined, message); }); - // Retried once, same as the existing "500 does not permanently - // fail" behavior -- this fix must not change that. assertEquals(queuedMessages.length, 1); const deliverySpan = exporter.getSpan("activitypub.send_activity"); @@ -8518,8 +8512,6 @@ test( assertNotEquals(deliveryCtx.spanId, workerCtx.spanId); - // Logged by sendActivityInternal() while the delivery span (a - // child of the outbox worker span) is active. const deliveryLog = records.find((r) => String(r.rawMessage).startsWith( "Failed to send activity {activityId} to {inbox} ({status}", @@ -8531,9 +8523,6 @@ test( assertEquals(deliveryLog.properties.messageId, message.id); assertEquals(deliveryLog.properties.batchId, "outer-batch"); - // Logged by the outbox worker's own retry handling, after - // sendActivity()'s span has already ended -- it must keep the - // worker's own span, not the delivery attempt's. const retryLog = records.find((r) => String(r.rawMessage).startsWith( "Failed to send activity {activityId} to {inbox} (attempt", diff --git a/packages/fedify/src/federation/send.test.ts b/packages/fedify/src/federation/send.test.ts index 222a6dcc5..4ce70f272 100644 --- a/packages/fedify/src/federation/send.test.ts +++ b/packages/fedify/src/federation/send.test.ts @@ -618,12 +618,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => contextLocalStorage: new AsyncLocalStorage(), }); - // A sentinel outer context standing in for whatever span/context an - // enclosing queue worker (e.g. #runWorkerSpan's own withContext()) would - // already have established. Using an arbitrary, unmistakable value -- - // rather than just asserting "not equal to the inner span's ID" -- - // proves the outer context is restored to its *actual* prior value, not - // merely cleared. const outerContext = { requestId: "outer-request", traceId: "outer0000000000000000000000sentinel", @@ -637,10 +631,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => fetchMock.post("https://example.com/inbox-warn", { status: 202 }); await withContext({ requestId: "outer-request-warn" }, async () => { - // A non-RSA-only key set makes sendActivityInternal() log a - // warning ("No supported key found to sign...") but still - // complete the delivery, which is exactly the "warning beside a - // successful activity record" case #902 wants to surface. await sendActivity({ ...deliveryParams( "https://example.com/activity/warn", @@ -658,9 +648,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => assert(span != null); const { traceId, spanId } = span.spanContext(); - // Feed the real captured span through the real exporter so the - // comparison is against the actual TraceActivityRecord that would be - // persisted for a debugger to read, not a hand-matched ID. const kv = new MemoryKvStore(); const fedifyExporter = new FedifySpanExporter(kv); await new Promise((resolve) => { @@ -677,8 +664,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => assert(warnLog != null); assertEquals(warnLog.properties.spanId, record.spanId); assertEquals(warnLog.properties.traceId, record.traceId); - // Properties set by an enclosing withContext() must survive the - // inner one scoped to the delivery span. assertEquals(warnLog.properties.requestId, "outer-request-warn"); exporter.clear(); @@ -709,8 +694,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => ), SendActivityError, ); - // Logged right after the rejected promise settles, i.e. once - // sendActivity()'s own span has already ended. getLogger(["fedify", "federation", "send.test"]).info( "after failure", ); @@ -720,12 +703,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => assert(span != null); const { traceId, spanId } = span.spanContext(); - // A delivery that never succeeds never gets a persisted - // TraceActivityRecord (FedifySpanExporter only extracts one from the - // "activitypub.activity.sent" event, which is added on success only). - // The fix must still scope the failure log to this span's own ID - // -- the same ID a record derived from it would carry -- rather than - // the enclosing worker's. const kv = new MemoryKvStore(); const fedifyExporter = new FedifySpanExporter(kv); await new Promise((resolve) => { @@ -741,8 +718,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => assert(failureLog != null); assertEquals(failureLog.properties.spanId, spanId); assertEquals(failureLog.properties.traceId, traceId); - // Properties set by an enclosing withContext() must survive the - // inner one scoped to the delivery span. assertEquals(failureLog.properties.requestId, outerContext.requestId); const afterLog = records.find((r) => r.rawMessage === "after failure"); @@ -802,9 +777,6 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => const gate = new Promise((resolve) => { release = resolve; }); - // Each handler blocks until both requests have actually arrived, so - // the two sendActivity() calls are genuinely in flight at once - // instead of merely racing a timer. const handler = () => async () => { arrived++; if (arrived >= 2) release(); diff --git a/packages/fedify/src/federation/send.ts b/packages/fedify/src/federation/send.ts index 8c20e0f8a..ab89d327a 100644 --- a/packages/fedify/src/federation/send.ts +++ b/packages/fedify/src/federation/send.ts @@ -296,11 +296,6 @@ export function sendActivity( } const spanContext = span.spanContext(); try { - // Scope the LogTape context to this delivery attempt's own span so - // that logs emitted while it's active carry the same traceId/spanId - // as the TraceActivityRecord derived from this span, rather than the - // enclosing queue worker's. AsyncLocalStorage-backed withContext() - // restores the outer context once this callback settles. await withContext( { traceId: spanContext.traceId, spanId: spanContext.spanId }, () => sendActivityInternal({ ...options, tracerProvider }, span), From c008eadcd5bb4328e3ea35db9deb8aac26c359ce Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:25:33 +0900 Subject: [PATCH 04/10] Pin the #1205 changelog link to the pull request CHANGES.md's generic issue-link template pointed #1205 at /issues/1205, but #1205 is this change's pull request, not an issue. Add a links frontmatter entry to the change fragment so Sacho pins it to /pull/1205 instead, and resync the generated section. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171934222 Assisted-by: Claude Code:claude-sonnet-5 --- CHANGES.md | 3 ++- changes.d/fedify/align-log-context-with-activity-spans.md | 5 +++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 2b449a8db..d1b02bc83 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -17,9 +17,10 @@ To be released. of the `TraceActivityRecord` that operation produces (as exposed by `@fedify/fedify/otel`), including when the operation runs inside a background queue worker, which makes it possible to correlate a failure - log with the specific activity it belongs to. [[#1030] by u-zzn\] + log with the specific activity it belongs to. [[#1030], [#1205] by u-zzn\] [#1030]: https://github.com/fedify-dev/fedify/issues/1030 +[#1205]: https://github.com/fedify-dev/fedify/pull/1205 ### @fedify/cli diff --git a/changes.d/fedify/align-log-context-with-activity-spans.md b/changes.d/fedify/align-log-context-with-activity-spans.md index 23c4b3bc1..4217839dc 100644 --- a/changes.d/fedify/align-log-context-with-activity-spans.md +++ b/changes.d/fedify/align-log-context-with-activity-spans.md @@ -1,3 +1,8 @@ +--- +links: + '#1030': https://github.com/fedify-dev/fedify/issues/1030 + '#1205': https://github.com/fedify-dev/fedify/pull/1205 +--- - Fixed log entries emitted during an individual outbound activity delivery or inbound activity processing carrying the enclosing HTTP request's or queue worker's `traceId`/`spanId` instead of the delivery or processing From 4164ce1bf373e994f4b3bc4e39ef571a0c0759b5 Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:28:08 +0900 Subject: [PATCH 05/10] Run each alignment step's cleanup in a finally block Each step's exporter.clear()/records.length = 0/fetchMock.hardReset() sat after its assertions, so a setup, dispatch, or assertion failure would skip them. Since the span exporter and log buffer are shared across all four steps, a mid-step failure would leak stale spans and log records into the next step, which can turn one real failure into several misleading ones. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171973566 Assisted-by: Claude Code:claude-sonnet-5 --- packages/fedify/src/federation/send.test.ts | 372 +++++++++++--------- 1 file changed, 196 insertions(+), 176 deletions(-) diff --git a/packages/fedify/src/federation/send.test.ts b/packages/fedify/src/federation/send.test.ts index 4ce70f272..e36f11293 100644 --- a/packages/fedify/src/federation/send.test.ts +++ b/packages/fedify/src/federation/send.test.ts @@ -628,48 +628,52 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => "a warning logged during a successful send matches the resulting " + "TraceActivityRecord", async () => { - fetchMock.post("https://example.com/inbox-warn", { status: 202 }); - - await withContext({ requestId: "outer-request-warn" }, async () => { - await sendActivity({ - ...deliveryParams( - "https://example.com/activity/warn", - "https://example.com/inbox-warn", - tracerProvider, - ), - keys: [{ - keyId: ed25519Multikey.id!, - privateKey: ed25519PrivateKey, - }], + try { + fetchMock.post("https://example.com/inbox-warn", { status: 202 }); + + await withContext({ requestId: "outer-request-warn" }, async () => { + await sendActivity({ + ...deliveryParams( + "https://example.com/activity/warn", + "https://example.com/inbox-warn", + tracerProvider, + ), + keys: [{ + keyId: ed25519Multikey.id!, + privateKey: ed25519PrivateKey, + }], + }); }); - }); - const span = exporter.getSpan("activitypub.send_activity"); - assert(span != null); - const { traceId, spanId } = span.spanContext(); + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); - const kv = new MemoryKvStore(); - const fedifyExporter = new FedifySpanExporter(kv); - await new Promise((resolve) => { - fedifyExporter.export([span], () => resolve()); - }); - const [record] = await fedifyExporter.getActivitiesByTraceId(traceId); - assert(record != null); - assertEquals(record.direction, "outbound"); - assertEquals(record.spanId, spanId); - - const warnLog = records.find((r) => - String(r.rawMessage).startsWith("No supported key found to sign") - ); - assert(warnLog != null); - assertEquals(warnLog.properties.spanId, record.spanId); - assertEquals(warnLog.properties.traceId, record.traceId); - assertEquals(warnLog.properties.requestId, "outer-request-warn"); + const kv = new MemoryKvStore(); + const fedifyExporter = new FedifySpanExporter(kv); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + const [record] = await fedifyExporter.getActivitiesByTraceId( + traceId, + ); + assert(record != null); + assertEquals(record.direction, "outbound"); + assertEquals(record.spanId, spanId); - exporter.clear(); - records.length = 0; - fetchMock.hardReset(); - fetchMock.spyGlobal(); + const warnLog = records.find((r) => + String(r.rawMessage).startsWith("No supported key found to sign") + ); + assert(warnLog != null); + assertEquals(warnLog.properties.spanId, record.spanId); + assertEquals(warnLog.properties.traceId, record.traceId); + assertEquals(warnLog.properties.requestId, "outer-request-warn"); + } finally { + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + } }, ); @@ -677,166 +681,182 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => "a failed delivery's error log matches its own span, and the outer " + "context is restored once the rejection propagates", async () => { - fetchMock.post("https://example.com/inbox-failing", { - status: 500, - body: "Internal Server Error", - }); + try { + fetchMock.post("https://example.com/inbox-failing", { + status: 500, + body: "Internal Server Error", + }); - await withContext(outerContext, async () => { - await assertRejects( - () => - sendActivity( - deliveryParams( - "https://example.com/activity/failing", - "https://example.com/inbox-failing", - tracerProvider, + await withContext(outerContext, async () => { + await assertRejects( + () => + sendActivity( + deliveryParams( + "https://example.com/activity/failing", + "https://example.com/inbox-failing", + tracerProvider, + ), ), - ), - SendActivityError, - ); - getLogger(["fedify", "federation", "send.test"]).info( - "after failure", - ); - }); + SendActivityError, + ); + getLogger(["fedify", "federation", "send.test"]).info( + "after failure", + ); + }); - const span = exporter.getSpan("activitypub.send_activity"); - assert(span != null); - const { traceId, spanId } = span.spanContext(); + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); - const kv = new MemoryKvStore(); - const fedifyExporter = new FedifySpanExporter(kv); - await new Promise((resolve) => { - fedifyExporter.export([span], () => resolve()); - }); - assertEquals(await fedifyExporter.getActivitiesByTraceId(traceId), []); + const kv = new MemoryKvStore(); + const fedifyExporter = new FedifySpanExporter(kv); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + assertEquals( + await fedifyExporter.getActivitiesByTraceId(traceId), + [], + ); - const failureLog = records.find((r) => - String(r.rawMessage).startsWith( - "Failed to send activity {activityId} to {inbox} ({status}", - ) - ); - assert(failureLog != null); - assertEquals(failureLog.properties.spanId, spanId); - assertEquals(failureLog.properties.traceId, traceId); - assertEquals(failureLog.properties.requestId, outerContext.requestId); - - const afterLog = records.find((r) => r.rawMessage === "after failure"); - assert(afterLog != null); - assertEquals(afterLog.properties.requestId, outerContext.requestId); - assertEquals(afterLog.properties.traceId, outerContext.traceId); - assertEquals(afterLog.properties.spanId, outerContext.spanId); - - exporter.clear(); - records.length = 0; - fetchMock.hardReset(); - fetchMock.spyGlobal(); + const failureLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to send activity {activityId} to {inbox} ({status}", + ) + ); + assert(failureLog != null); + assertEquals(failureLog.properties.spanId, spanId); + assertEquals(failureLog.properties.traceId, traceId); + assertEquals( + failureLog.properties.requestId, + outerContext.requestId, + ); + + const afterLog = records.find((r) => + r.rawMessage === "after failure" + ); + assert(afterLog != null); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + } finally { + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + } }, ); await t.step( "outer context is restored after a successful send", async () => { - fetchMock.post("https://example.com/inbox-ok", { status: 202 }); - - await withContext(outerContext, async () => { - await sendActivity( - deliveryParams( - "https://example.com/activity/ok", - "https://example.com/inbox-ok", - tracerProvider, - ), - ); - getLogger(["fedify", "federation", "send.test"]).info( - "after success", - ); - }); - - const span = exporter.getSpan("activitypub.send_activity"); - assert(span != null); - const { spanId } = span.spanContext(); + try { + fetchMock.post("https://example.com/inbox-ok", { status: 202 }); + + await withContext(outerContext, async () => { + await sendActivity( + deliveryParams( + "https://example.com/activity/ok", + "https://example.com/inbox-ok", + tracerProvider, + ), + ); + getLogger(["fedify", "federation", "send.test"]).info( + "after success", + ); + }); - const afterLog = records.find((r) => r.rawMessage === "after success"); - assert(afterLog != null); - assertNotEquals(afterLog.properties.spanId, spanId); - assertEquals(afterLog.properties.requestId, outerContext.requestId); - assertEquals(afterLog.properties.traceId, outerContext.traceId); - assertEquals(afterLog.properties.spanId, outerContext.spanId); + const span = exporter.getSpan("activitypub.send_activity"); + assert(span != null); + const { spanId } = span.spanContext(); - exporter.clear(); - records.length = 0; - fetchMock.hardReset(); - fetchMock.spyGlobal(); + const afterLog = records.find((r) => + r.rawMessage === "after success" + ); + assert(afterLog != null); + assertNotEquals(afterLog.properties.spanId, spanId); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + } finally { + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + fetchMock.spyGlobal(); + } }, ); await t.step( "concurrent deliveries do not leak context into each other", async () => { - let arrived = 0; - let release: () => void; - const gate = new Promise((resolve) => { - release = resolve; - }); - const handler = () => async () => { - arrived++; - if (arrived >= 2) release(); - await gate; - return { status: 500, body: "Internal Server Error" }; - }; - fetchMock.post("https://example.com/inbox-concurrent-a", handler()); - fetchMock.post("https://example.com/inbox-concurrent-b", handler()); - - const [resultA, resultB] = await Promise.allSettled([ - sendActivity( - deliveryParams( - "https://example.com/activity/a", - "https://example.com/inbox-concurrent-a", - tracerProvider, + try { + let arrived = 0; + let release: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const handler = () => async () => { + arrived++; + if (arrived >= 2) release(); + await gate; + return { status: 500, body: "Internal Server Error" }; + }; + fetchMock.post("https://example.com/inbox-concurrent-a", handler()); + fetchMock.post("https://example.com/inbox-concurrent-b", handler()); + + const [resultA, resultB] = await Promise.allSettled([ + sendActivity( + deliveryParams( + "https://example.com/activity/a", + "https://example.com/inbox-concurrent-a", + tracerProvider, + ), ), - ), - sendActivity( - deliveryParams( - "https://example.com/activity/b", - "https://example.com/inbox-concurrent-b", - tracerProvider, + sendActivity( + deliveryParams( + "https://example.com/activity/b", + "https://example.com/inbox-concurrent-b", + tracerProvider, + ), ), - ), - ]); - assertEquals(resultA.status, "rejected"); - assertEquals(resultB.status, "rejected"); - - const spans = exporter.getSpans("activitypub.send_activity"); - assertEquals(spans.length, 2); - const spanA = spans.find((s) => - s.attributes["activitypub.activity.id"] === - "https://example.com/activity/a" - ); - const spanB = spans.find((s) => - s.attributes["activitypub.activity.id"] === - "https://example.com/activity/b" - ); - assert(spanA != null && spanB != null); - assertNotEquals( - spanA.spanContext().spanId, - spanB.spanContext().spanId, - ); - - const logA = records.find((r) => - r.properties.activityId === "https://example.com/activity/a" && - String(r.rawMessage).startsWith("Failed to send activity") - ); - const logB = records.find((r) => - r.properties.activityId === "https://example.com/activity/b" && - String(r.rawMessage).startsWith("Failed to send activity") - ); - assert(logA != null && logB != null); - assertEquals(logA.properties.spanId, spanA.spanContext().spanId); - assertEquals(logB.properties.spanId, spanB.spanContext().spanId); - assertNotEquals(logA.properties.spanId, logB.properties.spanId); + ]); + assertEquals(resultA.status, "rejected"); + assertEquals(resultB.status, "rejected"); + + const spans = exporter.getSpans("activitypub.send_activity"); + assertEquals(spans.length, 2); + const spanA = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/a" + ); + const spanB = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/b" + ); + assert(spanA != null && spanB != null); + assertNotEquals( + spanA.spanContext().spanId, + spanB.spanContext().spanId, + ); - exporter.clear(); - records.length = 0; - fetchMock.hardReset(); + const logA = records.find((r) => + r.properties.activityId === "https://example.com/activity/a" && + String(r.rawMessage).startsWith("Failed to send activity") + ); + const logB = records.find((r) => + r.properties.activityId === "https://example.com/activity/b" && + String(r.rawMessage).startsWith("Failed to send activity") + ); + assert(logA != null && logB != null); + assertEquals(logA.properties.spanId, spanA.spanContext().spanId); + assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertNotEquals(logA.properties.spanId, logB.properties.spanId); + } finally { + exporter.clear(); + records.length = 0; + fetchMock.hardReset(); + } }, ); } finally { From beadc1d68ad1f50b8f8dd0ce7b0985f747de94ba Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:29:20 +0900 Subject: [PATCH 06/10] Run each alignment step's cleanup in a finally block Same issue as the sendActivity() alignment steps: each step's own exporter.clear()/records.length = 0 sat after its assertions, so a setup, dispatch, or assertion failure would skip it and leak stale spans and log records into the next step via the shared exporter and log buffer. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171973566 Assisted-by: Claude Code:claude-sonnet-5 --- .../fedify/src/federation/handler.test.ts | 338 ++++++++++-------- 1 file changed, 180 insertions(+), 158 deletions(-) diff --git a/packages/fedify/src/federation/handler.test.ts b/packages/fedify/src/federation/handler.test.ts index 06045e244..267575ec8 100644 --- a/packages/fedify/src/federation/handler.test.ts +++ b/packages/fedify/src/federation/handler.test.ts @@ -5044,65 +5044,69 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => "and the outer context is restored afterward", async () => { const [tracerProvider, exporter] = createTestTracerProvider(); - const kv = new MemoryKvStore(); - const federation = createFederation({ kv, tracerProvider }); - const listeners = new ActivityListenerSet>(); - listeners.add(Create, () => { - throw new Error("listener boom"); - }); - - const signed = await buildSignedRequest( - "https://example.com/activity/listener-failure", - "https://example.com/note/listener-failure", - ); + try { + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + const listeners = new ActivityListenerSet>(); + listeners.add(Create, () => { + throw new Error("listener boom"); + }); - await withContext(outerContext, async () => { - const response = await callHandleInbox( - federation, - kv, - tracerProvider, - listeners, - signed, - ); - assertEquals(response.status, 500); - getLogger(["fedify", "federation", "handler.test"]).info( - "after handled failure", + const signed = await buildSignedRequest( + "https://example.com/activity/listener-failure", + "https://example.com/note/listener-failure", ); - }); - const span = exporter.getSpan("activitypub.inbox"); - assert(span != null); - const { traceId, spanId } = span.spanContext(); + await withContext(outerContext, async () => { + const response = await callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signed, + ); + assertEquals(response.status, 500); + getLogger(["fedify", "federation", "handler.test"]).info( + "after handled failure", + ); + }); - const fedifyExporter = new FedifySpanExporter(new MemoryKvStore()); - await new Promise((resolve) => { - fedifyExporter.export([span], () => resolve()); - }); - const [record] = await fedifyExporter.getActivitiesByTraceId(traceId); - assert(record != null); - assertEquals(record.direction, "inbound"); - assertEquals(record.spanId, spanId); - - const failureLog = records.find((r) => - String(r.rawMessage).startsWith( - "Failed to process the incoming activity", - ) - ); - assert(failureLog != null); - assertEquals(failureLog.properties.spanId, record.spanId); - assertEquals(failureLog.properties.traceId, record.traceId); - assertEquals(failureLog.properties.requestId, "outer-request"); + const span = exporter.getSpan("activitypub.inbox"); + assert(span != null); + const { traceId, spanId } = span.spanContext(); - const afterLog = records.find((r) => - r.rawMessage === "after handled failure" - ); - assert(afterLog != null); - assertEquals(afterLog.properties.requestId, outerContext.requestId); - assertEquals(afterLog.properties.traceId, outerContext.traceId); - assertEquals(afterLog.properties.spanId, outerContext.spanId); + const fedifyExporter = new FedifySpanExporter(new MemoryKvStore()); + await new Promise((resolve) => { + fedifyExporter.export([span], () => resolve()); + }); + const [record] = await fedifyExporter.getActivitiesByTraceId( + traceId, + ); + assert(record != null); + assertEquals(record.direction, "inbound"); + assertEquals(record.spanId, spanId); + + const failureLog = records.find((r) => + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + assert(failureLog != null); + assertEquals(failureLog.properties.spanId, record.spanId); + assertEquals(failureLog.properties.traceId, record.traceId); + assertEquals(failureLog.properties.requestId, "outer-request"); - exporter.clear(); - records.length = 0; + const afterLog = records.find((r) => + r.rawMessage === "after handled failure" + ); + assert(afterLog != null); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + } finally { + exporter.clear(); + records.length = 0; + } }, ); @@ -5110,47 +5114,51 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => "outer context is restored after successful inbox processing", async () => { const [tracerProvider, exporter] = createTestTracerProvider(); - const kv = new MemoryKvStore(); - const federation = createFederation({ kv, tracerProvider }); - const listeners = new ActivityListenerSet>(); - let received: Activity | null = null; - listeners.add(Create, (_ctx, activity) => { - received = activity; - }); - - const signed = await buildSignedRequest( - "https://example.com/activity/listener-success", - "https://example.com/note/listener-success", - ); + try { + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + const listeners = new ActivityListenerSet>(); + let received: Activity | null = null; + listeners.add(Create, (_ctx, activity) => { + received = activity; + }); - await withContext(outerContext, async () => { - const response = await callHandleInbox( - federation, - kv, - tracerProvider, - listeners, - signed, + const signed = await buildSignedRequest( + "https://example.com/activity/listener-success", + "https://example.com/note/listener-success", ); - assertEquals(response.status, 202); - getLogger(["fedify", "federation", "handler.test"]).info( - "after success", - ); - }); - assert(received != null); - const span = exporter.getSpan("activitypub.inbox"); - assert(span != null); - const { spanId } = span.spanContext(); + await withContext(outerContext, async () => { + const response = await callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signed, + ); + assertEquals(response.status, 202); + getLogger(["fedify", "federation", "handler.test"]).info( + "after success", + ); + }); + assert(received != null); - const afterLog = records.find((r) => r.rawMessage === "after success"); - assert(afterLog != null); - assertNotEquals(afterLog.properties.spanId, spanId); - assertEquals(afterLog.properties.requestId, outerContext.requestId); - assertEquals(afterLog.properties.traceId, outerContext.traceId); - assertEquals(afterLog.properties.spanId, outerContext.spanId); + const span = exporter.getSpan("activitypub.inbox"); + assert(span != null); + const { spanId } = span.spanContext(); - exporter.clear(); - records.length = 0; + const afterLog = records.find((r) => + r.rawMessage === "after success" + ); + assert(afterLog != null); + assertNotEquals(afterLog.properties.spanId, spanId); + assertEquals(afterLog.properties.requestId, outerContext.requestId); + assertEquals(afterLog.properties.traceId, outerContext.traceId); + assertEquals(afterLog.properties.spanId, outerContext.spanId); + } finally { + exporter.clear(); + records.length = 0; + } }, ); @@ -5158,77 +5166,91 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => "concurrent inbox requests do not leak context into each other", async () => { const [tracerProvider, exporter] = createTestTracerProvider(); - const kv = new MemoryKvStore(); - const federation = createFederation({ kv, tracerProvider }); - - let arrived = 0; - let release: () => void; - const gate = new Promise((resolve) => { - release = resolve; - }); - const listeners = new ActivityListenerSet>(); - listeners.add(Create, async () => { - arrived++; - if (arrived >= 2) release(); - await gate; - throw new Error("listener boom"); - }); - - const [signedA, signedB] = await Promise.all([ - buildSignedRequest( - "https://example.com/activity/concurrent-a", - "https://example.com/note/concurrent-a", - ), - buildSignedRequest( - "https://example.com/activity/concurrent-b", - "https://example.com/note/concurrent-b", - ), - ]); - - const [responseA, responseB] = await Promise.all([ - callHandleInbox(federation, kv, tracerProvider, listeners, signedA), - callHandleInbox(federation, kv, tracerProvider, listeners, signedB), - ]); - assertEquals(responseA.status, 500); - assertEquals(responseB.status, 500); - - const spans = exporter.getSpans("activitypub.inbox"); - assertEquals(spans.length, 2); - const spanA = spans.find((s) => - s.attributes["activitypub.activity.id"] === - "https://example.com/activity/concurrent-a" - ); - const spanB = spans.find((s) => - s.attributes["activitypub.activity.id"] === - "https://example.com/activity/concurrent-b" - ); - assert(spanA != null && spanB != null); - assertNotEquals( - spanA.spanContext().spanId, - spanB.spanContext().spanId, - ); + try { + const kv = new MemoryKvStore(); + const federation = createFederation({ kv, tracerProvider }); + + let arrived = 0; + let release: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const listeners = new ActivityListenerSet>(); + listeners.add(Create, async () => { + arrived++; + if (arrived >= 2) release(); + await gate; + throw new Error("listener boom"); + }); - const logA = records.find((r) => - r.properties.activityId === - "https://example.com/activity/concurrent-a" && - String(r.rawMessage).startsWith( - "Failed to process the incoming activity", - ) - ); - const logB = records.find((r) => - r.properties.activityId === - "https://example.com/activity/concurrent-b" && - String(r.rawMessage).startsWith( - "Failed to process the incoming activity", - ) - ); - assert(logA != null && logB != null); - assertEquals(logA.properties.spanId, spanA.spanContext().spanId); - assertEquals(logB.properties.spanId, spanB.spanContext().spanId); - assertNotEquals(logA.properties.spanId, logB.properties.spanId); + const [signedA, signedB] = await Promise.all([ + buildSignedRequest( + "https://example.com/activity/concurrent-a", + "https://example.com/note/concurrent-a", + ), + buildSignedRequest( + "https://example.com/activity/concurrent-b", + "https://example.com/note/concurrent-b", + ), + ]); + + const [responseA, responseB] = await Promise.all([ + callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signedA, + ), + callHandleInbox( + federation, + kv, + tracerProvider, + listeners, + signedB, + ), + ]); + assertEquals(responseA.status, 500); + assertEquals(responseB.status, 500); + + const spans = exporter.getSpans("activitypub.inbox"); + assertEquals(spans.length, 2); + const spanA = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/concurrent-a" + ); + const spanB = spans.find((s) => + s.attributes["activitypub.activity.id"] === + "https://example.com/activity/concurrent-b" + ); + assert(spanA != null && spanB != null); + assertNotEquals( + spanA.spanContext().spanId, + spanB.spanContext().spanId, + ); - exporter.clear(); - records.length = 0; + const logA = records.find((r) => + r.properties.activityId === + "https://example.com/activity/concurrent-a" && + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + const logB = records.find((r) => + r.properties.activityId === + "https://example.com/activity/concurrent-b" && + String(r.rawMessage).startsWith( + "Failed to process the incoming activity", + ) + ); + assert(logA != null && logB != null); + assertEquals(logA.properties.spanId, spanA.spanContext().spanId); + assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertNotEquals(logA.properties.spanId, logB.properties.spanId); + } finally { + exporter.clear(); + records.length = 0; + } }, ); } finally { From 952ae47c4e755a9818ba45b2515719e2d121e67f Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:29:55 +0900 Subject: [PATCH 07/10] Check traceId alongside spanId in the concurrency test The concurrent-deliveries step only compared each log's spanId against its span. Add the matching traceId check for both logs, without assuming the two deliveries' traceIds must differ from each other -- only that each log matches its own span. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171985376 Assisted-by: Claude Code:claude-sonnet-5 --- packages/fedify/src/federation/send.test.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/packages/fedify/src/federation/send.test.ts b/packages/fedify/src/federation/send.test.ts index e36f11293..0774d492c 100644 --- a/packages/fedify/src/federation/send.test.ts +++ b/packages/fedify/src/federation/send.test.ts @@ -851,6 +851,14 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => assert(logA != null && logB != null); assertEquals(logA.properties.spanId, spanA.spanContext().spanId); assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertEquals( + logA.properties.traceId, + spanA.spanContext().traceId, + ); + assertEquals( + logB.properties.traceId, + spanB.spanContext().traceId, + ); assertNotEquals(logA.properties.spanId, logB.properties.spanId); } finally { exporter.clear(); From 359fe27d5d37b36b137935c78edf03c20bde377f Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:30:29 +0900 Subject: [PATCH 08/10] Check traceId alongside spanId in the concurrency test Same gap as the sendActivity() concurrency test: only spanId was checked. Add the matching traceId check for both logs, without assuming the two inbox requests' traceIds must differ from each other. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171985376 Assisted-by: Claude Code:claude-sonnet-5 --- packages/fedify/src/federation/handler.test.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/packages/fedify/src/federation/handler.test.ts b/packages/fedify/src/federation/handler.test.ts index 267575ec8..bda979481 100644 --- a/packages/fedify/src/federation/handler.test.ts +++ b/packages/fedify/src/federation/handler.test.ts @@ -5246,6 +5246,14 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => assert(logA != null && logB != null); assertEquals(logA.properties.spanId, spanA.spanContext().spanId); assertEquals(logB.properties.spanId, spanB.spanContext().spanId); + assertEquals( + logA.properties.traceId, + spanA.spanContext().traceId, + ); + assertEquals( + logB.properties.traceId, + spanB.spanContext().traceId, + ); assertNotEquals(logA.properties.spanId, logB.properties.spanId); } finally { exporter.clear(); From 6461e811cb4df305fbf0c4268ee44bcee72424b5 Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:31:09 +0900 Subject: [PATCH 09/10] Guarantee the concurrency gate opens if one delivery fails early If one sendActivity() call failed before reaching the mocked handler (e.g. an earlier validation error), its counter increment never happened, so the other call's handler stayed blocked on `await gate` forever -- confirmed by forcing this path, which hung with Deno reporting a dangling pending promise. Attach .finally(release) to each dispatch so the gate opens once either settles, regardless of whether it ever reached the handler. allSettled already waited for both, so this closes the only remaining path to a stuck test. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171973566 Assisted-by: Claude Code:claude-sonnet-5 --- packages/fedify/src/federation/send.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/fedify/src/federation/send.test.ts b/packages/fedify/src/federation/send.test.ts index 0774d492c..0473c1ee0 100644 --- a/packages/fedify/src/federation/send.test.ts +++ b/packages/fedify/src/federation/send.test.ts @@ -792,7 +792,7 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => async () => { try { let arrived = 0; - let release: () => void; + let release: () => void = () => {}; const gate = new Promise((resolve) => { release = resolve; }); @@ -812,14 +812,14 @@ test("sendActivity() aligns the LogTape context with its own span", async (t) => "https://example.com/inbox-concurrent-a", tracerProvider, ), - ), + ).finally(release), sendActivity( deliveryParams( "https://example.com/activity/b", "https://example.com/inbox-concurrent-b", tracerProvider, ), - ), + ).finally(release), ]); assertEquals(resultA.status, "rejected"); assertEquals(resultB.status, "rejected"); From 853b828d8768bbb502f80d13274646195bee403f Mon Sep 17 00:00:00 2001 From: u-zzn Date: Mon, 5 Oct 2026 17:31:58 +0900 Subject: [PATCH 10/10] Guarantee the gate opens and both dispatches finish before cleanup Forced an early rejection on one dispatch to check this path: Promise.all rejected as soon as it did, which would let the step's finally clear the shared exporter/records while the other dispatch was potentially still mid-flight, and -- separately -- that other dispatch's listener could stay blocked on `await gate` forever if its own counter increment never happened. Switch to Promise.allSettled so cleanup only runs after both dispatches actually finish, and attach .finally(release) to each so the gate opens regardless of how either settles. These two changes have to land together: allSettled alone still leaves the gate closed if one side fails early, and .finally(release) alone doesn't stop Promise.all from moving on while the other call is still running. https://github.com/fedify-dev/fedify/pull/1205#discussion_r4171973566 Assisted-by: Claude Code:claude-sonnet-5 --- packages/fedify/src/federation/handler.test.ts | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/packages/fedify/src/federation/handler.test.ts b/packages/fedify/src/federation/handler.test.ts index bda979481..b15df577c 100644 --- a/packages/fedify/src/federation/handler.test.ts +++ b/packages/fedify/src/federation/handler.test.ts @@ -5171,7 +5171,7 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => const federation = createFederation({ kv, tracerProvider }); let arrived = 0; - let release: () => void; + let release: () => void = () => {}; const gate = new Promise((resolve) => { release = resolve; }); @@ -5194,24 +5194,26 @@ test("handleInbox() aligns the LogTape context with its own span", async (t) => ), ]); - const [responseA, responseB] = await Promise.all([ + const [resultA, resultB] = await Promise.allSettled([ callHandleInbox( federation, kv, tracerProvider, listeners, signedA, - ), + ).finally(release), callHandleInbox( federation, kv, tracerProvider, listeners, signedB, - ), + ).finally(release), ]); - assertEquals(responseA.status, 500); - assertEquals(responseB.status, 500); + assert(resultA.status === "fulfilled"); + assert(resultB.status === "fulfilled"); + assertEquals(resultA.value.status, 500); + assertEquals(resultB.value.status, 500); const spans = exporter.getSpans("activitypub.inbox"); assertEquals(spans.length, 2);