feat(sdk): opt-in OTLP payload compression (gzip + zstd)

src/sdk/otel.ts

@@ -731,6 +731,22 @@ function bootObservability(opts: OtelOptions, env: Record<string, unknown>): voi
   // Priority: auth token (lowest) → DECO_OTEL_HEADERS env → otlpHeaders option (highest).
   const otlpHeaders: Record<string, string> = { ...authHeader, ...parsedEnvHeaders, ...(opts.otlpHeaders ?? {}) };
 
+  // Compression: opt-in via `DECO_OTEL_COMPRESSION=gzip|zstd`. Default
+  // `"none"` preserves the pre-flag wire behavior for sites whose collector
+  // isn't configured for `Content-Encoding`. `zstd` silently falls back to
+  // gzip if the runtime lacks `node:zlib.zstdCompress` (workerd without
+  // `nodejs_compat`, Node <22, etc.); the request always advertises the
+  // actual encoding via `Content-Encoding`.
+  const otlpCompressionRaw = (
+    (env["DECO_OTEL_COMPRESSION"] as string | undefined) ?? ""
+  ).toLowerCase();
+  const otlpCompression: "zstd" | "gzip" | "none" =
+    otlpCompressionRaw === "zstd"
+      ? "zstd"
+      : otlpCompressionRaw === "gzip"
+        ? "gzip"
+        : "none";
+
   const errorPromotionEnvVar =
     opts.otlpTracesErrorPromotionEnvVar ?? "DECO_OTEL_ERROR_PROMOTION";
   const errorPromotionEnabled =
@@ -744,6 +760,7 @@ function bootObservability(opts: OtelOptions, env: Record<string, unknown>): voi
       scopeVersion: decoRuntimeVersion,
       minLevel: otlpLogsMinLevel,
       headers: otlpHeaders,
+      compression: otlpCompression,
       fetchImpl: opts.otlpLogsFetchImpl,
       promoteTrace: errorPromotionEnabled
         ? (traceId) => state.otlpTracer?.promoteTrace(traceId)
@@ -813,6 +830,7 @@ function bootObservability(opts: OtelOptions, env: Record<string, unknown>): voi
       resourceAttributes: floor,
       scopeVersion: decoRuntimeVersion,
       headers: otlpHeaders,
+      compression: otlpCompression,
       fetchImpl: opts.otlpMetricsFetchImpl,
       metricMetadata: METRIC_METADATA,
       onError: (kind, err) => {
@@ -883,6 +901,7 @@ function bootObservability(opts: OtelOptions, env: Record<string, unknown>): voi
       headSamplingRate: samplingRateOverride ?? opts.otlpTracesSamplingRate ?? 0.01,
       errorPromotionRate,
       headers: otlpHeaders,
+      compression: otlpCompression,
       fetchImpl: opts.otlpTracesFetchImpl,
       getActiveSpanForParent: () => getActiveSpan(),
       getRequestTraceContext,

src/sdk/otelHttpLog.ts

@@ -41,6 +41,7 @@
 
 import { getActiveSpan } from "./observability";
 import type { LoggerAdapter, LogLevel } from "./logger";
+import { compressJsonPayload, type OtlpCompression } from "./otlpCompression";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -86,6 +87,12 @@ export interface OtlpHttpLogOptions {
    * `Content-Type: application/json` is always set and cannot be overridden here.
    */
   headers?: Record<string, string>;
+  /**
+   * Payload compression. `"gzip"` uses the runtime's native `CompressionStream`
+   * and sets `Content-Encoding: gzip`. Defaults to `"none"` — the caller
+   * (`bootObservability`) opts into gzip via `DECO_OTEL_COMPRESSION`.
+   */
+  compression?: OtlpCompression;
   /** Test seam — override fetch. */
   fetchImpl?: typeof fetch;
   /** Test seam — override Date.now(). */
@@ -154,6 +161,7 @@ export function createOtlpHttpLogAdapter(options: OtlpHttpLogOptions): OtlpHttpL
   const refillPerMinute = options.rateLimitRefillPerMinute ?? 1000;
   const minSeverity = SEVERITY_NUMBER[options.minLevel ?? "error"];
   const extraHeaders = options.headers ?? {};
+  const compression: OtlpCompression = options.compression ?? "none";
   const fetchImpl = options.fetchImpl ?? fetch;
   const now = options.nowMs ?? (() => Date.now());
   const onError = options.onError;
@@ -301,10 +309,16 @@ export function createOtlpHttpLogAdapter(options: OtlpHttpLogOptions): OtlpHttpL
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), flushTimeoutMs);
     try {
+      const { body, contentEncoding } = await compressJsonPayload(payload, compression);
+      const headers: Record<string, string> = {
+        ...extraHeaders,
+        "Content-Type": "application/json",
+      };
+      if (contentEncoding) headers["Content-Encoding"] = contentEncoding;
       const res = await fetchImpl(endpoint, {
         method: "POST",
-        headers: { ...extraHeaders, "Content-Type": "application/json" },
-        body: JSON.stringify(payload),
+        headers,
+        body,
         signal: controller.signal,
       });
       if (!res.ok) {

src/sdk/otelHttpMeter.ts

@@ -32,6 +32,7 @@
  */
 
 import type { MeterAdapter } from "./observability";
+import { compressJsonPayload, type OtlpCompression } from "./otlpCompression";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -65,6 +66,11 @@ export interface OtlpHttpMeterOptions {
    * `Content-Type: application/json` is always set and cannot be overridden here.
    */
   headers?: Record<string, string>;
+  /**
+   * Payload compression. `"gzip"` uses the runtime's native `CompressionStream`
+   * and sets `Content-Encoding: gzip`. Defaults to `"none"`.
+   */
+  compression?: OtlpCompression;
   /**
    * Test seam — override fetch for the flush path so unit tests can
    * inspect the OTLP payload without going to the network.
@@ -153,6 +159,7 @@ export function createOtlpHttpMeterAdapter(options: OtlpHttpMeterOptions): OtlpH
   const minFlushIntervalMs = options.minFlushIntervalMs ?? 5000;
   const flushTimeoutMs = options.flushTimeoutMs ?? 5000;
   const extraHeaders = options.headers ?? {};
+  const compression: OtlpCompression = options.compression ?? "none";
   const fetchImpl = options.fetchImpl ?? fetch;
   const now = options.nowMs ?? (() => Date.now());
   const onError = options.onError;
@@ -317,10 +324,16 @@ export function createOtlpHttpMeterAdapter(options: OtlpHttpMeterOptions): OtlpH
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), flushTimeoutMs);
     try {
+      const { body, contentEncoding } = await compressJsonPayload(payload, compression);
+      const headers: Record<string, string> = {
+        ...extraHeaders,
+        "Content-Type": "application/json",
+      };
+      if (contentEncoding) headers["Content-Encoding"] = contentEncoding;
       const res = await fetchImpl(endpoint, {
         method: "POST",
-        headers: { ...extraHeaders, "Content-Type": "application/json" },
-        body: JSON.stringify(payload),
+        headers,
+        body,
         signal: controller.signal,
       });
       if (!res.ok) {

src/sdk/otelHttpTracer.ts

@@ -34,6 +34,7 @@
  */
 
 import type { Span, TracerAdapter } from "../middleware/observability";
+import { compressJsonPayload, type OtlpCompression } from "./otlpCompression";
 
 // ---------------------------------------------------------------------------
 // W3C traceparent parsing
@@ -201,6 +202,11 @@ export interface OtlpHttpTracerOptions {
    * `Content-Type: application/json` is always set and cannot be overridden here.
    */
   headers?: Record<string, string>;
+  /**
+   * Payload compression. `"gzip"` uses the runtime's native `CompressionStream`
+   * and sets `Content-Encoding: gzip`. Defaults to `"none"`.
+   */
+  compression?: OtlpCompression;
   /**
    * Test seam — override `fetch` for the flush path. Same role as in
    * `otelHttpMeter.ts`.
@@ -259,6 +265,7 @@ export function createOtlpHttpTracerAdapter(options: OtlpHttpTracerOptions): Otl
   const minFlushIntervalMs = options.minFlushIntervalMs ?? 5000;
   const flushTimeoutMs = options.flushTimeoutMs ?? 5000;
   const extraHeaders = options.headers ?? {};
+  const compression: OtlpCompression = options.compression ?? "none";
   const fetchImpl = options.fetchImpl ?? fetch;
   const now = options.nowMs ?? (() => Date.now());
   const onError = options.onError;
@@ -412,10 +419,16 @@ export function createOtlpHttpTracerAdapter(options: OtlpHttpTracerOptions): Otl
     const controller = new AbortController();
     const timer = setTimeout(() => controller.abort(), flushTimeoutMs);
     try {
+      const { body, contentEncoding } = await compressJsonPayload(payload, compression);
+      const headers: Record<string, string> = {
+        ...extraHeaders,
+        "Content-Type": "application/json",
+      };
+      if (contentEncoding) headers["Content-Encoding"] = contentEncoding;
       const res = await fetchImpl(endpoint, {
         method: "POST",
-        headers: { ...extraHeaders, "Content-Type": "application/json" },
-        body: JSON.stringify(payload),
+        headers,
+        body,
         signal: controller.signal,
       });
       if (!res.ok) {

src/sdk/otlpCompression.ts

@@ -0,0 +1,82 @@
+/**
+ * OTLP/HTTP payload compression.
+ *
+ * Two algorithms:
+ *  - `gzip`: native `CompressionStream("gzip")`. Zero deps, works in every
+ *    runtime (workerd, Node ≥18, Deno, Bun).
+ *  - `zstd`: `node:zlib` `zstdCompress`. Requires `nodejs_compat` on workerd
+ *    (zstd landed in Mar 2026 via cloudflare/workerd#4013) and Node ≥22 on k8s.
+ *    ~15-25% smaller than gzip on OTLP JSON batches and cheaper for the
+ *    collector to decode. `CompressionStream("zstd")` is still non-standard
+ *    (Bun-only, Feb 2026) — revisit when workerd ships it too, then this
+ *    file collapses to a single line per format.
+ *
+ * When `compression: "zstd"` is requested but the runtime doesn't provide
+ * `node:zlib.zstdCompress` (no `nodejs_compat`, Node <22, etc.), we silently
+ * fall back to gzip. The response header always reflects the actual encoding
+ * used, so the collector doesn't need to know which transport was picked.
+ */
+
+export type OtlpCompression = "zstd" | "gzip" | "none";
+
+export async function compressJsonPayload(
+  payload: unknown,
+  compression: OtlpCompression,
+): Promise<{ body: BodyInit; contentEncoding?: string }> {
+  const json = JSON.stringify(payload);
+  if (compression === "none") return { body: json };
+
+  const bytes = new TextEncoder().encode(json);
+
+  if (compression === "zstd") {
+    const zstdBytes = await tryZstd(bytes);
+    if (zstdBytes) return { body: zstdBytes, contentEncoding: "zstd" };
+    // Fall through to gzip when zstd is unavailable in this runtime.
+  }
+
+  const stream = new Response(bytes).body!.pipeThrough(
+    new CompressionStream("gzip"),
+  );
+  const gzipBytes = new Uint8Array(await new Response(stream).arrayBuffer());
+  return { body: gzipBytes, contentEncoding: "gzip" };
+}
+
+// Lazy, cached probe of `node:zlib` zstd. `undefined` = not yet probed,
+// `null` = probed and unavailable, function = probed and usable.
+let zstdImpl: ((bytes: Uint8Array) => Promise<Uint8Array>) | null | undefined;
+
+async function tryZstd(bytes: Uint8Array): Promise<Uint8Array | null> {
+  if (zstdImpl === undefined) {
+    zstdImpl = await probeZstd();
+  }
+  if (!zstdImpl) return null;
+  try {
+    return await zstdImpl(bytes);
+  } catch {
+    return null;
+  }
+}
+
+async function probeZstd(): Promise<((bytes: Uint8Array) => Promise<Uint8Array>) | null> {
+  try {
+    // Namespace import so bundlers don't try to resolve `node:zlib` at build
+    // time in browser/edge builds that never call this path.
+    const zlib = await import(/* @vite-ignore */ "node:zlib");
+    const zstdCompress = (zlib as unknown as {
+      zstdCompress?: (
+        buf: Uint8Array,
+        cb: (err: Error | null, result: Buffer) => void,
+      ) => void;
+    }).zstdCompress;
+    if (typeof zstdCompress !== "function") return null;
+    return (buf) =>
+      new Promise((resolve, reject) => {
+        zstdCompress(buf, (err, result) => {
+          if (err) reject(err);
+          else resolve(new Uint8Array(result));
+        });
+      });
+  } catch {
+    return null;
+  }
+}