feat(webapp): tag Prisma spans with db.datasource attribute (#3422)

## Summary

Stamp every Prisma span with `db.datasource: "writer" | "replica"` so
traces can distinguish which client the query went through.

Both `PrismaClient` instances share the same global
`@prisma/instrumentation`, so their spans come out with identical names
and attributes today. This makes them trivially filterable.

## How

Two pieces in `apps/webapp/app/`:

1. **`v3/tracer.server.ts`** — a `DatasourceAttributeSpanProcessor`
reads an OTel context key in `onStart` and calls
`span.setAttribute("db.datasource", value)`. Registered as the first
span processor.
2. **`db.server.ts`** — `tagDatasource(datasource, client)` wraps each
`PrismaClient` with `$extends({ query: { $allOperations } })`. The
middleware sets the context key around the query and directly tags the
active span (to catch `prisma:client:operation`, which Prisma creates
before the middleware fires).

### Context-propagation gotcha

`PrismaPromise` is lazy — `query(args)` returns a thenable that only
starts when someone `.then()`s it. The naive `context.with(ctx, () =>
query(args))` restores ALS synchronously, so when Prisma's internal code
awaits the thenable later, the engine spans fire with the original ALS.
Wrapping as `async () => await query(args)` forces the `.then()` inside
the `context.with` callback, so ALS stays on our context for the engine
spans.

### Coverage

- **Tagged**: all `prisma:engine:*` (`connection`, `db_query`,
`serialize`, `query`, etc.), `prisma:client:operation`,
`prisma:client:serialize`, `prisma:client:connect`
- **Not tagged**: `prisma:client:load_engine` — one-time startup, fires
before any query

Concurrent `Promise.all([writer.x, replica.y])` correctly tags each pool
separately (ALS isolates per-Promise chain).

### Performance

One `context.with` (~200ns) and one `setAttribute` per span (effectively
free per OTel JS benchmarks) per Prisma op. Negligible against a query
path measured in milliseconds.

## Test plan

- [ ] Verify `db.datasource` appears on `prisma:engine:connection` spans
after the webapp is restarted
- [ ] Spot-check a handful of real traces carry the attribute

.server-changes/prisma-span-datasource-attribute.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Tag Prisma spans with `db.datasource: "writer" | "replica"` so monitors and trace queries can distinguish the writer pool from the replica pool. Applies to all `prisma:engine:*` spans (including `prisma:engine:connection` used by the connection-pool monitors) and the outer `prisma:client:operation` span.

apps/webapp/app/db.server.ts

@@ -13,8 +13,8 @@ import { env } from "./env.server";
 import { logger } from "./services/logger.server";
 import { isValidDatabaseUrl } from "./utils/db";
 import { singleton } from "./utils/singleton";
-import { startActiveSpan } from "./v3/tracer.server";
-import { Span } from "@opentelemetry/api";
+import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server";
+import { context, Span, trace } from "@opentelemetry/api";
 import { queryPerformanceMonitor } from "./utils/queryPerformanceMonitor.server";
 
 export type {
@@ -98,12 +98,30 @@ export async function $transaction<R>(
 
 export { Prisma };
 
-export const prisma = singleton("prisma", getClient);
+function tagDatasource<T extends PrismaClient>(
+  datasource: "writer" | "replica",
+  client: T
+): T {
+  return client.$extends({
+    name: "datasource-tagger",
+    query: {
+      $allOperations: ({ query, args }) => {
+        trace.getActiveSpan()?.setAttribute("db.datasource", datasource);
+        return context.with(
+          context.active().setValue(DATASOURCE_CONTEXT_KEY, datasource),
+          async () => await query(args)
+        );
+      },
+    },
+  }) as unknown as T;
+}
 
-export const $replica: PrismaReplicaClient = singleton(
-  "replica",
-  () => getReplicaClient() ?? prisma
-);
+export const prisma = singleton("prisma", () => tagDatasource("writer", getClient()));
+
+export const $replica: PrismaReplicaClient = singleton("replica", () => {
+  const replica = getReplicaClient();
+  return replica ? tagDatasource("replica", replica) : prisma;
+});
 
 function getClient() {
   const { DATABASE_URL } = process.env;

apps/webapp/app/v3/tracer.server.ts

@@ -1,6 +1,7 @@
 import {
   type Attributes,
   type Context,
+  createContextKey,
   DiagConsoleLogger,
   DiagLogLevel,
   type Link,
@@ -61,6 +62,24 @@ import { performance } from "node:perf_hooks";
 
 export const SEMINTATTRS_FORCE_RECORDING = "forceRecording";
 
+export const DATASOURCE_CONTEXT_KEY = createContextKey("trigger.db.datasource");
+
+class DatasourceAttributeSpanProcessor implements SpanProcessor {
+  onStart(span: Span, parentContext: Context): void {
+    const ds = parentContext.getValue(DATASOURCE_CONTEXT_KEY);
+    if (typeof ds === "string") {
+      span.setAttribute("db.datasource", ds);
+    }
+  }
+  onEnd(): void {}
+  shutdown(): Promise<void> {
+    return Promise.resolve();
+  }
+  forceFlush(): Promise<void> {
+    return Promise.resolve();
+  }
+}
+
 class CustomWebappSampler implements Sampler {
   constructor(private readonly _baseSampler: Sampler) {}
 
@@ -205,7 +224,7 @@ function setupTelemetry() {
 
   const samplingRate = 1.0 / Math.max(parseInt(env.INTERNAL_OTEL_TRACE_SAMPLING_RATE, 10), 1);
 
-  const spanProcessors: SpanProcessor[] = [];
+  const spanProcessors: SpanProcessor[] = [new DatasourceAttributeSpanProcessor()];
 
   if (env.INTERNAL_OTEL_TRACE_EXPORTER_URL) {
     const headers = parseInternalTraceHeaders() ?? {};

references/hello-world/src/trigger/example.ts

@@ -1,4 +1,4 @@
-import { batch, logger, task, tasks, timeout, wait } from "@trigger.dev/sdk";
+import { batch, logger, task, tasks, timeout, wait, waitUntil } from "@trigger.dev/sdk";
 import { setTimeout } from "timers/promises";
 import { ResourceMonitor } from "../resourceMonitor.js";
 import { fixedLengthTask } from "./batches.js";
@@ -21,6 +21,10 @@ export const helloWorldTask = task({
       env: process.env,
     });
 
+    waitUntil((async () => {
+      logger.info("Hello, world from the waitUntil hook", { payload });
+    })());
+
     logger.debug("debug: Hello, worlds!", { payload });
     logger.info("info: Hello, world!", { payload });
     logger.log("log: Hello, world!", { payload });