perf(telemetry): batch product analytics exports

- send product analytics spans in a single OTLP request
- add exponential retry backoff and warning throttling
- cover batching payload and retry delay helpers with tests

Author: tyulyukov

apps/server/src/telemetry/Layers/AnalyticsService.test.ts

@@ -0,0 +1,12 @@
+import { assert, describe, it } from "@effect/vitest";
+
+import { productAnalyticsRetryDelayMs } from "./AnalyticsService.ts";
+
+describe("productAnalyticsRetryDelayMs", () => {
+  it("backs product analytics retries off exponentially with a cap", () => {
+    assert.equal(productAnalyticsRetryDelayMs(1), 30_000);
+    assert.equal(productAnalyticsRetryDelayMs(2), 60_000);
+    assert.equal(productAnalyticsRetryDelayMs(3), 120_000);
+    assert.equal(productAnalyticsRetryDelayMs(20), 300_000);
+  });
+});

apps/server/src/telemetry/Layers/AnalyticsService.ts

@@ -6,7 +6,7 @@ import { AnalyticsService } from "../Services/AnalyticsService.ts";
 import { getTelemetryIdentifier } from "../Identify.ts";
 import {
   JIRA_ACCESS_TOKEN_HEADER,
-  makeProductSpanPayload,
+  makeProductSpanBatchPayload,
   productAnalyticsUrlFromConfig,
   shouldAttachJiraProof,
 } from "../OtlpProduct.ts";
@@ -24,16 +24,35 @@ interface BufferedAnalyticsEvent {
 
 const MAX_BUFFERED_EVENTS = 1_000;
 const FLUSH_BATCH_SIZE = 20;
+const FLUSH_INTERVAL_MS = 1_000;
+const INITIAL_RETRY_DELAY_MS = 30_000;
+const MAX_RETRY_DELAY_MS = 5 * 60_000;
+const FAILURE_LOG_THROTTLE_MS = 60_000;
+
+interface ProductAnalyticsExportState {
+  readonly failures: number;
+  readonly nextFlushAt: number;
+  readonly lastWarningAt: number;
+}
 
 class ProductAnalyticsExportError extends Data.TaggedError("ProductAnalyticsExportError")<{
   readonly cause: unknown;
 }> {}
 
+export function productAnalyticsRetryDelayMs(failures: number): number {
+  return Math.min(MAX_RETRY_DELAY_MS, INITIAL_RETRY_DELAY_MS * 2 ** Math.max(0, failures - 1));
+}
+
 const makeAnalyticsService = Effect.gen(function* () {
   const config = yield* ServerConfig;
   const jiraTokenService = yield* Effect.serviceOption(JiraTokenService);
   const identifier = yield* getTelemetryIdentifier;
   const bufferRef = yield* Ref.make<ReadonlyArray<BufferedAnalyticsEvent>>([]);
+  const exportStateRef = yield* Ref.make<ProductAnalyticsExportState>({
+    failures: 0,
+    nextFlushAt: 0,
+    lastWarningAt: 0,
+  });
   const productAnalyticsUrl = productAnalyticsUrlFromConfig(config);
 
   const makeBaseAttributes = () => ({
@@ -71,40 +90,41 @@ const makeAnalyticsService = Effect.gen(function* () {
     Effect.gen(function* () {
       if (!productAnalyticsUrl || events.length === 0) return;
       const proofHeader = yield* getJiraProofHeader;
-      yield* Effect.forEach(
-        events,
-        (event) =>
-          Effect.tryPromise({
-            try: async () => {
-              const response = await fetch(productAnalyticsUrl, {
-                method: "POST",
-                headers: {
-                  "Content-Type": "application/json",
-                  ...proofHeader,
-                },
-                body: JSON.stringify(
-                  makeProductSpanPayload({
-                    event: event.event,
-                    capturedAt: event.capturedAt,
-                    attributes: {
-                      ...makeBaseAttributes(),
-                      ...event.properties,
-                    },
-                  }),
-                ),
-              });
-              if (!response.ok) {
-                throw new Error(`Product analytics export failed with status ${response.status}`);
-              }
+      yield* Effect.tryPromise({
+        try: async () => {
+          const response = await fetch(productAnalyticsUrl, {
+            method: "POST",
+            headers: {
+              "Content-Type": "application/json",
+              ...proofHeader,
             },
-            catch: (cause) => new ProductAnalyticsExportError({ cause }),
-          }),
-        { discard: true, concurrency: 2 },
-      );
+            body: JSON.stringify(
+              makeProductSpanBatchPayload(
+                events.map((event) => ({
+                  event: event.event,
+                  capturedAt: event.capturedAt,
+                  attributes: {
+                    ...makeBaseAttributes(),
+                    ...event.properties,
+                  },
+                })),
+              ),
+            ),
+          });
+          if (!response.ok) {
+            throw new Error(`Product analytics export failed with status ${response.status}`);
+          }
+        },
+        catch: (cause) => new ProductAnalyticsExportError({ cause }),
+      });
     });
 
   const flush = Effect.gen(function* () {
     while (true) {
+      const now = Date.now();
+      const exportState = yield* Ref.get(exportStateRef);
+      if (now < exportState.nextFlushAt) return;
+
       const batch = yield* Ref.modify(bufferRef, (current) => {
         if (current.length === 0) {
           return [[] as ReadonlyArray<BufferedAnalyticsEvent>, current] as const;
@@ -115,14 +135,34 @@ const makeAnalyticsService = Effect.gen(function* () {
       });
       if (batch.length === 0) return;
       yield* sendBatch(batch).pipe(
+        Effect.tap(() =>
+          Ref.set(exportStateRef, {
+            failures: 0,
+            nextFlushAt: 0,
+            lastWarningAt: 0,
+          }),
+        ),
         Effect.catch((cause) =>
-          Effect.all(
-            [
-              Ref.update(bufferRef, (current) => [...batch, ...current]),
-              Effect.logWarning("Failed to flush product analytics", { cause }),
-            ],
-            { discard: true },
-          ),
+          Effect.gen(function* () {
+            const previous = yield* Ref.get(exportStateRef);
+            const failures = previous.failures + 1;
+            const retryDelayMs = productAnalyticsRetryDelayMs(failures);
+            const shouldLog =
+              previous.failures === 0 || now - previous.lastWarningAt >= FAILURE_LOG_THROTTLE_MS;
+
+            yield* Ref.update(bufferRef, (current) => [...batch, ...current]);
+            yield* Ref.set(exportStateRef, {
+              failures,
+              nextFlushAt: now + retryDelayMs,
+              lastWarningAt: shouldLog ? now : previous.lastWarningAt,
+            });
+            if (shouldLog) {
+              yield* Effect.logWarning("Failed to flush product analytics", {
+                cause,
+                retryDelayMs,
+              });
+            }
+          }),
         ),
       );
     }
@@ -151,7 +191,7 @@ const makeAnalyticsService = Effect.gen(function* () {
       });
     });
 
-  yield* Effect.forever(Effect.sleep(1000).pipe(Effect.flatMap(() => flush)), {
+  yield* Effect.forever(Effect.sleep(FLUSH_INTERVAL_MS).pipe(Effect.flatMap(() => flush)), {
     disableYield: true,
   }).pipe(Effect.forkScoped);
   yield* Effect.addFinalizer(() => flush);

apps/server/src/telemetry/OtlpProduct.test.ts

@@ -0,0 +1,24 @@
+import { assert, describe, it } from "@effect/vitest";
+
+import { makeProductSpanBatchPayload } from "./OtlpProduct.ts";
+
+describe("makeProductSpanBatchPayload", () => {
+  it("packs multiple product analytics events into one OTLP trace request", () => {
+    const payload = makeProductSpanBatchPayload([
+      {
+        event: "marcode.first",
+        capturedAt: "2026-05-04T10:00:00.000Z",
+        attributes: { alpha: "one" },
+      },
+      {
+        event: "marcode.second",
+        capturedAt: "2026-05-04T10:00:01.000Z",
+        attributes: { beta: 2 },
+      },
+    ]);
+
+    assert.equal(payload.resourceSpans.length, 2);
+    assert.equal(payload.resourceSpans[0]?.scopeSpans[0]?.spans[0]?.name, "marcode.first");
+    assert.equal(payload.resourceSpans[1]?.scopeSpans[0]?.spans[0]?.name, "marcode.second");
+  });
+});

apps/server/src/telemetry/OtlpProduct.ts

@@ -67,6 +67,18 @@ export function makeProductSpanPayload(input: {
   };
 }
 
+export function makeProductSpanBatchPayload(
+  events: ReadonlyArray<{
+    readonly event: string;
+    readonly attributes: ProductAnalyticsAttributes;
+    readonly capturedAt: string;
+  }>,
+) {
+  return {
+    resourceSpans: events.flatMap((event) => makeProductSpanPayload(event).resourceSpans),
+  };
+}
+
 export function productAnalyticsUrlFromConfig(input: {
   readonly productAnalyticsTracesUrl: string | undefined;
   readonly jiraTokenProxyUrl: string | undefined;