From 5af4e432e2547a4097452df5c23de96fc8743c64 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Wed, 14 Oct 2026 13:12:14 +0200 Subject: [PATCH 1/2] fix(telemetry): record an allowlist of HTTP headers on Maple's Workers effect 4.0.4 records no request or response headers by default. Maple's Workers now record a short allowlist on server and client spans: host and forwarding headers (OAuth redirect URIs are built from them), content type and length, Cloudflare ray/IP/country, and vendor request ids and rate-limit state. An allowlist rather than everything-but-redacted, because outgoing requests carry credentials no redaction list knows (a customer's ClickHouse password goes out as X-ClickHouse-Key, and was recorded under effect 4.0.1). The SDK no longer anonymizes a ResponseError that carries a cause or description: that is a failed response write, a real error of its own. --- apps/ai/src/chat/apply-proposal.ts | 4 +- apps/ai/src/chat/turn-runner.ts | 8 ++- apps/chat-bot/src/relay/run.ts | 3 +- .../src/shared/flushable-tracer.test.ts | 26 ++++++++ .../effect-sdk/src/shared/flushable-tracer.ts | 10 ++- .../src/cloudflare/worker-telemetry.test.ts | 54 +++++++++++++++ .../infra/src/cloudflare/worker-telemetry.ts | 65 +++++++++++++++++-- 7 files changed, 159 insertions(+), 11 deletions(-) create mode 100644 packages/infra/src/cloudflare/worker-telemetry.test.ts diff --git a/apps/ai/src/chat/apply-proposal.ts b/apps/ai/src/chat/apply-proposal.ts index 1654dac915..671932ca38 100644 --- a/apps/ai/src/chat/apply-proposal.ts +++ b/apps/ai/src/chat/apply-proposal.ts @@ -25,7 +25,7 @@ import type { TenantContext } from "@maple/backend/services/auth/tenant-context" import { OrgMembershipService } from "@maple/backend/services/auth/OrgMembershipService" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" import { Cause, Effect, Layer, ManagedRuntime, Option, Schema } from "effect" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { MUTATING_TOOL_NAMES } from "../mcp/tools/mutating" @@ -183,7 +183,7 @@ export const applyChatProposal = async (input: ApplyChatProposalInput): Promise< Layer.provideMerge(layerMapleDbConnection(input.env)), Layer.provideMerge(envPorts(input.env)), Layer.provideMerge(chatSessionsLayerIfBound(input.chatSessions, input.env)), - Layer.provideMerge(telemetry.layer), + Layer.provideMerge(Layer.merge(telemetry.layer, MapleHttpTracingLive)), ), ) diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 44a8032871..534c9676f5 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -25,7 +25,7 @@ import type { PrReviewFailureReason } from "@maple/domain/http" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" import { forkRequestScoped } from "@maple/backend/platform/fork-request-scoped" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" import { Cause, Effect, Exit, Layer, ManagedRuntime, Match, Option, Redacted } from "effect" import { FetchHttpClient } from "effect/http" import type { WorkersAiBinding } from "../platform/WorkersAiHttpClient" @@ -296,7 +296,11 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis Layer.provideMerge(layerMapleDbConnection(input.env)), Layer.provideMerge(envPorts(input.env)), Layer.provideMerge(chatSessionsLayerIfBound(input.chatSessions, input.env)), - Layer.provideMerge(ReturnedToolFailuresOkLive.pipe(Layer.provideMerge(telemetry.layer))), + Layer.provideMerge( + ReturnedToolFailuresOkLive.pipe( + Layer.provideMerge(Layer.merge(telemetry.layer, MapleHttpTracingLive)), + ), + ), ), ) diff --git a/apps/chat-bot/src/relay/run.ts b/apps/chat-bot/src/relay/run.ts index f2ade05419..80ab498631 100644 --- a/apps/chat-bot/src/relay/run.ts +++ b/apps/chat-bot/src/relay/run.ts @@ -12,7 +12,7 @@ import { connectors } from "@maple/chat-platform/connectors" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import type { ChatConnectorId, OrgId } from "@maple/domain/primitives" import type { IntegrationsPersistenceError } from "@maple/domain/http" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" import { ChatSessions, type ChatSessionsApi } from "@maple/backend/platform/bindings" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" @@ -309,6 +309,7 @@ const inRuntime = envPorts(env), chatSessionsLayerIfBound(chatSessions, env), telemetry.layer, + MapleHttpTracingLive, ), ), // On the fiber rather than in a `finally`: the flush is what exports this event's spans, diff --git a/packages/effect-sdk/src/shared/flushable-tracer.test.ts b/packages/effect-sdk/src/shared/flushable-tracer.test.ts index d575bd4110..20d7781720 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.test.ts @@ -286,6 +286,32 @@ describe("makeSpanBuffer rendered 5xx responses", () => { }), ) + it.effect("keeps a failed response write as its own error", () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + const error = new HttpServerError.HttpServerError({ + reason: new HttpServerError.ResponseError({ + request: HttpServerRequest.fromWeb(new Request("http://localhost/api/sync/shape")), + response: HttpServerResponse.empty({ status: 500 }), + description: "Error writing raw response", + cause: new Error("socket hang up"), + }), + }) + yield* Effect.annotateCurrentSpan({ "http.response.status_code": 500 }).pipe( + Effect.andThen(Effect.fail(error)), + Effect.withSpan("GET", { kind: "server" }), + Effect.provide(buffer.tracerLayer), + Effect.ignore, + ) + const [span] = buffer.drain() + assert.strictEqual(span?.status.code, 2 /* Error */) + const types = span!.events + .filter((event) => event.name === "exception") + .map((event) => event.attributes.find((attribute) => attribute.key === "exception.type")?.value) + assert.deepStrictEqual(types, [{ stringValue: "HttpServerError" }]) + }), + ) + it.effect("keeps an exception the handler recorded itself instead of relabelling it", () => Effect.gen(function* () { const buffer = makeSpanBuffer() diff --git a/packages/effect-sdk/src/shared/flushable-tracer.ts b/packages/effect-sdk/src/shared/flushable-tracer.ts index 371863d706..e3c377c5e3 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -259,7 +259,13 @@ const renderedServerError = (self: SpanImpl): number | undefined => { } // Effect 4.0.2+ fails a 5xx server span itself, with an `HttpServerError` whose reason is -// `ResponseError`. That names nothing the rendered 5xx did not, so it takes the same road. +// `ResponseError`. That names nothing the rendered 5xx did not, so it takes the same road. One +// carrying a cause or description is a failed response write, a real error of its own. +const isStatusOnlyResponseError = (error: unknown): boolean => + Predicate.isTagged(error, "ResponseError") && + (!Predicate.hasProperty(error, "cause") || error.cause === undefined) && + (!Predicate.hasProperty(error, "description") || error.description === undefined) + const isResponseFailure = (exit: Exit.Exit): boolean => { if (exit._tag === "Success") return true const [reason, ...rest] = exit.cause.reasons @@ -269,7 +275,7 @@ const isResponseFailure = (exit: Exit.Exit): boolean => { Cause.isFailReason(reason) && Predicate.isTagged(reason.error, "HttpServerError") && Predicate.hasProperty(reason.error, "reason") && - Predicate.isTagged(reason.error.reason, "ResponseError") + isStatusOnlyResponseError(reason.error.reason) ) } diff --git a/packages/infra/src/cloudflare/worker-telemetry.test.ts b/packages/infra/src/cloudflare/worker-telemetry.test.ts new file mode 100644 index 0000000000..6f6b4709de --- /dev/null +++ b/packages/infra/src/cloudflare/worker-telemetry.test.ts @@ -0,0 +1,54 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Layer, Tracer } from "effect" +import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/http" +import { MapleHttpTracingLive } from "./worker-telemetry.ts" + +describe("MapleHttpTracingLive", () => { + it.effect("records the allowlisted headers and never a credential", () => + Effect.gen(function* () { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + const client = yield* HttpClient.HttpClient + yield* client + .execute( + HttpClientRequest.post("https://ch.example.test/").pipe( + HttpClientRequest.setHeaders({ + "Content-Type": "text/plain", + "X-ClickHouse-Key": "secret", + Authorization: "Bearer secret", + }), + ), + ) + .pipe(Effect.withTracer(tracer)) + const attributes = spans[0]?.attributes + assert.strictEqual(attributes?.get("http.request.header.content-type"), "text/plain") + assert.strictEqual(attributes?.get("http.response.header.x-ratelimit-remaining"), "9") + assert.isFalse(attributes?.has("http.request.header.x-clickhouse-key")) + assert.isFalse(attributes?.has("http.request.header.authorization")) + assert.isFalse(attributes?.has("http.response.header.set-cookie")) + }).pipe( + Effect.provide( + Layer.mergeAll( + FetchHttpClient.layer.pipe( + Layer.provide( + Layer.succeed( + FetchHttpClient.Fetch, + async () => + new Response("ok", { + headers: { "x-ratelimit-remaining": "9", "set-cookie": "session=secret" }, + }), + ), + ), + ), + MapleHttpTracingLive, + ), + ), + ), + ) +}) diff --git a/packages/infra/src/cloudflare/worker-telemetry.ts b/packages/infra/src/cloudflare/worker-telemetry.ts index 36f6384939..0aa2540832 100644 --- a/packages/infra/src/cloudflare/worker-telemetry.ts +++ b/packages/infra/src/cloudflare/worker-telemetry.ts @@ -6,10 +6,60 @@ import { Telemetry, type TelemetrySdkOptions } from "@maple-dev/alchemy/telemetr import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { WorkerEnvironment } from "alchemy/Cloudflare" +import * as AlchemyTelemetry from "alchemy/Telemetry" +import { HttpClient, HttpMiddleware } from "effect/http" import * as Layer from "effect/Layer" export const MAPLE_REPOSITORY_URL = "https://github.com/MapleTechLabs/maple" +/** + * The headers worth recording on HTTP spans, by phase. effect 4.0.4 records none by default; an + * allowlist rather than "all but redacted" because outgoing requests carry credentials no + * redaction list knows about (a customer's ClickHouse password goes out as `x-clickhouse-key`). + */ +const RECORDED_HEADERS = { + // Which host and client a request came through: OAuth redirect URIs are built from these. + request: new Set([ + "host", + "x-forwarded-host", + "x-forwarded-proto", + "origin", + "content-type", + "content-length", + "cf-ray", + "cf-connecting-ip", + "cf-ipcountry", + "mcp-protocol-version", + ]), + // Vendor request ids and rate-limit state, what a 429 or a support ticket needs. + response: new Set([ + "content-type", + "content-length", + "cf-ray", + "cf-placement", + "retry-after", + "x-request-id", + "x-slack-req-id", + "ratelimit", + "ratelimit-policy", + "ratelimit-limit", + "ratelimit-remaining", + "ratelimit-reset", + "x-ratelimit-limit", + "x-ratelimit-remaining", + "x-ratelimit-reset", + ]), +} as const + +const isRecordedHeader = (name: string, phase: "request" | "response"): boolean => + RECORDED_HEADERS[phase].has(name.toLowerCase()) + +/** Header capture for every HTTP span a Maple Worker records, server and client. */ +export const MapleHttpTracingLive = Layer.mergeAll( + Layer.succeed(HttpMiddleware.TracerHeaderFilter, isRecordedHeader), + Layer.succeed(HttpClient.TracerHeaderFilter, isRecordedHeader), +) + export interface WorkerTelemetryOptions { readonly serviceName: string /** Added to the domain's `ANTICIPATED_ERROR_IDENTIFIERS` (e.g. the MCP set). */ @@ -31,7 +81,11 @@ export const workerTelemetryConfig = (options: WorkerTelemetryOptions): Telemetr }) export const WorkerTelemetry = (options: WorkerTelemetryOptions): Layer.Layer => - Telemetry(workerTelemetryConfig(options)) + Layer.mergeAll( + Telemetry(workerTelemetryConfig(options)), + // Read by the bridge's HTTP tracer, so it is built into every event with the exporters. + AlchemyTelemetry.layer(MapleHttpTracingLive), + ) /** * The SDK under another service name, for background work (queue, cron) whose spans must not @@ -40,7 +94,10 @@ export const WorkerTelemetry = (options: WorkerTelemetryOptions): Layer.Layer => - MapleCloudflareSDK.make(workerTelemetryConfig(options)).requestLayer.pipe( - // Same key as alchemy's tag; a type bridge only. - Layer.provide(Layer.effect(MapleCloudflareSDK.WorkerEnvironment, WorkerEnvironment)), + Layer.mergeAll( + MapleCloudflareSDK.make(workerTelemetryConfig(options)).requestLayer.pipe( + // Same key as alchemy's tag; a type bridge only. + Layer.provide(Layer.effect(MapleCloudflareSDK.WorkerEnvironment, WorkerEnvironment)), + ), + MapleHttpTracingLive, ) From 4c8b40e17c4c589a51a740945b0e1265fbae49f4 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Wed, 14 Oct 2026 13:30:10 +0200 Subject: [PATCH 2/2] fix(telemetry): apply the effect v4 review of the header allowlist - Drop cf-connecting-ip: an end user's address on every span is personal data, and cf-ipcountry is enough. - Record ClickHouse's query id, exception code and summary, vendor request ids (request-id, x-github-request-id), www-authenticate, cf-cache-status and the request's accept header. - runtimeTelemetry builds the SDK with the allowlist already in its layer, so a Durable Object runtime cannot forget it. - Effect lowercases header names before the filter, so no toLowerCase. - Tests: the server span through alchemy's bridge records an allowlisted header and no credential; each kind of failed response write keeps its own error. --- apps/ai/src/chat/apply-proposal.ts | 15 ++--- apps/ai/src/chat/turn-runner.ts | 21 +++---- apps/api/src/http/api-observability.ts | 9 +-- apps/chat-bot/src/relay/run.ts | 14 ++--- apps/electric-sync/src/worker-bridge.test.ts | 34 +++++++++-- .../src/shared/flushable-tracer.test.ts | 59 +++++++++++-------- .../infra/src/cloudflare/worker-telemetry.ts | 28 +++++++-- 7 files changed, 109 insertions(+), 71 deletions(-) diff --git a/apps/ai/src/chat/apply-proposal.ts b/apps/ai/src/chat/apply-proposal.ts index 671932ca38..bd0ecde965 100644 --- a/apps/ai/src/chat/apply-proposal.ts +++ b/apps/ai/src/chat/apply-proposal.ts @@ -11,7 +11,6 @@ * connector half of the approval policy; a connector approver's chat account rides along as * forensic context on the audit entry. */ -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { connectorApprovalTenant, decidedBy, @@ -25,19 +24,17 @@ import type { TenantContext } from "@maple/backend/services/auth/tenant-context" import { OrgMembershipService } from "@maple/backend/services/auth/OrgMembershipService" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" -import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { Cause, Effect, Layer, ManagedRuntime, Option, Schema } from "effect" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { MUTATING_TOOL_NAMES } from "../mcp/tools/mutating" import { toTenantContext, withConnectorActor } from "./turn-actor" /** Its own instance, at module scope, so an approval exports its spans from the isolate it ran in. */ -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat", - anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat", + anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, +}) export type ApplyChatProposalInput = ChatProposalApproval & { readonly env: Record @@ -183,7 +180,7 @@ export const applyChatProposal = async (input: ApplyChatProposalInput): Promise< Layer.provideMerge(layerMapleDbConnection(input.env)), Layer.provideMerge(envPorts(input.env)), Layer.provideMerge(chatSessionsLayerIfBound(input.chatSessions, input.env)), - Layer.provideMerge(Layer.merge(telemetry.layer, MapleHttpTracingLive)), + Layer.provideMerge(telemetry.layer), ), ) diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 534c9676f5..8d3f139605 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -18,14 +18,13 @@ * `env: workerEnv ?? {}` fallback silently degrading the model config when a caller forgot to * thread the worker env through. */ -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { ChatMessage, type ChatTurnOrigin, type ChatTurnTenantEncoded } from "@maple/domain/chat-session" import type { PrReviewFailureReason } from "@maple/domain/http" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" import { forkRequestScoped } from "@maple/backend/platform/fork-request-scoped" -import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { Cause, Effect, Exit, Layer, ManagedRuntime, Match, Option, Redacted } from "effect" import { FetchHttpClient } from "effect/http" import type { WorkersAiBinding } from "../platform/WorkersAiHttpClient" @@ -91,13 +90,11 @@ const ENGINE_BOOKKEEPING_SPANS = [ // Deliberately not `maple-api`: background work sharing the request-facing // service's name skewed its percentiles (p99 32s, 2026-09-04). -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat", - anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, - dropSpanNames: ENGINE_BOOKKEEPING_SPANS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat", + anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, + dropSpanNames: ENGINE_BOOKKEEPING_SPANS, +}) export interface RunChatSessionTurnInput { /** The Durable Object itself. Appends are direct calls, not stub RPC. */ @@ -296,11 +293,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis Layer.provideMerge(layerMapleDbConnection(input.env)), Layer.provideMerge(envPorts(input.env)), Layer.provideMerge(chatSessionsLayerIfBound(input.chatSessions, input.env)), - Layer.provideMerge( - ReturnedToolFailuresOkLive.pipe( - Layer.provideMerge(Layer.merge(telemetry.layer, MapleHttpTracingLive)), - ), - ), + Layer.provideMerge(ReturnedToolFailuresOkLive.pipe(Layer.provideMerge(telemetry.layer))), ), ) diff --git a/apps/api/src/http/api-observability.ts b/apps/api/src/http/api-observability.ts index 8384d3f3b7..e34deb7ad7 100644 --- a/apps/api/src/http/api-observability.ts +++ b/apps/api/src/http/api-observability.ts @@ -35,10 +35,11 @@ export const ApiObservabilityLive = Layer.mergeAll( EMAIL_UNSUBSCRIBE_PATH.test(request.url) || /\.(png|ico|jpg|jpeg|gif|css|js|svg|webp|woff2?)(\?.*)?$/i.test(request.url), ), - // Every request header lands on the server span as `http.request.header.`. - // Effect's defaults cover the usual credential headers; the provider webhook - // signatures are ours to add (a GitHub webhook HMAC is replayable alongside its - // body, and there is no reason to retain it). + // Spans record only the allowlisted headers (`MapleHttpTracingLive`), none of + // these; the list still redacts them wherever a `Headers` value is rendered or + // logged. Effect's defaults cover the usual credential headers; the provider + // webhook signatures are ours to add (a GitHub webhook HMAC is replayable + // alongside its body, and there is no reason to retain it). Layer.succeed(Headers.CurrentRedactedNames, [ "authorization", "cookie", diff --git a/apps/chat-bot/src/relay/run.ts b/apps/chat-bot/src/relay/run.ts index 80ab498631..eedc58ae3f 100644 --- a/apps/chat-bot/src/relay/run.ts +++ b/apps/chat-bot/src/relay/run.ts @@ -9,10 +9,9 @@ */ import type { ChatConnector, ConnectorCredentials, InboundEvent } from "@maple/chat-platform" import { connectors } from "@maple/chat-platform/connectors" -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import type { ChatConnectorId, OrgId } from "@maple/domain/primitives" import type { IntegrationsPersistenceError } from "@maple/domain/http" -import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { ChatSessions, type ChatSessionsApi } from "@maple/backend/platform/bindings" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" @@ -48,12 +47,10 @@ import { CHAT_ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/chat-platform/anticip * for, so the spans a turn produces are exported from here or not at all — and flushed when the * event is done, because nothing else closes a scope around it. */ -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat-bot", - anticipatedErrorIdentifiers: CHAT_ANTICIPATED_ERROR_IDENTIFIERS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat-bot", + anticipatedErrorIdentifiers: CHAT_ANTICIPATED_ERROR_IDENTIFIERS, +}) const APP_BASE_URL_FALLBACK = "https://app.maple.dev" @@ -309,7 +306,6 @@ const inRuntime = envPorts(env), chatSessionsLayerIfBound(chatSessions, env), telemetry.layer, - MapleHttpTracingLive, ), ), // On the fiber rather than in a `finally`: the flush is what exports this event's spans, diff --git a/apps/electric-sync/src/worker-bridge.test.ts b/apps/electric-sync/src/worker-bridge.test.ts index e693221e6d..85ad410cb4 100644 --- a/apps/electric-sync/src/worker-bridge.test.ts +++ b/apps/electric-sync/src/worker-bridge.test.ts @@ -1,6 +1,6 @@ import { assert, describe, it } from "@effect/vitest" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" import * as Cloudflare from "alchemy/Cloudflare" import { Effect, Exit, Layer, Schema, Scope } from "effect" import { HttpRouter } from "effect/http" @@ -73,7 +73,11 @@ const appLayer = (options: { readonly ensureConfigured?: Effect.Effect = appLayer()) => +const event = ( + path: string, + app: ReturnType = appLayer(), + headers: Record = {}, +) => Effect.gen(function* () { const recorded: Array = [] const realFetch = globalThis.fetch @@ -87,17 +91,21 @@ const event = (path: string, app: ReturnType = appLayer()) => const isolate = yield* Scope.make() const handler = yield* HttpRouter.toHttpEffect(app).pipe(Scope.provide(isolate)) const request = yield* Scope.make() + // What `WorkerTelemetry` registers with alchemy: the SDK and the header allowlist. const services = yield* Layer.buildWithScope( - MapleCloudflareSDK.make( - workerTelemetryConfig({ serviceName: "electric-sync" }), - ).requestLayer.pipe(Layer.provide(Layer.succeed(MapleCloudflareSDK.WorkerEnvironment, env))), + Layer.merge( + MapleCloudflareSDK.make( + workerTelemetryConfig({ serviceName: "electric-sync" }), + ).requestLayer.pipe(Layer.provide(Layer.succeed(MapleCloudflareSDK.WorkerEnvironment, env))), + MapleHttpTracingLive, + ), request, ) const fetchEvent: Effect.Effect | undefined = Cloudflare.Workers.makeRequestHandler(handler)({ kind: "Cloudflare.Workers.WorkerEvent", type: "fetch", - input: new Request(`http://sync.maple.test${path}`), + input: new Request(`http://sync.maple.test${path}`, { headers }), }) assert.isDefined(fetchEvent) const response = yield* fetchEvent!.pipe(Effect.provide(services), Scope.provide(request)) @@ -147,4 +155,18 @@ describe("electric-sync through alchemy's Worker bridge", () => { assert.strictEqual(statusCodeOf(server[0]), 500) }), ) + + it.effect("records the allowlisted request headers on the server span and no credential", () => + Effect.gen(function* () { + const { server } = yield* event("/nope", appLayer(), { + "cf-ray": "8f1a2b3c4d5e6f70-IAD", + authorization: "Bearer secret", + cookie: "session=secret", + }) + const keys = (server[0]?.attributes ?? []).map((attribute) => attribute.key) + assert.include(keys, "http.request.header.cf-ray") + assert.notInclude(keys, "http.request.header.authorization") + assert.notInclude(keys, "http.request.header.cookie") + }), + ) }) diff --git a/packages/effect-sdk/src/shared/flushable-tracer.test.ts b/packages/effect-sdk/src/shared/flushable-tracer.test.ts index 20d7781720..3385b039d4 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.test.ts @@ -286,31 +286,40 @@ describe("makeSpanBuffer rendered 5xx responses", () => { }), ) - it.effect("keeps a failed response write as its own error", () => - Effect.gen(function* () { - const buffer = makeSpanBuffer() - const error = new HttpServerError.HttpServerError({ - reason: new HttpServerError.ResponseError({ - request: HttpServerRequest.fromWeb(new Request("http://localhost/api/sync/shape")), - response: HttpServerResponse.empty({ status: 500 }), - description: "Error writing raw response", - cause: new Error("socket hang up"), - }), - }) - yield* Effect.annotateCurrentSpan({ "http.response.status_code": 500 }).pipe( - Effect.andThen(Effect.fail(error)), - Effect.withSpan("GET", { kind: "server" }), - Effect.provide(buffer.tracerLayer), - Effect.ignore, - ) - const [span] = buffer.drain() - assert.strictEqual(span?.status.code, 2 /* Error */) - const types = span!.events - .filter((event) => event.name === "exception") - .map((event) => event.attributes.find((attribute) => attribute.key === "exception.type")?.value) - assert.deepStrictEqual(types, [{ stringValue: "HttpServerError" }]) - }), - ) + // `@effect/platform-node` builds these when writing the response fails. + const writeFailures = [ + ["a cause", { cause: new Error("socket hang up") }], + ["a description", { description: "Error writing raw response" }], + ["both", { description: "Error writing raw response", cause: new Error("socket hang up") }], + ] as const + + for (const [label, detail] of writeFailures) { + it.effect(`keeps a failed response write carrying ${label} as its own error`, () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + const error = new HttpServerError.HttpServerError({ + reason: new HttpServerError.ResponseError({ + request: HttpServerRequest.fromWeb(new Request("http://localhost/api/sync/shape")), + response: HttpServerResponse.empty({ status: 500 }), + ...detail, + }), + }) + yield* Effect.annotateCurrentSpan({ "http.response.status_code": 500 }).pipe( + Effect.andThen(Effect.fail(error)), + Effect.withSpan("GET", { kind: "server" }), + Effect.provide(buffer.tracerLayer), + Effect.ignore, + ) + const [span] = buffer.drain() + assert.isDefined(span) + assert.strictEqual(span!.status.code, 2 /* Error */) + const types = span!.events + .filter((event) => event.name === "exception") + .map((event) => event.attributes.find((attribute) => attribute.key === "exception.type")?.value) + assert.deepStrictEqual(types, [{ stringValue: "HttpServerError" }]) + }), + ) + } it.effect("keeps an exception the handler recorded itself instead of relabelling it", () => Effect.gen(function* () { diff --git a/packages/infra/src/cloudflare/worker-telemetry.ts b/packages/infra/src/cloudflare/worker-telemetry.ts index 0aa2540832..bb3a1ad856 100644 --- a/packages/infra/src/cloudflare/worker-telemetry.ts +++ b/packages/infra/src/cloudflare/worker-telemetry.ts @@ -18,28 +18,37 @@ export const MAPLE_REPOSITORY_URL = "https://github.com/MapleTechLabs/maple" * redaction list knows about (a customer's ClickHouse password goes out as `x-clickhouse-key`). */ const RECORDED_HEADERS = { - // Which host and client a request came through: OAuth redirect URIs are built from these. + // Which host and client a request came through: OAuth redirect URIs are built from these. No + // client IP: country is enough, and an address on every span is personal data. request: new Set([ "host", "x-forwarded-host", "x-forwarded-proto", "origin", + "accept", "content-type", "content-length", "cf-ray", - "cf-connecting-ip", "cf-ipcountry", "mcp-protocol-version", ]), - // Vendor request ids and rate-limit state, what a 429 or a support ticket needs. + // Vendor request ids, rate-limit state and auth challenges: what a 429, a 401 or a support + // ticket needs. ClickHouse's query id is the key into the customer's `system.query_log`. response: new Set([ "content-type", "content-length", "cf-ray", "cf-placement", + "cf-cache-status", + "www-authenticate", "retry-after", + "request-id", "x-request-id", + "x-github-request-id", "x-slack-req-id", + "x-clickhouse-query-id", + "x-clickhouse-exception-code", + "x-clickhouse-summary", "ratelimit", "ratelimit-policy", "ratelimit-limit", @@ -51,8 +60,9 @@ const RECORDED_HEADERS = { ]), } as const +// Effect lowercases header names before the filter sees them. const isRecordedHeader = (name: string, phase: "request" | "response"): boolean => - RECORDED_HEADERS[phase].has(name.toLowerCase()) + RECORDED_HEADERS[phase].has(name) /** Header capture for every HTTP span a Maple Worker records, server and client. */ export const MapleHttpTracingLive = Layer.mergeAll( @@ -101,3 +111,13 @@ export const eventTelemetry = ( ), MapleHttpTracingLive, ) + +/** + * The SDK for a runtime the Worker bridge does not wrap (a Durable Object's own + * `ManagedRuntime`), its `layer` already carrying {@link MapleHttpTracingLive}. Call once at + * module scope: buffers are per isolate. + */ +export const runtimeTelemetry = (options: WorkerTelemetryOptions): MapleCloudflareSDK.Telemetry => { + const telemetry = MapleCloudflareSDK.make(workerTelemetryConfig(options)) + return { ...telemetry, layer: Layer.merge(telemetry.layer, MapleHttpTracingLive) } +}