fix(core): send logs and metrics through one OTLP batch sender

.changeset/logs-retry-transient-failures.md

@@ -0,0 +1,6 @@
+---
+'posthog-react-native': patch
+'@posthog/core': patch
+---
+
+Fix buffered logs being dropped instead of retried after HTTP 408, 429 or 5xx

.changeset/web-logs-metrics-retry-408.md

@@ -0,0 +1,5 @@
+---
+'posthog-js': patch
+---
+
+Fix logs and metrics batches being dropped instead of retried after HTTP 408

packages/browser/src/__tests__/posthog-logs.test.ts

@@ -1098,6 +1098,11 @@ describe('posthog-logs', () => {
                 expect((logs as any)._queue).toHaveLength(1)
             })
 
+            it('keeps records on a 408 so they retry later', async () => {
+                await flushWith(408)
+                expect((logs as any)._queue).toHaveLength(1)
+            })
+
             it('drops records on a 4xx client error', async () => {
                 await flushWith(400)
                 expect((logs as any)._queue).toHaveLength(0)

packages/browser/src/__tests__/posthog-metrics.test.ts

@@ -93,6 +93,16 @@ describe('posthog-metrics', () => {
         expect(last.sum.dataPoints[0].asDouble).toBe(2)
     })
 
+    it('retains the window on a 408', async () => {
+        respondWithStatus(408)
+        metrics.count('a', 5)
+        await metrics.flush()
+
+        respondWithStatus(200)
+        await metrics.flush()
+        expect(sentRequests()[1].data.resourceMetrics[0].scopeMetrics[0].metrics[0].sum.dataPoints[0].asDouble).toBe(5)
+    })
+
     it('captures nothing when the instance is not capturing', async () => {
         ;(mockPostHog.is_capturing as jest.Mock).mockReturnValue(false)
         metrics.count('a', 1)

packages/browser/src/posthog-logs.ts

@@ -343,8 +343,8 @@ export class PostHogLogs implements Extension {
                         settle({ kind: 'ok' })
                     } else if (status === 413) {
                         settle({ kind: 'too-large' })
-                    } else if (status === 0 || status === 429 || status >= 500) {
-                        // Transient (network / rate-limit / server error): keep and retry.
+                    } else if (status === 0 || status === 408 || status === 429 || status >= 500) {
+                        // Transient (network / timeout / rate-limit / server error): keep and retry.
                         if (status === 0) {
                             // `_send_request` already logs fetch failures. Bare status 0 is the
                             // XHR/synthetic path, so keep one warning for it here.

packages/browser/src/posthog-metrics.ts

@@ -151,8 +151,8 @@ export class PostHogMetrics implements Extension {
                         settle({ kind: 'ok' })
                     } else if (status === 413) {
                         settle({ kind: 'too-large' })
-                    } else if (status === 0 || status === 429 || status >= 500) {
-                        // Transient (network / rate-limit / server error): keep and retry.
+                    } else if (status === 0 || status === 408 || status === 429 || status >= 500) {
+                        // Transient (network / timeout / rate-limit / server error): keep and retry.
                         settle({
                             kind: 'retry-later',
                             error: response.error ?? new Error(`metrics request failed with status ${status}`),

packages/core/src/__tests__/posthog.flush.spec.ts

@@ -760,4 +760,84 @@ describe('PostHog Core', () => {
       ])
     })
   })
+
+  describe('OTLP batch senders', () => {
+    // Both share one `_sendOtlpBatch`; the table pins them to the same
+    // classification so a wrapper can't reintroduce a per-signal retry policy.
+    const senders = {
+      logs: (client: PostHogCoreTestClient) => client._sendLogsBatch({ resourceLogs: [] }),
+      metrics: (client: PostHogCoreTestClient) => client._sendMetricsBatch({ resourceMetrics: [] }),
+    }
+
+    const cases: [number, string][] = [
+      [408, 'retry-later'],
+      [429, 'retry-later'],
+      [500, 'retry-later'],
+      [503, 'retry-later'],
+      [413, 'too-large'],
+      [400, 'fatal'],
+      [401, 'fatal'],
+    ]
+
+    describe.each(Object.entries(senders))('%s', (_name, send) => {
+      it.each(cases)('classifies an exhausted %i as %s', async (status, kind) => {
+        ;[posthog, mocks] = createTestClient('TEST_API_KEY', {
+          fetchRetryCount: 0,
+          preloadFeatureFlags: false,
+        })
+        mocks.fetch.mockResolvedValue({
+          status,
+          text: async () => 'err',
+          json: async () => ({ status: 'err' }),
+        })
+
+        await expect(send(posthog)).resolves.toMatchObject({ kind })
+        expect(mocks.fetch).toHaveBeenCalledTimes(1)
+      })
+
+      // The classification cases run with retries off, so they never reach the
+      // `retryCheck` the senders share. This pins it: 413 has to leave the
+      // transport on the first response for the caller to shrink its batch and
+      // retry the same records, where a 5xx is worth re-sending as-is.
+      it.each([
+        [408, 3],
+        [429, 3],
+        [500, 3],
+        [503, 3],
+        [413, 1],
+        [400, 1],
+      ])('sends %i %i time(s) before returning', async (status, attempts) => {
+        jest.useRealTimers()
+        ;[posthog, mocks] = createTestClient('TEST_API_KEY', {
+          fetchRetryCount: 2,
+          fetchRetryDelay: 1,
+          preloadFeatureFlags: false,
+        })
+        mocks.fetch.mockResolvedValue({
+          status,
+          text: async () => 'err',
+          json: async () => ({ status: 'err' }),
+        })
+
+        await send(posthog)
+        expect(mocks.fetch).toHaveBeenCalledTimes(attempts)
+      })
+
+      it('carries the HTTP error on a retry-later outcome', async () => {
+        ;[posthog, mocks] = createTestClient('TEST_API_KEY', {
+          fetchRetryCount: 0,
+          preloadFeatureFlags: false,
+        })
+        mocks.fetch.mockResolvedValue({
+          status: 503,
+          text: async () => 'err',
+          json: async () => ({ status: 'err' }),
+        })
+
+        await expect(send(posthog)).resolves.toMatchObject({
+          error: { name: 'PostHogFetchHttpError', status: 503 },
+        })
+      })
+    })
+  })
 })

packages/core/src/logs/index.spec.ts

@@ -1,4 +1,5 @@
 import { PostHogPersistedProperty } from '../types'
+import { createTestClient, PostHogCoreTestClient } from '../testing'
 import type { Logger } from '../types'
 import { PostHogLogs } from './index'
 import type { BufferedLogEntry, ResolvedPostHogLogsConfig } from './types'
@@ -80,6 +81,82 @@ const getContextFor = (instance: any) => (): { distinctId?: string; sessionId?:
   sessionId: instance.getSessionId() || undefined,
 })
 
+// Drives a real core host rather than a stubbed `_sendLogsBatch`, so the sender's
+// error classification and the queue bookkeeping are exercised together.
+describe('PostHogLogs over the core sender', () => {
+  const createLogsOverCore = (status: number): { logs: PostHogLogs; client: PostHogCoreTestClient } => {
+    const [client, mocks] = createTestClient('TEST_API_KEY', {
+      fetchRetryCount: 0,
+      preloadFeatureFlags: false,
+    })
+    mocks.fetch.mockResolvedValue({
+      status,
+      text: () => Promise.resolve('err'),
+      json: () => Promise.resolve({ status: 'err' }),
+    })
+    const logs = new PostHogLogs(
+      client,
+      resolveForTest(),
+      createMockLogger(),
+      () => ({ distinctId: 'user-123' }),
+      immediateOnReady
+    )
+    return { logs, client }
+  }
+
+  const queueOf = (client: PostHogCoreTestClient): BufferedLogEntry[] =>
+    client.getPersistedProperty<BufferedLogEntry[]>(PostHogPersistedProperty.LogsQueue) ?? []
+
+  it.each([408, 429, 500, 503])('keeps records queued when the endpoint answers %i', async (status) => {
+    const { logs, client } = createLogsOverCore(status)
+    logs.captureLog({ body: 'keep me' })
+
+    await expect(logs.flush()).rejects.toHaveProperty('name', 'PostHogFetchHttpError')
+
+    expect(queueOf(client)).toHaveLength(1)
+  })
+
+  it('retains and resends records after transport retries are exhausted', async () => {
+    jest.useRealTimers()
+    const [client, mocks] = createTestClient('TEST_API_KEY', {
+      fetchRetryCount: 2,
+      fetchRetryDelay: 1,
+      preloadFeatureFlags: false,
+    })
+    const unavailableResponse = { status: 503, text: async () => 'unavailable', json: async () => ({}) }
+    mocks.fetch
+      .mockResolvedValueOnce(unavailableResponse)
+      .mockResolvedValueOnce(unavailableResponse)
+      .mockResolvedValueOnce(unavailableResponse)
+      .mockResolvedValueOnce({ status: 200, text: async () => 'ok', json: async () => ({}) })
+    const logs = new PostHogLogs(
+      client,
+      resolveForTest(),
+      createMockLogger(),
+      () => ({ distinctId: 'user-123' }),
+      immediateOnReady
+    )
+
+    logs.captureLog({ body: 'retry me' })
+    await expect(logs.flush()).rejects.toHaveProperty('name', 'PostHogFetchHttpError')
+    expect(mocks.fetch).toHaveBeenCalledTimes(3)
+    expect(queueOf(client)).toHaveLength(1)
+
+    await expect(logs.flush()).resolves.toBeUndefined()
+    expect(mocks.fetch).toHaveBeenCalledTimes(4)
+    expect(queueOf(client)).toHaveLength(0)
+  })
+
+  it('drops the batch when the endpoint answers 401', async () => {
+    const { logs, client } = createLogsOverCore(401)
+    logs.captureLog({ body: 'unauthorized' })
+
+    await expect(logs.flush()).rejects.toHaveProperty('name', 'PostHogFetchHttpError')
+
+    expect(queueOf(client)).toHaveLength(0)
+  })
+})
+
 describe('PostHogLogs', () => {
   let mockInstance: any
   let logger: Logger

packages/core/src/logs/index.ts

@@ -262,7 +262,7 @@ export class PostHogLogs {
       }
 
       if (outcome.kind === 'retry-later') {
-        // Network error: keep records in the queue for the next flush cycle
+        // Transient failure: keep records in the queue for the next flush cycle
         // and surface the error so the caller can log/react.
         throw outcome.error
       }

packages/core/src/posthog-core-stateless.ts

@@ -222,13 +222,12 @@ function isPostHogEventProperties(value: JsonType | undefined): value is PostHog
 }
 
 /**
- * Outcome of a logs batch send. Keeps HTTP error classification inside core
- * (single source of truth — same policy events already use in `_flush()`) so
+ * Outcome of a logs batch send. Keeps HTTP error classification inside core so
  * PostHogLogs doesn't need to know about specific error types.
  *
  *   - ok            → records are accepted; drop them from the queue
  *   - too-large     → 413; caller should halve batch size and retry same records
- *   - retry-later   → network error; caller keeps records and retries next cycle
+ *   - retry-later   → retryable network or HTTP error; caller keeps records and retries next cycle
  *   - fatal         → anything else (auth, malformed, etc.); caller drops the
  *                     batch and surfaces the error
  */
@@ -238,6 +237,17 @@ export type SendLogsBatchOutcome =
   | { kind: 'retry-later'; error: unknown }
   | { kind: 'fatal'; error: unknown }
 
+/**
+ * Each signal keeps its own exported outcome type because each belongs to a
+ * separate host contract. The wrappers return this value directly, so one
+ * drifting out of shape fails to compile.
+ */
+type SendOtlpBatchOutcome =
+  | { kind: 'ok' }
+  | { kind: 'too-large' }
+  | { kind: 'retry-later'; error: unknown }
+  | { kind: 'fatal'; error: unknown }
+
 export enum QuotaLimitedFeature {
   FeatureFlags = 'feature_flags',
   Recordings = 'recordings',
@@ -1633,22 +1643,28 @@ export abstract class PostHogCoreStateless {
   }
 
   /**
-   * Sends a pre-built OTLP logs payload to `/i/v1/logs`. Returns a tagged
-   * outcome instead of throwing so PostHogLogs doesn't have to know about the
-   * core's error class hierarchy. Error classification lives here (single
-   * source of truth, same policy the events `_flush()` uses for its own
-   * 413 / network / fatal handling).
+   * Shared implementation behind the OTLP senders, which differ only in path.
+   * Returns a tagged outcome instead of throwing so the queue owners don't
+   * have to know the core's error class hierarchy.
    *
-   * 413 is passed through as `too-large` (not auto-retried) so the caller can
-   * shrink `maxBatchRecordsPerPost` and retry the same records.
+   * Exhausted 408/429/5xx stay `retry-later`, unlike the events `_flush()`
+   * which drops anything that isn't a network error: every OTLP queue is
+   * bounded and retried with backoff, so holding a batch through an outage can
+   * neither grow without limit nor spin.
    */
-  async _sendLogsBatch(payload: OtlpLogsPayload): Promise<SendLogsBatchOutcome> {
+  private async _sendOtlpBatch({
+    path,
+    payload,
+  }: {
+    path: 'logs' | 'metrics'
+    payload: OtlpLogsPayload | OtlpMetricsPayload
+  }): Promise<SendOtlpBatchOutcome> {
     if (this.disabled) {
       return { kind: 'fatal', error: new Error('The client is disabled') }
     }
 
     const serialized = JSON.stringify(payload)
-    const url = `${this.host}/i/v1/logs?token=${encodeURIComponent(this.apiKey)}`
+    const url = `${this.host}/i/v1/${path}?token=${encodeURIComponent(this.apiKey)}`
 
     const gzippedPayload = !this.disableCompression ? await this.compressPayload(serialized) : null
     const fetchOptions: PostHogFetchOptions = {
@@ -1680,65 +1696,19 @@ export abstract class PostHogCoreStateless {
       if (isPostHogFetchContentTooLargeError(err)) {
         return { kind: 'too-large' }
       }
-      if (err instanceof PostHogFetchNetworkError) {
+      if (isPostHogFetchRetryableError(err)) {
         return { kind: 'retry-later', error: err }
       }
       return { kind: 'fatal', error: err }
     }
   }
 
-  /**
-   * Sends a pre-built OTLP metrics payload to `/i/v1/metrics`. Same tagged
-   * outcome contract and error classification as `_sendLogsBatch` — this is
-   * the `MetricsHost._sendMetricsBatch` implementation, so `PostHogMetrics`
-   * can use any core-based SDK as its host.
-   */
-  async _sendMetricsBatch(payload: OtlpMetricsPayload): Promise<SendMetricsBatchOutcome> {
-    if (this.disabled) {
-      return { kind: 'fatal', error: new Error('The client is disabled') }
-    }
-
-    const serialized = JSON.stringify(payload)
-    const url = `${this.host}/i/v1/metrics?token=${encodeURIComponent(this.apiKey)}`
-
-    const gzippedPayload = !this.disableCompression ? await this.compressPayload(serialized) : null
-    const fetchOptions: PostHogFetchOptions = {
-      method: 'POST',
-      headers: {
-        ...this.getCustomHeaders(),
-        'Content-Type': 'application/json',
-        ...(gzippedPayload !== null && { 'Content-Encoding': 'gzip' }),
-      },
-      body: gzippedPayload || serialized,
-    }
+  async _sendLogsBatch(payload: OtlpLogsPayload): Promise<SendLogsBatchOutcome> {
+    return this._sendOtlpBatch({ path: 'logs', payload })
+  }
 
-    try {
-      await this.fetchWithRetry(
-        url,
-        fetchOptions,
-        { type: 'successful-write' },
-        {
-          retryCheck: (err) => {
-            if (isPostHogFetchContentTooLargeError(err)) {
-              return false
-            }
-            return isPostHogFetchRetryableError(err)
-          },
-        }
-      )
-      return { kind: 'ok' }
-    } catch (err) {
-      if (isPostHogFetchContentTooLargeError(err)) {
-        return { kind: 'too-large' }
-      }
-      // Exhausted retries on a retryable failure (network error, 408/429/5xx)
-      // still classify as retry-later so the window rides the next flush; only
-      // non-retryable HTTP errors (and 413 above) drop the batch.
-      if (isPostHogFetchRetryableError(err)) {
-        return { kind: 'retry-later', error: err }
-      }
-      return { kind: 'fatal', error: err }
-    }
+  async _sendMetricsBatch(payload: OtlpMetricsPayload): Promise<SendMetricsBatchOutcome> {
+    return this._sendOtlpBatch({ path: 'metrics', payload })
   }
 
   private fetchWithRetry<T>(