test(cloudflare): Add shared streamed-span helpers to the integration test runner

Adds `spanUtils.ts` next to `runner.ts`, and two span collection methods on
the runner. The suites ported to span streaming then do not each write their
own envelope handling.

`spanUtils.ts` reads a single envelope: `getSpanContainer` and
`getSpansFromEnvelope`. It re-exports `getSpanOp` from
`@sentry-internal/test-utils`, which this package already depends on. That
avoids a third copy of the function in the repo. The runner reads across
envelopes: `collectStreamedSpans` and `collectStreamedSpansUntilSegment`. The
names follow `dev-packages/test-utils/src/event-proxy-server.ts`, so a streamed
span assertion reads the same in this package and in the E2E apps.

A trace does not arrive in one envelope when several isolates send spans. The
collecting helpers therefore group the spans by trace, and resolve on the first
trace that satisfies the predicate.

Span waiters observe the envelope stream and never consume it. `.expect(...)`
stays usable for error envelopes at the same time. Span waiters also receive the
runner rejection, so a worker that fails to boot gives the real error instead of
a Vitest timeout.

`collectStreamedSpansUntilSegment` is for asserting on the segment span alone. Each
envelope is its own request to the mock server, so the segment can be received
before the envelope carrying its children even though it ends last. A suite that
asserts on the children waits for those children instead, by name or by count.

A runner now tears down its own workers from `onTestFinished`, and that teardown
kills only its own workers rather than running the shared cleanup. Both halves
are needed. A suite that asserts on streamed spans never calls `completed()`, so
its runner never settles and its `wrangler dev` would otherwise stay alive until
the Vitest process exits. A full run left 32 of them behind. Running the shared
cleanup instead would kill the worker the next test had already started. The
process-exit cleanup still covers every worker.

Renames `public-api/startSpan-streamed` to `public-api/startSpan`, and
`tracing/ignoreSpans-streamed` to `tracing/ignoreSpans`. Ports both suites onto
the helpers, which proves the shape before the bulk of the port. Span streaming
is the default, so a `-streamed` suffix no longer marks a difference, and the
explicit `traceLifecycle: 'stream'` those four suites carried says nothing
either. It goes with the suffix.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: JPeer264

dev-packages/cloudflare-integration-tests/runner.ts

@@ -1,11 +1,12 @@
-import type { Envelope, EnvelopeItemType } from '@sentry/core';
+import type { Envelope, EnvelopeItemType, SerializedStreamedSpan } from '@sentry/core';
 import { normalize } from '@sentry/core';
 import { createBasicSentryServer } from '@sentry-internal/test-utils';
 import { spawn, spawnSync } from 'child_process';
 import { existsSync, readdirSync, readFileSync } from 'fs';
 import { join } from 'path';
 import { inspect } from 'util';
-import { expect } from 'vitest';
+import { expect, onTestFinished } from 'vitest';
+import { getSpansFromEnvelope } from './spanUtils';
 
 const CLEANUP_STEPS = new Set<() => void>();
 
@@ -139,6 +140,9 @@ function deferredPromise<T = void>(
 
 type Expected = Envelope | ((envelope: Envelope) => void);
 
+/** Either the name of the segment span, or a predicate over it. */
+type SegmentMatcher = string | ((segmentSpan: SerializedStreamedSpan) => boolean);
+
 type StartResult = {
   completed(): Promise<void>;
   makeRequest<T>(
@@ -152,6 +156,25 @@ type StartResult = {
     expected: Expected | Expected[],
     options?: { headers?: Record<string, string>; data?: BodyInit; expectError?: boolean },
   ): Promise<T | undefined>;
+  /**
+   * Accumulates spans across envelopes, grouped by trace, and resolves with the spans of the first
+   * trace that satisfies `isDone`.
+   *
+   * A trace reaches the mock server in more than one envelope: the span buffer flushes on a timer, so
+   * a segment that is still open when its children flush arrives separately, and a Durable Object or a
+   * service binding sends its own spans from its own isolate. Anything asserting on a whole trace has
+   * to accumulate rather than read a single envelope.
+   */
+  collectStreamedSpans(isDone: (spansOfTrace: SerializedStreamedSpan[]) => boolean): Promise<SerializedStreamedSpan[]>;
+  /**
+   * Accumulates the spans of a trace until its segment span has arrived.
+   *
+   * Only use this to assert on the segment span itself. The segment span ends last, but each
+   * envelope is its own request to the mock server, so the segment can still be *received* before
+   * the envelope carrying its children. A suite that asserts on the children has to wait for those
+   * children by name or by count through `collectStreamedSpans`.
+   */
+  collectStreamedSpansUntilSegment(segment: SegmentMatcher): Promise<SerializedStreamedSpan[]>;
 };
 
 /** Creates a test runner */
@@ -211,17 +234,55 @@ export function createRunner(...paths: string[]) {
       return this;
     },
     start: function (signal?: AbortSignal): StartResult {
-      const { resolve, reject, promise: isComplete } = deferredPromise(cleanupChildProcesses);
+      let child: ReturnType<typeof spawn> | undefined;
+      let childSubWorker: ReturnType<typeof spawn> | undefined;
+
+      // Tears down this runner only. `cleanupChildProcesses` tears down every registered runner, so
+      // running it here would kill a worker another test has already started: a runner whose
+      // `isComplete` settles after its own test (a suite that asserts on streamed spans never calls
+      // `completed()`, so the abort signal settles it) would take the next test's worker with it.
+      // The mock server has to close here as well, otherwise one server per scenario stays listening
+      // for the whole run.
+      function cleanupThisRunner(): void {
+        child?.kill();
+        childSubWorker?.kill();
+        closeMockServer?.();
+        closeMockServer = undefined;
+      }
+
+      // A suite that asserts on streamed spans never calls `completed()`, so `isComplete` never
+      // settles and its worker would stay alive until the vitest process exits. With one such suite
+      // per file, a full run ends up with dozens of `wrangler dev` processes competing for the
+      // machine, and the later suites time out. Tie the teardown to the test instead.
+      onTestFinished(cleanupThisRunner);
+
+      let closeMockServer: (() => void) | undefined;
+
+      const { resolve, reject, promise: isComplete } = deferredPromise(cleanupThisRunner);
+
+      const spanWaiters: {
+        onSpans: (spans: SerializedStreamedSpan[]) => boolean;
+        resolve: () => void;
+        reject: (e: unknown) => void;
+      }[] = [];
+      let failure: unknown;
 
       // `reject` is called from background event handlers (child process `error`/`exit`, mock server
       // callbacks) that fire at arbitrary times relative to the test's `await` points. If `reject` runs
       // while nothing is awaiting `isComplete` yet (e.g. a child transiently exits while the test is
       // parked in `makeRequest`), the rejection has no handler attached and surfaces as an unhandled
       // promise rejection — which Vitest reports as a spurious "Unhandled error" that fails the whole
-      // suite. Attaching a no-op catch keeps the promise "handled"; the real rejection is still delivered
+      // suite. Attaching a catch keeps the promise "handled"; the real rejection is still delivered
       // to callers via `completed()`, so genuine failures still fail the test.
-      isComplete.catch(() => {
-        // handled in `completed()`
+      //
+      // A test that only asserts on streamed spans never calls `completed()`, so the same rejection is
+      // handed to the span waiters as well. Without it, a worker that fails to boot would surface as a
+      // Vitest timeout instead of the actual error.
+      isComplete.catch(e => {
+        failure = e;
+        for (const waiter of spanWaiters.splice(0)) {
+          waiter.reject(e);
+        }
       });
 
       const expectedEnvelopeCount = expectedEnvelopes.length;
@@ -237,8 +298,6 @@ export function createRunner(...paths: string[]) {
       workerPortPromise.catch(() => {
         // handled in `makeRequest`
       });
-      let child: ReturnType<typeof spawn> | undefined;
-      let childSubWorker: ReturnType<typeof spawn> | undefined;
 
       /** Called after each expect callback to check if we're complete */
       function expectCallbackCalled(): void {
@@ -254,6 +313,44 @@ export function createRunner(...paths: string[]) {
         });
       }
 
+      /** Resolves once `onSpans` returns true for the spans of an arriving span envelope. */
+      function waitForSpans(onSpans: (spans: SerializedStreamedSpan[]) => boolean): Promise<void> {
+        return new Promise((resolveWaiter, rejectWaiter) => {
+          if (failure) {
+            rejectWaiter(failure);
+            return;
+          }
+          spanWaiters.push({ onSpans, resolve: resolveWaiter, reject: rejectWaiter });
+        });
+      }
+
+      /**
+       * Span waiters observe the envelope stream, they never consume from it: a suite can assert on
+       * streamed spans and on error envelopes at the same time.
+       */
+      function notifySpanWaiters(envelope: Envelope): void {
+        const spans = getSpansFromEnvelope(envelope);
+        if (!spans.length) {
+          return;
+        }
+
+        for (const waiter of spanWaiters.slice()) {
+          let done: boolean;
+          try {
+            done = waiter.onSpans(spans);
+          } catch (e) {
+            spanWaiters.splice(spanWaiters.indexOf(waiter), 1);
+            waiter.reject(e);
+            continue;
+          }
+
+          if (done) {
+            spanWaiters.splice(spanWaiters.indexOf(waiter), 1);
+            waiter.resolve();
+          }
+        }
+      }
+
       function assertEnvelopeMatches(expected: Expected, envelope: Envelope): void {
         if (typeof expected === 'function') {
           expected(envelope);
@@ -265,6 +362,8 @@ export function createRunner(...paths: string[]) {
       function newEnvelope(envelope: Envelope): void {
         if (process.env.DEBUG) log('newEnvelope', inspect(envelope, false, null, true));
 
+        notifySpanWaiters(envelope);
+
         const envelopeItemType = envelope[1][0][0].type;
 
         if (ignored.has(envelopeItemType)) {
@@ -332,6 +431,7 @@ export function createRunner(...paths: string[]) {
       createBasicSentryServer(newEnvelope)
         .then(async ([mockServerPort, mockServerClose]) => {
           if (mockServerClose) {
+            closeMockServer = mockServerClose;
             CLEANUP_STEPS.add(() => {
               mockServerClose();
             });
@@ -501,6 +601,44 @@ export function createRunner(...paths: string[]) {
           await Promise.all(envelopePromises);
           return result;
         },
+        collectStreamedSpans: async function (
+          isDone: (spansOfTrace: SerializedStreamedSpan[]) => boolean,
+        ): Promise<SerializedStreamedSpan[]> {
+          const spansByTrace = new Map<string, SerializedStreamedSpan[]>();
+          let matched: SerializedStreamedSpan[] = [];
+
+          await waitForSpans(spans => {
+            for (const span of spans) {
+              const spansOfTrace = spansByTrace.get(span.trace_id);
+              if (spansOfTrace) {
+                spansOfTrace.push(span);
+              } else {
+                spansByTrace.set(span.trace_id, [span]);
+              }
+            }
+
+            // Every trace is a candidate, so a trace that never satisfies `isDone` cannot hold up the
+            // one that does. Insertion order means the earliest-arriving trace wins a tie.
+            for (const spansOfTrace of spansByTrace.values()) {
+              if (isDone(spansOfTrace)) {
+                matched = spansOfTrace;
+                return true;
+              }
+            }
+
+            return false;
+          });
+
+          return matched;
+        },
+        collectStreamedSpansUntilSegment: function (segment: SegmentMatcher): Promise<SerializedStreamedSpan[]> {
+          const matchesSegment =
+            typeof segment === 'string' ? (span: SerializedStreamedSpan) => span.name === segment : segment;
+
+          return this.collectStreamedSpans(spansOfTrace =>
+            spansOfTrace.some(span => span.is_segment && matchesSegment(span)),
+          );
+        },
       };
     },
   };

dev-packages/cloudflare-integration-tests/spanUtils.ts

@@ -0,0 +1,18 @@
+import type { Envelope, SerializedStreamedSpan, SerializedStreamedSpanContainer } from '@sentry/core';
+
+export { getSpanOp } from '@sentry-internal/test-utils';
+
+/**
+ * The span v2 container of an envelope, or `undefined` when the envelope carries no span item.
+ */
+export function getSpanContainer(envelope: Envelope): SerializedStreamedSpanContainer | undefined {
+  const spanItem = envelope[1].find(item => item[0].type === 'span');
+  return spanItem?.[1] as SerializedStreamedSpanContainer | undefined;
+}
+
+/**
+ * The spans of an envelope, or an empty array when the envelope carries no span item.
+ */
+export function getSpansFromEnvelope(envelope: Envelope): SerializedStreamedSpan[] {
+  return getSpanContainer(envelope)?.items ?? [];
+}