PostHog · turnipdabeets · Sep 8, 2026
@@ -57,7 +57,7 @@ describe('OTLP bodies over the endpoint limit', () => {
     const unserializable: any = spansOf(8)
     unserializable.resourceSpans[0].scopeSpans[0].spans[0].self = unserializable
 
-    await expect(send(unserializable)).resolves.toEqual({ kind: 'too-large' })
+    await expect(send(unserializable)).resolves.toEqual({ kind: 'too-large', measuredLocally: true })
     expect(mocks.fetch).not.toHaveBeenCalled()
   })
 
@@ -71,7 +71,7 @@ describe('OTLP bodies over the endpoint limit', () => {
     ['logs', (payload: any) => posthog._sendLogsBatch(payload)],
     ['metrics', (payload: any) => posthog._sendMetricsBatch(payload)],
   ])('reports a %s batch over the limit as too-large without a request', async (_signal, send) => {
-    await expect(send(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({ kind: 'too-large' })
+    await expect(send(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({ kind: 'too-large', measuredLocally: true })
     expect(mocks.fetch).not.toHaveBeenCalled()
   })
 
@@ -81,7 +81,10 @@ describe('OTLP bodies over the endpoint limit', () => {
     vi.spyOn(posthog as any, 'compressPayload').mockResolvedValue(new Uint8Array(1024))
     ;(posthog as any).disableCompression = false
 
-    await expect(posthog._sendTracesBatch(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({ kind: 'too-large' })
+    await expect(posthog._sendTracesBatch(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({
+      kind: 'too-large',
+      measuredLocally: true,
+    })
     expect(mocks.fetch).not.toHaveBeenCalled()
   })
 
@@ -102,6 +105,7 @@ describe('OTLP bodies over the endpoint limit', () => {
   it('reports a batch one byte over the limit as too-large', async () => {
     await expect(posthog._sendTracesBatch(spansOf(LIMIT_BYTES - overheadBytes() + 1))).resolves.toEqual({
       kind: 'too-large',
+      measuredLocally: true,
     })
     expect(mocks.fetch).not.toHaveBeenCalled()
   })
@@ -113,7 +117,10 @@ describe('OTLP bodies over the endpoint limit', () => {
     const compressPayload = vi.spyOn(posthog as any, 'compressPayload')
     ;(posthog as any).disableCompression = false
 
-    await expect(posthog._sendTracesBatch(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({ kind: 'too-large' })
+    await expect(posthog._sendTracesBatch(spansOf(OVER_LIMIT_BYTES))).resolves.toEqual({
+      kind: 'too-large',
+      measuredLocally: true,
+    })
     expect(compressPayload).not.toHaveBeenCalled()
   })
 

@@ -21,7 +21,15 @@ import type { BeforeSendMetricFn, MetricAttributeValue, OtlpMetricsPayload } fro
 export type SendMetricsBatchOutcome =
   | { kind: 'ok' }
   | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
-  | { kind: 'too-large' }
+  | {
+      kind: 'too-large'
+      /**
+       * True when the SDK measured the body itself rather than the endpoint
+       * refusing it, so the caller can split this drain without lowering the
+       * batch size it keeps between them.
+       */
+      measuredLocally?: boolean
+    }
   | { kind: 'fatal'; error: unknown }
 
 /**

@@ -288,7 +288,15 @@ function isPostHogEventProperties(value: JsonType | undefined): value is PostHog
  */
 export type SendLogsBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'too-large' }
+  | {
+      kind: 'too-large'
+      /**
+       * True when the SDK measured the body itself rather than the endpoint
+       * refusing it, so the caller can split this drain without lowering the
+       * batch size it keeps between them.
+       */
+      measuredLocally?: boolean
+    }
   | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
   | { kind: 'fatal'; error: unknown }
 
@@ -299,7 +307,15 @@ export type SendLogsBatchOutcome =
  */
 type SendOtlpBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'too-large' }
+  | {
+      kind: 'too-large'
+      /**
+       * True when the SDK measured the body itself rather than the endpoint
+       * refusing it, so the caller can split this drain without lowering the
+       * batch size it keeps between them.
+       */
+      measuredLocally?: boolean
+    }
   | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
   | { kind: 'fatal'; error: unknown }
 
@@ -1382,11 +1398,20 @@ export abstract class PostHogCoreStateless {
     if (this.pendingFlushPromise) {
       return
     }
-    void this.flush().catch(async (err) => {
+    void this.flushAutomatic().catch(async (err) => {
       await logFlushError(err)
     })
   }
 
+  /**
+   * The flush the SDK runs on its own, from the interval timer or the `flushAt`
+   * threshold. Separate from `flush()` so a host can hold back work that an
+   * endpoint has asked it to wait on, which an explicit flush overrides.
+   */
+  protected flushAutomatic(): Promise<void> {
+    return this.flush()
+  }
+
   private async waitForPendingPromises(
     maxPromiseId: number,
     ignoredPromises: (Promise<any> | null | undefined)[] = []
@@ -1731,7 +1756,7 @@ export abstract class PostHogCoreStateless {
       this.logMsgIfDebug(() =>
         console.warn(`[PostHog] Could not serialize a ${path} batch; reporting it as too large`, error)
       )
-      return { kind: 'too-large' }
+      return { kind: 'too-large', measuredLocally: true }
     }
 
     // Measured on the uncompressed payload: the endpoint decompresses the body
@@ -1747,7 +1772,7 @@ export abstract class PostHogCoreStateless {
           `[PostHog] Not sending a ${path} batch of ${payloadBytes} bytes: the endpoint accepts at most ${OTLP_MAX_BODY_BYTES}`
         )
       )
-      return { kind: 'too-large' }
+      return { kind: 'too-large', measuredLocally: true }
     }
 
     const url =

@@ -1844,6 +1844,33 @@ describe('PostHogTraces', () => {
       expect(batchSizes).toEqual([3, 1, 2])
     })
 
+    it('splits a locally measured batch without shrinking the next drain', async () => {
+      // The SDK measured this body itself, so it knows the oversized span is gone
+      // once the batch is isolated. The next drain starts at full size instead of
+      // ramping back one healthy batch at a time.
+      const instance = createMockInstance({
+        _sendTracesBatch: vi
+          .fn()
+          .mockResolvedValueOnce({ kind: 'too-large', measuredLocally: true })
+          .mockResolvedValue({ kind: 'ok' }),
+      })
+      const traces = createTraces({ maxExportBatchSize: 512 }, instance)
+      for (let i = 0; i < 8; i++) {
+        traces.startSpan(`span-${i}`).end()
+      }
+      await traces.flush()
+      expect(sentPayloads(instance).map((p) => p.resourceSpans[0].scopeSpans[0].spans.length)).toEqual([8, 4, 4])
+
+      instance._sendTracesBatch.mockClear()
+      for (let i = 0; i < 8; i++) {
+        traces.startSpan(`later-${i}`).end()
+      }
+      await traces.flush()
+
+      // One batch, not the 6-then-2 that a persistent shrink plus its +1 ramp gives.
+      expect(sentPayloads(instance).map((p) => p.resourceSpans[0].scopeSpans[0].spans.length)).toEqual([8])
+    })
+
     it('ramps the batch size back up after a 413 shrink', async () => {
       // A one-off oversized payload shouldn't permanently halve throughput.
       const instance = createMockInstance({

@@ -323,6 +323,14 @@ export class PostHogTraces {
    * wait costs a request rather than the spans. The periodic flush does wait it
    * out.
    */
+  /**
+   * Whether the endpoint has asked this queue to wait. An automatic flush skips
+   * it; the traces flush timer already drains once the window closes.
+   */
+  get throttled(): boolean {
+    return this._retryAfter.isOpen()
+  }
+
   async flush(): Promise<void> {
     for (;;) {
       if (!this._queue.length) {
@@ -807,6 +815,10 @@ export class PostHogTraces {
     // Bounded by queue depth at flush start, so mid-drain arrivals ride the next flush.
     let remaining = this._queue.length
     let removed = 0
+    // Splits this drain only. A batch the SDK measured as too large says nothing
+    // about the ones after the oversized span is gone, so the cap kept between
+    // drains stays where it is and the next one starts at full size.
+    let localCap = Number.POSITIVE_INFINITY
     const generation = this._generation
 
     try {
@@ -823,7 +835,7 @@ export class PostHogTraces {
           this._headBatchFailures > 0
             ? Math.min(this._maxExportBatchSize, this._headBatchSize)
             : this._maxExportBatchSize
-        const size = Math.max(1, Math.min(cap, remaining, this._queue.length))
+        const size = Math.max(1, Math.min(cap, localCap, remaining, this._queue.length))
         const batch = this._queue.slice(0, size)
         const spans = this._encodeBatch(batch)
 
@@ -875,12 +887,17 @@ export class PostHogTraces {
             this._resetHeadBatchBudget()
             continue
           }
-          // Halve the batch the server rejected, not the configured maximum: when the
+          // Halve the batch that was refused, not the configured maximum: when the
           // queue is shallower than the maximum, shrinking it resends an identical body.
-          this._maxExportBatchSize = Math.max(1, Math.floor(size / 2))
+          const halved = Math.max(1, Math.floor(size / 2))
+          if (outcome.measuredLocally) {
+            localCap = halved
+          } else {
+            this._maxExportBatchSize = halved
+          }
           // A different batch from here on, so its budget starts fresh.
           this._resetHeadBatchBudget()
-          this._logger.debug(`Batch too large; retrying the same spans in batches of ${this._maxExportBatchSize}`)
+          this._logger.debug(`Batch too large; retrying the same spans in batches of ${halved}`)
           continue
         }
 

@@ -30,7 +30,15 @@ import type {
 export type SendTracesBatchOutcome =
   | { kind: 'ok' }
   | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
-  | { kind: 'too-large' }
+  | {
+      kind: 'too-large'
+      /**
+       * True when the SDK measured the body itself rather than the endpoint
+       * refusing it, so the caller can split this drain without lowering the
+       * batch size it keeps between them.
+       */
+      measuredLocally?: boolean
+    }
   | { kind: 'fatal'; error: unknown }
 
 /** The minimal host surface `PostHogTraces` depends on; `PostHogCoreStateless` satisfies it structurally. */

@@ -537,6 +537,59 @@ describe('PostHog traces', () => {
     })
   })
 
+  describe('Retry-After', () => {
+    const refuseTraces = (retryAfterSeconds: number): void => {
+      mockedFetch.mockImplementation(async (url: any) =>
+        String(url).includes('/i/v1/traces')
+          ? ({
+              status: 429,
+              headers: {
+                get: (name: string) => (name.toLowerCase() === 'retry-after' ? `${retryAfterSeconds}` : null),
+              },
+              text: async () => '',
+              json: async () => ({}),
+            } as any)
+          : ({ status: 200, headers: { get: () => null }, text: async () => '', json: async () => ({}) } as any)
+      )
+    }
+
+    it('does not drain spans on the events flush while the window is open', async () => {
+      const client = createClient({ flushAt: 1, flushInterval: 1000 })
+      refuseTraces(300)
+
+      client.startSpan('refused').end()
+      await flushTraces()
+      const afterRefusal = traceRequests().length
+      expect(afterRefusal).toBeGreaterThan(0)
+
+      // Events keep arriving, so the events timer keeps firing. Spans must not
+      // ride along on it while the endpoint has asked the traces queue to wait.
+      client.startSpan('queued-during-window').end()
+      for (let i = 0; i < 5; i++) {
+        client.capture({ distinctId: 'user', event: 'tick' })
+        await vi.advanceTimersByTimeAsync(1000)
+        await waitForPromises()
+      }
+
+      expect(traceRequests()).toHaveLength(afterRefusal)
+      await client.shutdown()
+    })
+
+    it('still drains spans on an explicit flush', async () => {
+      const client = createClient({ flushAt: 100, flushInterval: 60_000 })
+      refuseTraces(300)
+
+      client.startSpan('refused').end()
+      await flushTraces()
+      const afterRefusal = traceRequests().length
+
+      await client.flush()
+
+      expect(traceRequests().length).toBeGreaterThan(afterRefusal)
+      await client.shutdown()
+    })
+  })
+
   describe('beforeSpanSend', () => {
     it('scrubs attributes before they leave the process', async () => {
       const client = createClient({

@@ -294,9 +294,9 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
    * failed span export leaves the spans queued rather than rejecting, since
    * callers already treat `flush()` as safe to leave unwrapped.
    */
-  private _flushEventsAndSpans(): Promise<void> {
+  private _flushEventsAndSpans(skipThrottledSpans = false): Promise<void> {
     const events = this.flushWithPendingPromises()
-    if (!this._traces) {
+    if (!this._traces || (skipThrottledSpans && this._traces.throttled)) {
       return events
     }
     // Settled, not `all`: `all` rejects the moment the event flush does, and a
@@ -310,8 +310,18 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
     })
   }
 
+  protected override flushAutomatic(): Promise<void> {
+    // The events timer must not drag spans through a `Retry-After` window the
+    // traces queue is honouring; its own timer picks them up when it closes.
+    return this._flushKeepingRuntimeAlive(true)
+  }
+
   override async flush(): Promise<void> {
-    const flushPromise = this._flushEventsAndSpans()
+    return this._flushKeepingRuntimeAlive(false)
+  }
+
+  private _flushKeepingRuntimeAlive(skipThrottledSpans: boolean): Promise<void> {
+    const flushPromise = this._flushEventsAndSpans(skipThrottledSpans)
     const waitUntil = this.options.waitUntil
     // Only register when no debounce promise is already keeping runtime alive
     if (waitUntil && !this._waitUntilCycle) {