extract span buffer to class, add serverSpanStreamingIntegration, use it in cloudflare, vercel edge and deno

Author: Lms24

packages/cloudflare/src/sdk.ts

@@ -8,6 +8,7 @@ import {
   initAndBind,
   linkedErrorsIntegration,
   requestDataIntegration,
+  serverSpanStreamingIntegration,
   stackParserFromStackParserOptions,
 } from '@sentry/core';
 import type { CloudflareClientOptions, CloudflareOptions } from './client';
@@ -36,6 +37,7 @@ export function getDefaultIntegrations(options: CloudflareOptions): Integration[
     // TODO(v11): the `include` object should be defined directly in the integration based on `sendDefaultPii`
     requestDataIntegration(sendDefaultPii ? undefined : { include: { cookies: false } }),
     consoleIntegration(),
+    ...(options.traceLifecycle === 'stream' ? [serverSpanStreamingIntegration()] : []),
   ];
 }
 

packages/cloudflare/test/sdk.test.ts

@@ -1,7 +1,7 @@
 import * as SentryCore from '@sentry/core';
-import { beforeEach, describe, expect, test, vi } from 'vitest';
+import { beforeEach, describe, expect, it, test, vi } from 'vitest';
 import { CloudflareClient } from '../src/client';
-import { init } from '../src/sdk';
+import { getDefaultIntegrations, init } from '../src/sdk';
 import { resetSdk } from './testUtils';
 
 describe('init', () => {
@@ -18,4 +18,65 @@ describe('init', () => {
     expect(client).toBeDefined();
     expect(client).toBeInstanceOf(CloudflareClient);
   });
+
+  describe('getDefaultIntegrations', () => {
+    it('returns list of integrations with default options', () => {
+      const integrations = getDefaultIntegrations({}).map(integration => integration.name);
+      expect(integrations).toEqual([
+        'Dedupe',
+        'InboundFilters',
+        'FunctionToString',
+        'LinkedErrors',
+        'Fetch',
+        'Hono',
+        'RequestData',
+        'Console',
+      ]);
+    });
+
+    it('adds dedupeIntegration if enableDedupe is true', () => {
+      const integrations = getDefaultIntegrations({ enableDedupe: true }).map(integration => integration.name);
+      expect(integrations).toEqual([
+        'Dedupe',
+        'InboundFilters',
+        'FunctionToString',
+        'LinkedErrors',
+        'Fetch',
+        'Hono',
+        'RequestData',
+        'Console',
+      ]);
+    });
+
+    it('adds serverSpanStreamingIntegration if traceLifecycle is stream', () => {
+      const integrations = getDefaultIntegrations({ traceLifecycle: 'stream' }).map(integration => integration.name);
+      expect(integrations).toEqual([
+        'Dedupe',
+        'InboundFilters',
+        'FunctionToString',
+        'LinkedErrors',
+        'Fetch',
+        'Hono',
+        'RequestData',
+        'Console',
+        'ServerSpanStreaming',
+      ]);
+    });
+
+    it('intializes requestDataIntegration to not include cookies if sendDefaultPii is false', () => {
+      const reqDataIntegrationSpy = vi.spyOn(SentryCore, 'requestDataIntegration');
+
+      getDefaultIntegrations({ sendDefaultPii: false }).map(integration => integration.name);
+
+      expect(reqDataIntegrationSpy).toHaveBeenCalledWith({ include: { cookies: false } });
+    });
+
+    it('intializes requestDataIntegration to include cookies if sendDefaultPii is true', () => {
+      const reqDataIntegrationSpy = vi.spyOn(SentryCore, 'requestDataIntegration');
+
+      getDefaultIntegrations({ sendDefaultPii: true }).map(integration => integration.name);
+
+      expect(reqDataIntegrationSpy).toHaveBeenCalledWith(undefined);
+    });
+  });
 });

packages/core/src/client.ts

@@ -1536,7 +1536,7 @@ function _validateBeforeSendResult(
 /**
  * Process the matching `beforeSendXXX` callback.
  */
-// eslint-disable-next-line complexity
+
 function processBeforeSend(
   client: Client,
   options: ClientOptions,

packages/core/src/index.ts

@@ -92,6 +92,7 @@ export {
 } from './utils/spanUtils';
 export { captureSpan } from './spans/captureSpan';
 export { safeSetSpanJSONAttributes } from './spans/spanFirstUtils';
+export { SpanBuffer, type SpanBufferOptions } from './spans/spanBuffer';
 export { _setSpanForScope as _INTERNAL_setSpanForScope } from './utils/spanOnScope';
 export { parseSampleRate } from './utils/parseSampleRate';
 export { applySdkMetadata } from './utils/sdkMetadata';
@@ -125,6 +126,7 @@ export { thirdPartyErrorFilterIntegration } from './integrations/third-party-err
 export { consoleIntegration } from './integrations/console';
 export { featureFlagsIntegration, type FeatureFlagsIntegration } from './integrations/featureFlags';
 export { growthbookIntegration } from './integrations/featureFlags';
+export { serverSpanStreamingIntegration } from './integrations/serverSpanStreaming';
 
 export { profiler } from './profiling';
 // eslint thinks the entire function is deprecated (while only one overload is actually deprecated)

packages/core/src/integrations/serverSpanStreaming.ts

@@ -0,0 +1,54 @@
+import { DEBUG_BUILD } from '../debug-build';
+import { defineIntegration } from '../integration';
+import { captureSpan } from '../spans/captureSpan';
+import { SpanBuffer } from '../spans/spanBuffer';
+import type { IntegrationFn } from '../types-hoist/integration';
+import { isV2BeforeSendSpanCallback } from '../utils/beforeSendSpan';
+import { debug } from '../utils/debug-logger';
+
+export interface ServerSpanStreamingOptions {
+  /** Max spans per envelope batch (default: 1000) */
+  maxSpanLimit?: number;
+  /** Flush interval in ms (default: 5000) */
+  flushInterval?: number;
+}
+
+const INTEGRATION_NAME = 'ServerSpanStreaming';
+
+const _serverSpanStreamingIntegration = ((options?: ServerSpanStreamingOptions) => {
+  return {
+    name: INTEGRATION_NAME,
+    setup(client) {
+      const clientOptions = client.getOptions();
+      const beforeSendSpan = clientOptions.beforeSendSpan;
+
+      const initialMessage = 'serverSpanStreamingIntegration requires';
+      const fallbackMsg = 'Falling back to static trace lifecycle.';
+
+      if (clientOptions.traceLifecycle !== 'stream') {
+        client.getOptions().traceLifecycle = 'static';
+        DEBUG_BUILD && debug.warn(`${initialMessage} \`traceLifecycle\` to be set to "stream"! ${fallbackMsg}`);
+        return;
+      }
+
+      if (beforeSendSpan && !isV2BeforeSendSpanCallback(beforeSendSpan)) {
+        client.getOptions().traceLifecycle = 'static';
+        DEBUG_BUILD &&
+          debug.warn(`${initialMessage} a beforeSendSpan callback using \`withStreamSpan\`! ${fallbackMsg}`);
+        return;
+      }
+
+      const buffer = new SpanBuffer(client, options);
+
+      client.on('enqueueSpan', spanJSON => {
+        buffer.addSpan(spanJSON);
+      });
+
+      client.on('afterSpanEnd', span => {
+        captureSpan(span, client);
+      });
+    },
+  };
+}) satisfies IntegrationFn;
+
+export const serverSpanStreamingIntegration = defineIntegration(_serverSpanStreamingIntegration);

packages/core/src/spans/spanBuffer.ts

@@ -0,0 +1,126 @@
+import type { Client } from '../client';
+import { DEBUG_BUILD } from '../debug-build';
+import { createSpanV2Envelope } from '../envelope';
+import { getDynamicSamplingContextFromSpan } from '../tracing/dynamicSamplingContext';
+import type { SpanV2JSON, SpanV2JSONWithSegmentRef } from '../types-hoist/span';
+import { debug } from '../utils/debug-logger';
+
+export interface SpanBufferOptions {
+  /** Max spans per trace before auto-flush (default: 1000) */
+  maxSpanLimit?: number;
+  /** Flush interval in ms (default: 5000) */
+  flushInterval?: number;
+}
+
+/**
+ * A buffer for span JSON objects that flushes them to Sentry in Span v2 envelopes.
+ * Handles interval-based flushing, size thresholds, and graceful shutdown.
+ */
+export class SpanBuffer {
+  private _spanTreeMap: Map<string, Set<SpanV2JSONWithSegmentRef>>;
+  private _flushIntervalId: ReturnType<typeof setInterval> | null;
+  private _client: Client;
+  private _maxSpanLimit: number;
+  private _flushInterval: number;
+
+  public constructor(client: Client, options?: SpanBufferOptions) {
+    this._spanTreeMap = new Map();
+    this._client = client;
+
+    const { maxSpanLimit, flushInterval } = options ?? {};
+
+    this._maxSpanLimit = maxSpanLimit && maxSpanLimit > 0 && maxSpanLimit <= 1000 ? maxSpanLimit : 1000;
+    this._flushInterval = flushInterval && flushInterval > 0 ? flushInterval : 5_000;
+
+    this._flushIntervalId = setInterval(() => {
+      this.flush();
+    }, this._flushInterval);
+
+    this._client.on('flush', () => {
+      this.flush();
+    });
+  }
+
+  /**
+   * Add a span to the buffer.
+   */
+  public addSpan(spanJSON: SpanV2JSONWithSegmentRef): void {
+    const traceId = spanJSON.trace_id;
+    let traceBucket = this._spanTreeMap.get(traceId);
+    if (traceBucket) {
+      traceBucket.add(spanJSON);
+    } else {
+      traceBucket = new Set([spanJSON]);
+      this._spanTreeMap.set(traceId, traceBucket);
+    }
+
+    if (traceBucket.size >= this._maxSpanLimit) {
+      this._flushTrace(traceId);
+      this._debounceFlushInterval();
+    }
+  }
+
+  /**
+   * Flush all buffered traces.
+   */
+  public flush(): void {
+    if (!this._spanTreeMap.size) {
+      return;
+    }
+
+    DEBUG_BUILD && debug.log(`Flushing span tree map with ${this._spanTreeMap.size} traces`);
+
+    this._spanTreeMap.forEach((_, traceId) => {
+      this._flushTrace(traceId);
+    });
+    this._debounceFlushInterval();
+  }
+
+  private _flushTrace(traceId: string): void {
+    const traceBucket = this._spanTreeMap.get(traceId);
+    if (!traceBucket) {
+      return;
+    }
+
+    if (!traceBucket.size) {
+      this._spanTreeMap.delete(traceId);
+      return;
+    }
+
+    const firstSpanJSON = traceBucket.values().next().value;
+
+    const segmentSpan = firstSpanJSON?._segmentSpan;
+    if (!segmentSpan) {
+      DEBUG_BUILD && debug.warn('No segment span reference found on span JSON, cannot compute DSC');
+      this._spanTreeMap.delete(traceId);
+      return;
+    }
+
+    const dsc = getDynamicSamplingContextFromSpan(segmentSpan);
+
+    const cleanedSpans: SpanV2JSON[] = Array.from(traceBucket).map(spanJSON => {
+      // eslint-disable-next-line @typescript-eslint/no-unused-vars
+      const { _segmentSpan, ...cleanSpanJSON } = spanJSON;
+      return cleanSpanJSON;
+    });
+
+    const envelope = createSpanV2Envelope(cleanedSpans, dsc, this._client);
+
+    DEBUG_BUILD && debug.log(`Sending span envelope for trace ${traceId} with ${cleanedSpans.length} spans`);
+
+    this._client.sendEnvelope(envelope).then(null, reason => {
+      DEBUG_BUILD && debug.error('Error while sending span stream envelope:', reason);
+    });
+
+    this._spanTreeMap.delete(traceId);
+  }
+
+  private _debounceFlushInterval(): void {
+    if (this._flushIntervalId) {
+      clearInterval(this._flushIntervalId);
+    }
+    this._flushIntervalId = setInterval(() => {
+      this.flush();
+    }, this._flushInterval);
+  }
+}