diff --git a/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts b/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts index 0ba60ae884..2126690ed3 100644 --- a/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts +++ b/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts @@ -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() }) diff --git a/packages/core/src/metrics/types.ts b/packages/core/src/metrics/types.ts index fb2685f421..d5bbc28668 100644 --- a/packages/core/src/metrics/types.ts +++ b/packages/core/src/metrics/types.ts @@ -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 } /** diff --git a/packages/core/src/posthog-core-stateless.ts b/packages/core/src/posthog-core-stateless.ts index 060c014eab..f35fe5ca67 100644 --- a/packages/core/src/posthog-core-stateless.ts +++ b/packages/core/src/posthog-core-stateless.ts @@ -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 { + return this.flush() + } + private async waitForPendingPromises( maxPromiseId: number, ignoredPromises: (Promise | 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 = diff --git a/packages/core/src/traces/index.spec.ts b/packages/core/src/traces/index.spec.ts index c9aefb36af..55bd0cfce1 100644 --- a/packages/core/src/traces/index.spec.ts +++ b/packages/core/src/traces/index.spec.ts @@ -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({ diff --git a/packages/core/src/traces/index.ts b/packages/core/src/traces/index.ts index 39d2928902..d82defaa7c 100644 --- a/packages/core/src/traces/index.ts +++ b/packages/core/src/traces/index.ts @@ -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 { 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 } diff --git a/packages/core/src/traces/types.ts b/packages/core/src/traces/types.ts index e992ae73c5..1f12eea7cf 100644 --- a/packages/core/src/traces/types.ts +++ b/packages/core/src/traces/types.ts @@ -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. */ diff --git a/packages/node/src/__tests__/traces.spec.ts b/packages/node/src/__tests__/traces.spec.ts index 964bc075ab..1e1b5c80e2 100644 --- a/packages/node/src/__tests__/traces.spec.ts +++ b/packages/node/src/__tests__/traces.spec.ts @@ -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({ diff --git a/packages/node/src/client.ts b/packages/node/src/client.ts index 92a9cd99c4..18f5becd14 100644 --- a/packages/node/src/client.ts +++ b/packages/node/src/client.ts @@ -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 { + private _flushEventsAndSpans(skipThrottledSpans = false): Promise { 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 { + // 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 { - const flushPromise = this._flushEventsAndSpans() + return this._flushKeepingRuntimeAlive(false) + } + + private _flushKeepingRuntimeAlive(skipThrottledSpans: boolean): Promise { + 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) {