diff --git a/.changeset/logs-backoff-cap.md b/.changeset/logs-backoff-cap.md
new file mode 100644
index 0000000000..0f1248b8b8
--- /dev/null
+++ b/.changeset/logs-backoff-cap.md
@@ -0,0 +1,7 @@
+---
+'posthog-js': patch
+'posthog-react-native': patch
+'@posthog/core': patch
+---
+
+Cap the retry delay for log exports at 30 seconds, the ceiling the logs contract states. It previously doubled to 64 times the flush interval — 192s on web, 640s on React Native — so a log export now resumes within 30 seconds of a failing endpoint recovering, at the cost of more retry requests while that endpoint is down.
diff --git a/.changeset/logs-backoff-survives-captures.md b/.changeset/logs-backoff-survives-captures.md
new file mode 100644
index 0000000000..7f68236c5b
--- /dev/null
+++ b/.changeset/logs-backoff-survives-captures.md
@@ -0,0 +1,7 @@
+---
+'posthog-js': patch
+'posthog-react-native': patch
+'@posthog/core': patch
+---
+
+Keep backing off a failing log flush while new records arrive, instead of the next record resetting the retry to the flush interval.
diff --git a/.changeset/otlp-honor-retry-after.md b/.changeset/otlp-honor-retry-after.md
new file mode 100644
index 0000000000..1ef2ab84a1
--- /dev/null
+++ b/.changeset/otlp-honor-retry-after.md
@@ -0,0 +1,7 @@
+---
+'posthog-node': patch
+'posthog-react-native': patch
+'@posthog/core': patch
+---
+
+Honor `Retry-After` when the ingestion endpoint refuses a logs or metrics batch, instead of retrying on the SDK's own schedule alone. A refusal naming a longer wait extends the one being served, up to five minutes from when it started. Retry delays now carry jitter so clients refused together do not return together, and metrics backs off exponentially across consecutive failures rather than retrying on a fixed interval.
diff --git a/.changeset/otlp-skip-oversized-bodies.md b/.changeset/otlp-skip-oversized-bodies.md
new file mode 100644
index 0000000000..85a0bd8a27
--- /dev/null
+++ b/.changeset/otlp-skip-oversized-bodies.md
@@ -0,0 +1,7 @@
+---
+'posthog-node': patch
+'posthog-react-native': patch
+'@posthog/core': patch
+---
+
+Stop sending logs and metrics batches over 10 MiB, or too large to serialize at all, instead of spending a request to discover the endpoint refuses them.
diff --git a/packages/browser/terser-mangled-names.json b/packages/browser/terser-mangled-names.json
index 8d68a22eaa..20ff9b2f45 100644
--- a/packages/browser/terser-mangled-names.json
+++ b/packages/browser/terser-mangled-names.json
@@ -113,7 +113,6 @@
         "_clearDebouncer",
         "_clearFlushBufferTimer",
         "_clearFlushTimeout",
-        "_clearFlushTimer",
         "_clearMouseSelection",
         "_clearPointer",
         "_clearSessionRegisteredProps",
@@ -235,6 +234,7 @@
         "_finishQueuedCompressionEvent",
         "_finishSetup",
         "_fireFeatureFlagsCallbacks",
+        "_firesAt",
         "_flagListenerCleanup",
         "_flagToExperiments",
         "_flagsLoadedFromRemote",
@@ -247,6 +247,7 @@
         "_flushInner",
         "_flushInterval",
         "_flushIntervalMs",
+        "_flushJitter",
         "_flushPendingActivityTimestamp",
         "_flushPromise",
         "_flushTimeout",
@@ -382,6 +383,7 @@
         "_initializeWidgetPromise",
         "_initialized",
         "_initializingClient",
+        "_installedAt",
         "_internalEventEmitter",
         "_internalFlagCheckSatisfied",
         "_intervalLogCount",
@@ -549,6 +551,7 @@
         "_onClick",
         "_onClickHandler",
         "_onDeadClick",
+        "_onFire",
         "_onFocusChange",
         "_onIdentityChanged",
         "_onIdentityCleared",
@@ -747,6 +750,7 @@
         "_resumeSavedTour",
         "_resyncIntervalMs",
         "_resyncTimer",
+        "_retryAfter",
         "_retryQueue",
         "_rrwebError",
         "_rrwebStartAttempted",
@@ -887,6 +891,7 @@
         "_teardown",
         "_throttledMutationsDropped",
         "_tickets",
+        "_timer",
         "_timestamp",
         "_totalBytes",
         "_touchStart",
@@ -916,6 +921,7 @@
         "_unsubscribeFeatureFlags",
         "_unsubscribeIdentifyListener",
         "_unsubscribeSessionId",
+        "_until",
         "_unwrapConsoleError",
         "_unwrapOnError",
         "_unwrapUnhandledRejection",
diff --git a/packages/core/src/__tests__/flush-timer.spec.ts b/packages/core/src/__tests__/flush-timer.spec.ts
new file mode 100644
index 0000000000..c21862b6ab
--- /dev/null
+++ b/packages/core/src/__tests__/flush-timer.spec.ts
@@ -0,0 +1,100 @@
+import { FlushTimer } from '../utils/flush-timer'
+
+describe('FlushTimer', () => {
+  it('fires once, after the delay', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(1000)
+
+    await vi.advanceTimersByTimeAsync(999)
+    expect(onFire).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(onFire).toHaveBeenCalledTimes(1)
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  it('releases the handle before firing, so the callback can arm again', async () => {
+    const timer: FlushTimer = new FlushTimer(() => {
+      pendingInsideCallback = timer.pending
+    })
+    let pendingInsideCallback: boolean | undefined
+    timer.arm(1000)
+    expect(timer.pending).toBe(true)
+
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(pendingInsideCallback).toBe(false)
+    expect(timer.pending).toBe(false)
+  })
+
+  it('arm replaces a pending timer outright, in either direction', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(10_000)
+    timer.arm(1000)
+
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  // The reason the deadline lives next to the handle: every capture reaches the
+  // arming path, and none of them may pull a flush in front of a longer wait.
+  it('armNoEarlierThan does not shorten a pending timer', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(10_000)
+    timer.armNoEarlierThan(1000)
+
+    await vi.advanceTimersByTimeAsync(9999)
+    expect(onFire).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(1)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  it('armNoEarlierThan lengthens a pending timer', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(1000)
+    timer.armNoEarlierThan(10_000)
+
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(onFire).not.toHaveBeenCalled()
+    await vi.advanceTimersByTimeAsync(9000)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  it('armNoEarlierThan counts down the remainder, not the whole delay', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(10_000)
+    await vi.advanceTimersByTimeAsync(6000)
+    // 4s left, so a 4s request is not longer and must not restart the wait.
+    timer.armNoEarlierThan(4000)
+
+    await vi.advanceTimersByTimeAsync(4000)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  it('armNoEarlierThan arms when nothing is pending', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.armNoEarlierThan(1000)
+    expect(timer.pending).toBe(true)
+
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(onFire).toHaveBeenCalledTimes(1)
+  })
+
+  it('clear stops a pending timer and is safe to repeat', async () => {
+    const onFire = vi.fn()
+    const timer = new FlushTimer(onFire)
+    timer.arm(1000)
+    timer.clear()
+    timer.clear()
+    expect(timer.pending).toBe(false)
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(onFire).not.toHaveBeenCalled()
+  })
+})
diff --git a/packages/core/src/__tests__/posthog.otlp-retry-after.spec.ts b/packages/core/src/__tests__/posthog.otlp-retry-after.spec.ts
new file mode 100644
index 0000000000..b2029831af
--- /dev/null
+++ b/packages/core/src/__tests__/posthog.otlp-retry-after.spec.ts
@@ -0,0 +1,108 @@
+import { createTestClient, PostHogCoreTestClient, PostHogCoreTestClientMocks } from '@/testing'
+
+// The header has to survive the whole path: response → PostHogFetchHttpError →
+// the retry-later outcome each queue reads. A unit test of the parser alone
+// would still pass if the plumbing were missing.
+describe('OTLP Retry-After', () => {
+  let posthog: PostHogCoreTestClient
+  let mocks: PostHogCoreTestClientMocks
+
+  const respondWith = (status: number, retryAfter?: string): void => {
+    mocks.fetch.mockResolvedValue({
+      status,
+      text: () => Promise.resolve(''),
+      json: () => Promise.resolve({}),
+      headers: { get: (name: string) => (name.toLowerCase() === 'retry-after' && retryAfter ? retryAfter : null) },
+    })
+  }
+
+  beforeEach(() => {
+    ;[posthog, mocks] = createTestClient('TEST_API_KEY', {
+      host: 'http://example.com',
+      preloadFeatureFlags: false,
+      disableCompression: true,
+      fetchRetryCount: 0,
+    })
+  })
+
+  it.each([
+    ['logs', () => posthog._sendLogsBatch({ resourceLogs: [] } as any)],
+    ['metrics', () => posthog._sendMetricsBatch({ resourceMetrics: [] } as any)],
+    ['traces', () => posthog._sendTracesBatch({ resourceSpans: [] } as any)],
+  ])("surfaces the endpoint's Retry-After to the %s queue", async (_signal, send) => {
+    respondWith(429, '120')
+
+    const outcome = await send()
+
+    expect(outcome).toMatchObject({ kind: 'retry-later', retryAfterMs: 120_000 })
+  })
+
+  it('sends once, not once per inner retry, when the endpoint names a wait', async () => {
+    // The inner retriable loop retries on a short fixed delay. Spending its
+    // attempts here would put three extra requests inside the very window the
+    // queue is about to back off for.
+    const [client, clientMocks] = createTestClient('TEST_API_KEY', {
+      host: 'http://example.com',
+      preloadFeatureFlags: false,
+      disableCompression: true,
+    })
+    clientMocks.fetch.mockResolvedValue({
+      status: 429,
+      text: () => Promise.resolve(''),
+      json: () => Promise.resolve({}),
+      headers: { get: (name: string) => (name.toLowerCase() === 'retry-after' ? '120' : null) },
+    })
+
+    const pending = client._sendTracesBatch({ resourceSpans: [] } as any)
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(await pending).toMatchObject({ kind: 'retry-later', retryAfterMs: 120_000 })
+    expect(clientMocks.fetch).toHaveBeenCalledTimes(1)
+  })
+
+  it('still retries internally when the response names no wait', async () => {
+    const [client, clientMocks] = createTestClient('TEST_API_KEY', {
+      host: 'http://example.com',
+      preloadFeatureFlags: false,
+      disableCompression: true,
+    })
+    clientMocks.fetch.mockResolvedValue({
+      status: 503,
+      text: () => Promise.resolve(''),
+      json: () => Promise.resolve({}),
+      headers: { get: () => null },
+    })
+
+    const pending = client._sendTracesBatch({ resourceSpans: [] } as any)
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect((await pending).kind).toBe('retry-later')
+    expect(clientMocks.fetch.mock.calls.length).toBeGreaterThan(1)
+  })
+
+  it('leaves retryAfterMs unset when the response sends no header', async () => {
+    respondWith(503)
+
+    const outcome = await posthog._sendTracesBatch({ resourceSpans: [] } as any)
+
+    expect(outcome.kind).toBe('retry-later')
+    expect((outcome as { retryAfterMs?: number }).retryAfterMs).toBeUndefined()
+  })
+
+  it('survives a transport whose headers accessor throws', async () => {
+    mocks.fetch.mockResolvedValue({
+      status: 503,
+      text: () => Promise.resolve(''),
+      json: () => Promise.resolve({}),
+      headers: {
+        get: () => {
+          throw new Error('hostile transport')
+        },
+      },
+    })
+
+    const outcome = await posthog._sendTracesBatch({ resourceSpans: [] } as any)
+
+    expect(outcome.kind).toBe('retry-later')
+  })
+})
diff --git a/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts b/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts
new file mode 100644
index 0000000000..2126690ed3
--- /dev/null
+++ b/packages/core/src/__tests__/posthog.otlp-too-large.spec.ts
@@ -0,0 +1,147 @@
+import { createTestClient, PostHogCoreTestClient, PostHogCoreTestClientMocks } from '@/testing'
+
+// The SDK will not put a body larger than the endpoint's configured limit on
+// the wire. Such a batch can only come back 413, so it is reported as too-large
+// without being sent — the caller's halving loop then isolates and drops the
+// oversized record without spending a request on every attempt.
+//
+// Mirrors OTLP_MAX_BODY_BYTES. Kept as a local literal so a change to the
+// shipped ceiling has to be made deliberately here too.
+const LIMIT_BYTES = 10 * 1024 * 1024
+const OVER_LIMIT_BYTES = LIMIT_BYTES + 1024
+
+describe('OTLP bodies over the endpoint limit', () => {
+  let posthog: PostHogCoreTestClient
+  let mocks: PostHogCoreTestClientMocks
+
+  const spansOf = (attributeBytes: number): any => ({
+    resourceSpans: [
+      {
+        resource: { attributes: [] },
+        scopeSpans: [
+          {
+            scope: { name: 'test' },
+            spans: [
+              { name: 'checkout', attributes: [{ key: 'blob', value: { stringValue: 'x'.repeat(attributeBytes) } }] },
+            ],
+          },
+        ],
+      },
+    ],
+  })
+
+  beforeEach(() => {
+    ;[posthog, mocks] = createTestClient('TEST_API_KEY', {
+      host: 'http://example.com',
+      preloadFeatureFlags: false,
+      disableCompression: true,
+      fetchRetryCount: 0,
+    })
+    mocks.fetch.mockResolvedValue({
+      status: 200,
+      text: () => Promise.resolve(''),
+      json: () => Promise.resolve({}),
+      headers: { get: () => null },
+    })
+  })
+
+  it.each([
+    ['traces', (payload: any) => posthog._sendTracesBatch(payload)],
+    ['logs', (payload: any) => posthog._sendLogsBatch(payload)],
+    ['metrics', (payload: any) => posthog._sendMetricsBatch(payload)],
+  ])('reports a %s batch it cannot serialize as too-large rather than throwing', async (_signal, send) => {
+    // A circular payload stands in for the reachable case: a batch past the
+    // runtime's max string length, which is too large to build in a test. Both
+    // leave `JSON.stringify` throwing, which unguarded escapes the tagged-outcome
+    // contract and leaves the caller retrying a batch it can never send.
+    const unserializable: any = spansOf(8)
+    unserializable.resourceSpans[0].scopeSpans[0].spans[0].self = unserializable
+
+    await expect(send(unserializable)).resolves.toEqual({ kind: 'too-large', measuredLocally: true })
+    expect(mocks.fetch).not.toHaveBeenCalled()
+  })
+
+  it('sends a batch inside the limit', async () => {
+    await expect(posthog._sendTracesBatch(spansOf(1024))).resolves.toEqual({ kind: 'ok' })
+    expect(mocks.fetch).toHaveBeenCalledTimes(1)
+  })
+
+  it.each([
+    ['traces', (payload: any) => posthog._sendTracesBatch(payload)],
+    ['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', measuredLocally: true })
+    expect(mocks.fetch).not.toHaveBeenCalled()
+  })
+
+  it('measures the payload, not the compressed body it is sent as', async () => {
+    // The endpoint decompresses before it measures, so a payload that gzips
+    // down to nothing is still refused on its decompressed size.
+    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',
+      measuredLocally: true,
+    })
+    expect(mocks.fetch).not.toHaveBeenCalled()
+  })
+
+  // The endpoint rejects a body *over* its limit, so one exactly at it is
+  // accepted on both sides. Pinning both directions keeps the comparison from
+  // drifting to `>=`, which would refuse an acceptable batch without a request
+  // and — in traces — halve it down and drop the span with no 413 to show for it.
+  const overheadBytes = (): number => JSON.stringify(spansOf(1024)).length - 1024
+
+  it('sends a batch of exactly the limit', async () => {
+    const exact = LIMIT_BYTES - overheadBytes()
+    expect(JSON.stringify(spansOf(exact)).length).toBe(LIMIT_BYTES)
+
+    await expect(posthog._sendTracesBatch(spansOf(exact))).resolves.toEqual({ kind: 'ok' })
+    expect(mocks.fetch).toHaveBeenCalledTimes(1)
+  })
+
+  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()
+  })
+
+  it('does not compress a batch it has already ruled out', async () => {
+    // Gzipping a multi-megabyte body only to throw it away is the whole cost of
+    // the attempt the size check exists to avoid, and the halving loops pay it
+    // again on every step down.
+    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',
+      measuredLocally: true,
+    })
+    expect(compressPayload).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    ['traces', (payload: any) => posthog._sendTracesBatch(payload)],
+    ['logs', (payload: any) => posthog._sendLogsBatch(payload)],
+    ['metrics', (payload: any) => posthog._sendMetricsBatch(payload)],
+  ])('sends a %s batch the service accepts but its fallback default would not', async (_signal, send) => {
+    // 3 MiB sits above the 2 MB the service falls back to and below the limit it
+    // is configured with, so it is accepted today. A ceiling set to the fallback
+    // would refuse it here — costing the records, since a batch of one that is
+    // refused for size is dropped, and metrics drops the whole window.
+    await expect(send(spansOf(3 * 1024 * 1024))).resolves.toEqual({ kind: 'ok' })
+    expect(mocks.fetch).toHaveBeenCalledTimes(1)
+  })
+
+  it('sends a compressible payload that is inside the limit before compression', async () => {
+    vi.spyOn(posthog as any, 'compressPayload').mockResolvedValue(new Uint8Array(1024))
+    ;(posthog as any).disableCompression = false
+
+    await expect(posthog._sendTracesBatch(spansOf(1024))).resolves.toEqual({ kind: 'ok' })
+    expect(mocks.fetch).toHaveBeenCalledTimes(1)
+  })
+})
diff --git a/packages/core/src/__tests__/retry-after.spec.ts b/packages/core/src/__tests__/retry-after.spec.ts
new file mode 100644
index 0000000000..8af1a12620
--- /dev/null
+++ b/packages/core/src/__tests__/retry-after.spec.ts
@@ -0,0 +1,191 @@
+import { MAX_RETRY_AFTER_MS, parseRetryAfterMs, RetryAfterWindow } from '../utils/retry-after'
+
+describe('parseRetryAfterMs', () => {
+  const now = Date.parse('2026-09-01T12:00:00Z')
+
+  it('reads delta-seconds', () => {
+    expect(parseRetryAfterMs('60', now)).toBe(60_000)
+    expect(parseRetryAfterMs('  7  ', now)).toBe(7_000)
+  })
+
+  it('reads an HTTP-date as a delay from now', () => {
+    expect(parseRetryAfterMs('Tue, 01 Sep 2026 12:00:30 GMT', now)).toBe(30_000)
+  })
+
+  it('reads the outermost hop when two proxies each append one', () => {
+    // `headers.get` joins repeated headers with ", ". The date form carries a
+    // comma of its own, so only the delta-seconds form is split.
+    expect(parseRetryAfterMs('60, 120', now)).toBe(60_000)
+    expect(parseRetryAfterMs('60,120', now)).toBe(60_000)
+    expect(parseRetryAfterMs('Tue, 01 Sep 2026 12:00:30 GMT', now)).toBe(30_000)
+  })
+
+  it('ignores a header that does not name a future time', () => {
+    // A past date and a zero delta both mean "retry now", which is the caller's own backoff.
+    expect(parseRetryAfterMs('Tue, 01 Sep 2026 11:59:00 GMT', now)).toBeUndefined()
+    expect(parseRetryAfterMs('0', now)).toBeUndefined()
+  })
+
+  it('ignores numeric junk rather than letting Date.parse read it as a year', () => {
+    // `Date.parse` reads all three as dates in 2001, so a device clock earlier
+    // than that would otherwise turn them into a real wait.
+    for (const clock of [now, Date.parse('1999-01-01T00:00:00Z')]) {
+      expect(parseRetryAfterMs('-5', clock)).toBeUndefined()
+      expect(parseRetryAfterMs('+5', clock)).toBeUndefined()
+      expect(parseRetryAfterMs('5.5', clock)).toBeUndefined()
+    }
+  })
+
+  it('ignores a header it cannot parse rather than guessing', () => {
+    // "10 minutes" must not read as 10 seconds.
+    expect(parseRetryAfterMs('10 minutes', now)).toBeUndefined()
+    expect(parseRetryAfterMs('soon', now)).toBeUndefined()
+    expect(parseRetryAfterMs('', now)).toBeUndefined()
+    expect(parseRetryAfterMs(null, now)).toBeUndefined()
+    expect(parseRetryAfterMs(undefined, now)).toBeUndefined()
+  })
+
+  it('caps an unbounded value so a bogus header cannot strand a queue', () => {
+    expect(parseRetryAfterMs('86400', now)).toBe(5 * 60_000)
+  })
+
+  it('ignores a value that is not a string', () => {
+    // The header comes from injected transport code, so `headers.get` can hand
+    // back anything at all.
+    expect(parseRetryAfterMs(60, now)).toBeUndefined()
+    expect(parseRetryAfterMs(['60'], now)).toBeUndefined()
+    expect(parseRetryAfterMs({}, now)).toBeUndefined()
+  })
+})
+
+describe('RetryAfterWindow', () => {
+  const open = (retryAfterMs: number): RetryAfterWindow => {
+    const window = new RetryAfterWindow()
+    window.record({ kind: 'retry-later', retryAfterMs })
+    return window
+  }
+
+  beforeEach(() => {
+    vi.useFakeTimers()
+    vi.setSystemTime(Date.parse('2026-09-01T12:00:00Z'))
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('counts down to the deadline the endpoint named', () => {
+    const window = open(60_000)
+    expect(window.remainingMs()).toBe(60_000)
+
+    vi.setSystemTime(Date.now() + 20_000)
+    expect(window.remainingMs()).toBe(40_000)
+    expect(window.isOpen()).toBe(true)
+  })
+
+  it('does not bring back a window that already ran out', () => {
+    // The deadline is wall clock, so a backward step after the wait was served
+    // would otherwise make the queue wait it out a second time.
+    const window = open(60_000)
+    const installedAt = Date.now()
+
+    vi.setSystemTime(installedAt + 600_000)
+    expect(window.remainingMs()).toBe(0)
+
+    vi.setSystemTime(installedAt + 30_000)
+    expect(window.remainingMs()).toBe(0)
+    expect(window.isOpen()).toBe(false)
+  })
+
+  it('survives a clock correction of a few milliseconds', () => {
+    // NTP nudges the clock backwards by milliseconds routinely; discarding a
+    // five-minute wait over one would send straight back into the rate limit.
+    const window = open(300_000)
+
+    vi.setSystemTime(Date.now() - 1)
+    expect(window.isOpen()).toBe(true)
+  })
+
+  it('discards the window on a real backward clock step', () => {
+    const window = open(300_000)
+
+    vi.setSystemTime(Date.now() - 3_600_000)
+    expect(window.remainingMs()).toBe(0)
+  })
+
+  it('extends an open window when a refusal names a longer wait', () => {
+    const window = open(60_000)
+    vi.setSystemTime(Date.now() + 20_000)
+
+    window.record({ kind: 'retry-later', retryAfterMs: 60_000 })
+    expect(window.remainingMs()).toBe(60_000)
+  })
+
+  it('does not pull an open window in when a refusal names a shorter wait', () => {
+    // A shorter header must not cut a wait the endpoint has already asked for.
+    const window = open(120_000)
+    vi.setSystemTime(Date.now() + 10_000)
+
+    window.record({ kind: 'retry-later', retryAfterMs: 5_000 })
+    expect(window.remainingMs()).toBe(110_000)
+  })
+
+  it('caps an extension at five minutes from where the window was installed', () => {
+    // The SDK's own ceiling, not OTLP's: without it a host refused faster than
+    // the window is long would refresh the deadline indefinitely.
+    const window = open(240_000)
+    vi.setSystemTime(Date.now() + 200_000)
+
+    window.record({ kind: 'retry-later', retryAfterMs: 240_000 })
+
+    // Asked for 440_000 from the install; capped at 300_000, of which 200_000
+    // has been served.
+    expect(window.remainingMs()).toBe(100_000)
+  })
+
+  it('keeps the window when a batch is refused for size', () => {
+    // `too-large` is a verdict on the body's size — the SDK's own or a 413 — so
+    // it carries nothing about the endpoint's rate limit.
+    const window = open(60_000)
+
+    window.record({ kind: 'too-large' })
+    expect(window.remainingMs()).toBe(60_000)
+  })
+
+  it('keeps the window when a retriable failure carries no header', () => {
+    // A network error, a timeout and a header-less 503 are what the outage that
+    // named the wait keeps producing; none of them revokes it.
+    const window = open(300_000)
+    vi.setSystemTime(Date.now() + 10_000)
+
+    window.record({ kind: 'retry-later' })
+    expect(window.remainingMs()).toBe(290_000)
+  })
+
+  it('installs a wait that arrives after the previous one has elapsed', () => {
+    // A send can outlive the window it was made under, and the refusal it comes
+    // back with names a deadline the endpoint still expects to be honored.
+    const window = open(60_000)
+    vi.setSystemTime(Date.now() + 60_000)
+
+    window.record({ kind: 'retry-later', retryAfterMs: 300_000 })
+    expect(window.remainingMs()).toBe(300_000)
+  })
+
+  it('caps a wait longer than the maximum when it is installed', () => {
+    // `retryAfterMs` reaches the window from the exported host interfaces and
+    // export outcomes, so it is not always a value the SDK parsed itself.
+    const window = open(60 * 60_000)
+
+    vi.setSystemTime(Date.now() + MAX_RETRY_AFTER_MS)
+    expect(window.isOpen()).toBe(false)
+  })
+
+  it('ends the window on an outcome that is not a retry', () => {
+    for (const outcome of [{ kind: 'ok' } as const, { kind: 'fatal' } as const]) {
+      const window = open(60_000)
+      window.record(outcome)
+      expect(window.isOpen()).toBe(false)
+    }
+  })
+})
diff --git a/packages/core/src/logs/index.spec.ts b/packages/core/src/logs/index.spec.ts
index 1ed7f2ae7b..4ed6b5542b 100644
--- a/packages/core/src/logs/index.spec.ts
+++ b/packages/core/src/logs/index.spec.ts
@@ -164,6 +164,9 @@ describe('PostHogLogs', () => {
   beforeEach(() => {
     mockInstance = createMockInstance()
     logger = createMockLogger()
+    // Retry delays carry jitter; pinned to its midpoint so every timing
+    // assertion here measures the backoff itself and cannot flake.
+    vi.spyOn(Math, 'random').mockReturnValue(0.5)
   })
 
   it('constructs without throwing', () => {
@@ -927,7 +930,7 @@ describe('PostHogLogs', () => {
       await logs.flush()
 
       expect(logger.warn).toHaveBeenCalledWith(
-        expect.stringContaining('Dropping a single log record after 413 with batch size 1')
+        expect.stringContaining('Dropping a single log record with batch size 1')
       )
     })
 
@@ -985,7 +988,7 @@ describe('PostHogLogs', () => {
       expect(sendSizes).toEqual([1])
       expect(readQueue(mockInstance)).toHaveLength(0)
       expect(logger.warn).toHaveBeenCalledWith(
-        expect.stringContaining('Dropping a single log record after 413 with batch size 1')
+        expect.stringContaining('Dropping a single log record with batch size 1')
       )
     })
 
@@ -1166,6 +1169,468 @@ describe('PostHogLogs', () => {
       expect(readQueue(mockInstance)).toHaveLength(0)
     })
 
+    it('waits out Retry-After even when a capture re-arms the timer mid-flush', async () => {
+      // The capture that lands while the send is in flight arms a timer at the
+      // plain interval; the 429 then asks for far longer. The earlier timer must
+      // not fire first, or the SDK sends inside the window it was told to skip.
+      mockInstance._sendLogsBatch = vi.fn(async () => {
+        logs.captureLog({ body: 'arrived mid-flush' })
+        return { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }
+      })
+
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(6000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not let the size trigger send inside a Retry-After window', async () => {
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000, maxBufferSize: 2 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // Enough records to trip the size trigger, well inside the window.
+      logs.captureLog({ body: 'second' })
+      logs.captureLog({ body: 'third' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not let onReconnect send inside a Retry-After window', async () => {
+      // `online` fires on every network handover; it says nothing about the
+      // rate-limit window the endpoint set.
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+    })
+
+    it('flushes on reconnect once the Retry-After window has passed', async () => {
+      const outcomes: any[] = [{ kind: 'retry-later', error: new Error('429'), retryAfterMs: 5000 }]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // Stop just short of the deadline, then cross it without letting the
+      // re-armed timer fire — otherwise the timer satisfies the assertion and
+      // the test says nothing about onReconnect.
+      await vi.advanceTimersByTimeAsync(4999)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+      vi.setSystemTime(Date.now() + 2)
+
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not let a capture after an explicit flush send inside the window', async () => {
+      // `flush()` is the lifecycle path (RN foreground/background, shutdown).
+      // It leaves no timer behind, so the next capture is the one that arms
+      // one — at the plain interval unless the window floors it.
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await logs.flush().catch(() => {})
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      logs.captureLog({ body: 'second' })
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(295_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not let a capture during an explicit flush send inside the window', async () => {
+      // The sibling case covers a capture *after* the flush settles. This one
+      // lands while the send is in flight, so it arms the timer at the plain
+      // interval before the window exists.
+      let release: (v: any) => void = () => {}
+      mockInstance._sendLogsBatch = vi.fn(
+        () =>
+          new Promise((resolve) => {
+            release = resolve
+          })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      const flushed = logs.flush().catch(() => {})
+      await vi.advanceTimersByTimeAsync(0)
+
+      logs.captureLog({ body: 'second' })
+      release({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      await flushed
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+    })
+
+    it('arms a timer when onReconnect lands inside the window after an explicit flush', async () => {
+      // `flush()` leaves no timer behind, so returning early here without
+      // arming one leaves the records with nothing scheduled at all.
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 10_000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await logs.flush().catch(() => {})
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      logs.onReconnect()
+
+      await vi.advanceTimersByTimeAsync(289_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+      await vi.advanceTimersByTimeAsync(2000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('releases a record captured mid-flush once that flush closes the window', async () => {
+      // The capture arms against the window that was open when it landed; the
+      // outcome then closes that window, so the timer has to come back down.
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 10_000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      let sends = 0
+      mockInstance._sendLogsBatch = vi.fn(async () => {
+        sends += 1
+        if (sends === 1) {
+          return { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }
+        }
+        await Promise.resolve()
+        if (sends === 2) {
+          logs.captureLog({ body: 'mid' })
+        }
+        return { kind: 'ok' }
+      })
+
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(10_000)
+      await logs.flush().catch(() => {})
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('keeps the Retry-After window when a batch is refused for size', async () => {
+      // `too-large` is a verdict on the body's size — the SDK's own or a 413 —
+      // so it says nothing about the endpoint's rate limit and must not end the
+      // wait.
+      const outcomes: any[] = [{ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'too-large' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000, maxBatchRecordsPerPost: 1 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // A batch of one the endpoint cannot accept: the record is dropped.
+      await logs.flush()
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+
+      logs.captureLog({ body: 'second' })
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('ends the window when an explicit flush succeeds', async () => {
+      // The endpoint just accepted a batch, so the wait it asked for earlier is
+      // over — the gated paths must not stay blocked for the rest of it.
+      const outcomes: any[] = [{ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000, maxBufferSize: 2 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await logs.flush()
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+
+      logs.captureLog({ body: 'second' })
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('drops a Retry-After wait on reset', async () => {
+      const outcomes: any[] = [{ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // Asserted through a gated path: a plain capture would flush either way.
+      logs.reset()
+      logs.captureLog({ body: 'second' })
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('keeps its own backoff when the endpoint asks for less', async () => {
+      // A proxy answering `Retry-After: 1` must not turn the retry into a
+      // one-second hot loop against an endpoint already refusing traffic.
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('503'), retryAfterMs: 10 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not restart a served-out wait for a capture during the retry', async () => {
+      // A deadline, not a duration. The retry's timer has already fired, so the
+      // capture below arms the next one — and it sees the wait still set,
+      // because the send it belongs to has not settled. Holding a duration here
+      // re-arms for the whole window again and leaves the record 300s behind an
+      // endpoint that has already recovered.
+      let settle: ((outcome: any) => void) | undefined
+      let call = 0
+      mockInstance._sendLogsBatch = vi.fn(() => {
+        call++
+        if (call === 1) {
+          return Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+        }
+        return new Promise((resolve) => {
+          settle = resolve
+        })
+      })
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // The wait elapses and the retry goes out, but hangs.
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+
+      logs.captureLog({ body: 'second' })
+      settle?.({ kind: 'ok' })
+      await vi.advanceTimersByTimeAsync(0)
+
+      // One interval, not another window.
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('closes the window at the ceiling for a host out-pacing it', async () => {
+      // RN takes flush() on every app-state transition, and each refusal pushes
+      // the deadline out. The ceiling is what stops the gated paths — the size
+      // trigger and onReconnect — from being suppressed for good.
+      mockInstance._sendLogsBatch = vi.fn(() =>
+        Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 30_000 })
+      )
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 60_000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await logs.flush().catch(() => {})
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      // Lifecycle flushes every 5s against a 30s window, for longer than the
+      // 5-minute ceiling. Sampled rather than asserted through onReconnect:
+      // whether a given moment falls inside a window is timing-dependent, but
+      // past the ceiling it must fall outside one.
+      let sawWindowClosed = false
+      for (let i = 0; i < 70; i++) {
+        await vi.advanceTimersByTimeAsync(5000)
+        // Sampled before the flush: a flush that finds the window closed opens
+        // a fresh one, so sampling after it would always look open.
+        if ((logs as any)._retryAfter.remainingMs() === 0) {
+          sawWindowClosed = true
+        }
+        logs.captureLog({ body: `line ${i}` })
+        await logs.flush().catch(() => {})
+      }
+      expect(sawWindowClosed).toBe(true)
+    })
+
+    it('keeps flushing after a backward clock step', async () => {
+      const outcomes: any[] = [{ kind: 'retry-later', error: new Error('429'), retryAfterMs: 60_000 }]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000, maxBufferSize: 2 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      const real = Date.now()
+      vi.spyOn(Date, 'now').mockImplementation(() => real - 3_600_000)
+
+      // A gated path: suppressed for the size of the step without the guard.
+      logs.captureLog({ body: 'second' })
+      logs.onReconnect()
+      await vi.advanceTimersByTimeAsync(1)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('ends the wait when a later failure names none', async () => {
+      // 429 with a window, then a plain 503: the queue drops back to its own
+      // backoff rather than waiting the old window out on every attempt.
+      const outcomes: any[] = [
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+        { kind: 'retry-later', error: new Error('503') },
+      ]
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 1000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'first' })
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(2)
+
+      // Back on the plain backoff, not another 300s.
+      await vi.advanceTimersByTimeAsync(4000)
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('keeps flushing on the interval while captures keep arriving', async () => {
+      // Every capture arms the timer. Re-arming a pending one would push the
+      // flush out for as long as logs keep coming, stranding a steady stream
+      // that never reaches the size trigger.
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 10_000, maxBufferSize: 100 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+
+      for (let i = 0; i < 30; i++) {
+        logs.captureLog({ body: `line ${i}` })
+        await vi.advanceTimersByTimeAsync(2000)
+      }
+
+      expect(mockInstance._sendLogsBatch).toHaveBeenCalled()
+      expect(readQueue(mockInstance)).toHaveLength(0)
+    })
+
     it('stops re-arming once the queue is empty', async () => {
       const logs = new PostHogLogs(
         mockInstance,
@@ -1185,6 +1650,29 @@ describe('PostHogLogs', () => {
       expect(mockInstance._sendLogsBatch).toHaveBeenCalledTimes(1)
     })
 
+    it('caps the retry delay at 30s however long the outage runs', async () => {
+      // The logs contract states the backoff as capped at ~30s. Without the cap
+      // a 5s interval reaches 320s after six doublings.
+      mockInstance._sendLogsBatch = vi.fn(() => Promise.resolve({ kind: 'retry-later', error: new Error('down') }))
+      const logs = new PostHogLogs(
+        mockInstance,
+        resolveForTest({ flushIntervalMs: 5000 }),
+        logger,
+        getContextFor(mockInstance),
+        immediateOnReady
+      )
+      logs.captureLog({ body: 'retry-me' })
+
+      // Ten failures, well past the six doublings the exponent allows.
+      for (let i = 0; i < 10; i++) {
+        await vi.advanceTimersByTimeAsync(30_000)
+      }
+      // Capped, the delay settles at 30s and 300s of outage buys ten retries on
+      // top of the two the first doublings allow. Uncapped it reaches 320s and
+      // buys six attempts in total, so the difference is not a rounding one.
+      expect(mockInstance._sendLogsBatch.mock.calls.length).toBeGreaterThanOrEqual(11)
+    })
+
     it('backs off exponentially across consecutive failed flushes', async () => {
       // Every flush fails, so the record stays queued and the retry interval grows:
       // base (initial), base (1st retry), 2x, 4x, ...
diff --git a/packages/core/src/logs/index.ts b/packages/core/src/logs/index.ts
index 06a7ca9b89..dd9f3e6cac 100644
--- a/packages/core/src/logs/index.ts
+++ b/packages/core/src/logs/index.ts
@@ -1,11 +1,13 @@
 import type { LogAttributeValue } from '@posthog/types'
 import { buildOtlpLogRecord, buildOtlpLogsPayload, buildResourceAttributes } from './logs-utils'
 import { Logger, PostHogPersistedProperty } from '../types'
-import { isArray, raceWithTimeout, safeSetTimeout } from '../utils'
+import { isArray, raceWithTimeout } from '../utils'
+import { FlushTimer } from '../utils/flush-timer'
+import { RetryAfterWindow } from '../utils/retry-after'
+import { MAX_FLUSH_BACKOFF_MS, NO_JITTER, backoffDelayMs, drawJitter } from '../utils/backoff'
 import type { BufferedLogEntry, CaptureLogOptions, LogSdkContext, LogsHost, ResolvedPostHogLogsConfig } from './types'
 
 // Caps the retry backoff at 2^6 = 64× the flush interval.
-const MAX_FLUSH_BACKOFF_EXPONENT = 6
 
 export class PostHogLogs {
   private _maxBufferSize: number
@@ -17,7 +19,7 @@ export class PostHogLogs {
   // one record after each successful send so a one-off oversized payload
   // (e.g. a giant stack trace) doesn't permanently degrade throughput.
   private _maxBatchRecordsPerPost: number
-  private _flushTimer?: ReturnType<typeof safeSetTimeout>
+  private readonly _flushTimer = new FlushTimer(() => this._flushInBackground())
   // Serializes concurrent flushes — the second caller awaits the first rather
   // than racing it and double-sending the same head-of-queue records.
   private _flushPromise: Promise<void> | null = null
@@ -27,9 +29,12 @@ export class PostHogLogs {
   // A batch captures this when it is assembled, so it can tell that the records it is
   // holding no longer correspond to anything queued.
   private _queueGeneration = 0
+  // Every path that can start a send checks this, not just the retry timer.
+  private _retryAfter = new RetryAfterWindow()
   // Consecutive failed flushes; drives exponential backoff on the retry timer.
   // A successful flush resets it to 0.
   private _consecutiveFlushFailures = 0
+  private _flushJitter = NO_JITTER
 
   // Fixed-window rate cap. Tumbling (not sliding) for cheap arithmetic on the
   // hot path. Window rolls the first time `captureLog` fires after the window
@@ -83,7 +88,7 @@ export class PostHogLogs {
    * and clears it separately (the browser empties its in-memory store).
    */
   reset(): void {
-    this._clearFlushTimer()
+    this._flushTimer.clear()
     // `_flushPromise` is deliberately left alone: clearing it would let a second flush
     // run alongside the in-flight one. Retiring that flush is `clearQueue`'s job,
     // because the queue it holds belongs to the host, not to this state.
@@ -91,14 +96,28 @@ export class PostHogLogs {
     this._intervalLogCount = 0
     this._droppedWarned = false
     this._consecutiveFlushFailures = 0
+    this._flushJitter = NO_JITTER
+    this._retryAfter.reset()
     this._maxBatchRecordsPerPost = this._config.maxBatchRecordsPerPost
   }
 
   // Call when connectivity is restored: clear the failure backoff and flush now,
   // so records don't wait out a (possibly minutes-long) backoff delay after the
   // network returns. The host owns connectivity detection (web: `online` event).
+  // A `Retry-After` window survives this: the network coming back says nothing
+  // about the rate limit the endpoint set, and browsers fire `online` on every
+  // network handover.
   onReconnect(): void {
     this._consecutiveFlushFailures = 0
+    this._flushJitter = NO_JITTER
+    if (this._retryAfter.isOpen()) {
+      // The wait outlives the reconnect, but something still has to schedule
+      // the retry: an explicit `flush()` leaves no timer behind.
+      if (this._hasQueuedRecords()) {
+        this._armFlushTimer()
+      }
+      return
+    }
     this._flushInBackground()
   }
 
@@ -248,7 +267,7 @@ export class PostHogLogs {
   }
 
   private async _flushInner(): Promise<void> {
-    this._clearFlushTimer()
+    this._flushTimer.clear()
 
     let queue = this._instance.getPersistedProperty<BufferedLogEntry[]>(PostHogPersistedProperty.LogsQueue) ?? []
     if (queue.length === 0) {
@@ -292,12 +311,22 @@ export class PostHogLogs {
       if (outcome.kind === 'too-large' && batch.length > 1) {
         this._maxBatchRecordsPerPost = Math.max(1, Math.floor(batch.length / 2))
         this._logger.warn(
-          `Received 413 when sending logs batch of size ${batch.length}, reducing batch size to ${this._maxBatchRecordsPerPost}`
+          `Logs batch of size ${batch.length} was too large for the ingestion endpoint, reducing batch size to ${this._maxBatchRecordsPerPost}`
         )
         // Don't advance the queue — retry the same records with the smaller cap.
         continue
       }
 
+      // Not on the background wrapper: every lifecycle hook takes `flush()`,
+      // which does not go through it.
+      this._retryAfter.record(outcome)
+
+      // Outright, not through the ratchet: a timer a mid-flight capture armed
+      // is measured against a window this outcome may just have closed.
+      if (this._flushTimer.pending) {
+        this._flushTimer.arm(Math.max(this._flushIntervalMs, this._retryAfter.remainingMs()))
+      }
+
       if (outcome.kind === 'retry-later') {
         // Transient failure: keep records in the queue for the next flush cycle
         // and surface the error so the caller can log/react.
@@ -305,13 +334,15 @@ export class PostHogLogs {
       }
 
       // ok | fatal | too-large-with-batch-of-1 → records are leaving the
-      // queue. 'fatal' and size-1 413s are dropped so we don't spin on the
-      // same record forever. Surface the size-1 413 explicitly so a single
+      // queue. 'fatal' and size-1 refusals are dropped so we don't spin on the
+      // same record forever. Surface the size-1 refusal explicitly so a single
       // oversized record (e.g. a giant body field) is visible in logs
       // instead of silently disappearing.
       if (outcome.kind === 'too-large') {
+        // Reached either from a 413 or from the size the SDK measured before
+        // sending, so the message names neither.
         this._logger.warn(
-          'Dropping a single log record after 413 with batch size 1 — the record is larger than the server cap and cannot be split further.'
+          'Dropping a single log record with batch size 1 — the record is larger than the server cap and cannot be split further.'
         )
       } else if (outcome.kind === 'ok' && this._maxBatchRecordsPerPost < this._config.maxBatchRecordsPerPost) {
         // Linear recovery: each healthy send pushes the cap back up by 1
@@ -364,34 +395,40 @@ export class PostHogLogs {
     this._instance.setPersistedProperty(PostHogPersistedProperty.LogsQueue, queue)
 
     // Flush trigger: drain now rather than waiting for the timer. The queue may
-    // grow past this up to the eviction cap while the flush is in flight.
-    if (queue.length >= this._maxBufferSize) {
+    // grow past this up to the eviction cap while the flush is in flight. Not
+    // while the endpoint has asked us to wait: the size trigger is the dominant
+    // one on a busy host, so sending here would ignore the window entirely.
+    if (queue.length >= this._maxBufferSize && !this._retryAfter.isOpen()) {
       this._flushInBackground()
       return
     }
 
-    // Arm one timer at a time; re-arming within the window would push the flush out.
+    // Arm one timer at a time; re-arming on every enqueue would push the flush out.
     this._armFlushTimer()
   }
 
   // Arms the flush timer if none is pending. One-shot: the callback clears the
   // handle so the next enqueue (or a flush that left records) schedules again.
-  private _armFlushTimer(delayMs: number = this._flushIntervalMs): void {
-    if (this._flushTimer) {
+  private _armFlushTimer(): void {
+    if (this._flushTimer.pending) {
       return
     }
-    this._flushTimer = safeSetTimeout(() => {
-      this._flushTimer = undefined
-      this._flushInBackground()
-    }, delayMs)
+    // Floored by any open window: `flush()` leaves no timer behind, so a
+    // capture after one arrives here.
+    this._flushTimer.arm(Math.max(this._flushIntervalMs, this._retryAfter.remainingMs()))
   }
 
   // Retry delay after a flush that left records: the first retry is at the base
-  // interval, then exponential backoff (capped) so a sustained outage isn't
-  // retried every interval.
+  // interval, then exponential backoff so a sustained outage isn't retried every
+  // interval. Jitter is drawn once per failure and reused, so two delays taken
+  // for the same failure cannot disagree.
   private _nextFlushDelay(): number {
-    const exponent = Math.min(Math.max(0, this._consecutiveFlushFailures - 1), MAX_FLUSH_BACKOFF_EXPONENT)
-    return this._flushIntervalMs * 2 ** exponent
+    // A floor, not a replacement: the header never retries us sooner than our
+    // own backoff would have.
+    return Math.max(
+      backoffDelayMs(this._flushIntervalMs, this._consecutiveFlushFailures, this._flushJitter, MAX_FLUSH_BACKOFF_MS),
+      this._retryAfter.remainingMs()
+    )
   }
 
   private _hasQueuedRecords(): boolean {
@@ -410,7 +447,7 @@ export class PostHogLogs {
    * fetchRetryDelay)`, which can exceed the caller's shutdown SLA.
    */
   async shutdown(timeoutMs?: number): Promise<void> {
-    this._clearFlushTimer()
+    this._flushTimer.clear()
     const flushPromise = this.flush().catch(() => {
       // Best-effort: a logs-flush failure during shutdown is not actionable
       // and must not prevent the rest of shutdown from running. Errors are
@@ -449,9 +486,11 @@ export class PostHogLogs {
       .then(
         () => {
           this._consecutiveFlushFailures = 0
+          this._flushJitter = NO_JITTER
         },
         (err) => {
           this._consecutiveFlushFailures++
+          this._flushJitter = drawJitter()
           this._logger.error('PostHog logs flush failed:', err)
         }
       )
@@ -460,15 +499,8 @@ export class PostHogLogs {
         // sit undelivered on a quiet page; re-arm so the timer retries them, backing
         // off on consecutive failures.
         if (!this._instance.isDisabled && this._hasQueuedRecords()) {
-          this._armFlushTimer(this._nextFlushDelay())
+          this._flushTimer.armNoEarlierThan(this._nextFlushDelay())
         }
       })
   }
-
-  private _clearFlushTimer(): void {
-    if (this._flushTimer) {
-      clearTimeout(this._flushTimer)
-      this._flushTimer = undefined
-    }
-  }
 }
diff --git a/packages/core/src/logs/types.ts b/packages/core/src/logs/types.ts
index 19c60ce5c8..93885fdbd7 100644
--- a/packages/core/src/logs/types.ts
+++ b/packages/core/src/logs/types.ts
@@ -128,8 +128,8 @@ export interface PostHogLogsConfig {
 
   /**
    * Max records per outbound POST. Keeps each request under the server's
-   * 2 MB cap. On a 413 response, the SDK halves this value, retries the
-   * same records, then ramps back up by 1 per healthy send. A 413 on a
+   * request body cap. On a 413 response, the SDK halves this value, retries
+   * the same records, then ramps back up by 1 per healthy send. A 413 on a
    * single-record batch drops the record (it's larger than the server can
    * accept regardless of batch size). Default: 50 (RN) / 100 (browser).
    */
diff --git a/packages/core/src/metrics/index.spec.ts b/packages/core/src/metrics/index.spec.ts
index 7b248b4c93..feabff47b3 100644
--- a/packages/core/src/metrics/index.spec.ts
+++ b/packages/core/src/metrics/index.spec.ts
@@ -50,6 +50,9 @@ describe('PostHogMetrics', () => {
     vi.useFakeTimers()
     mockInstance = createMockInstance()
     logger = createMockLogger()
+    // Retry delays carry jitter; pinned to its midpoint so every timing
+    // assertion here measures the backoff itself and cannot flake.
+    vi.spyOn(Math, 'random').mockReturnValue(0.5)
   })
 
   afterEach(() => {
@@ -243,6 +246,263 @@ describe('PostHogMetrics', () => {
       expect(mockInstance._sendMetricsBatch).toHaveBeenCalledTimes(1)
     })
 
+    it('keeps flushing on the interval while captures keep arriving', async () => {
+      // Every capture arms the timer, and metrics have no size trigger to fall
+      // back on: re-arming a pending one would stop the window ever being sent.
+      const metrics = createMetrics({ flushIntervalMs: 10_000 })
+
+      for (let i = 0; i < 60; i++) {
+        metrics.count('orders_created', 1)
+        await vi.advanceTimersByTimeAsync(1000)
+      }
+
+      expect(mockInstance._sendMetricsBatch).toHaveBeenCalled()
+    })
+
+    it('waits out Retry-After even when a flush timer is already pending', async () => {
+      // The capture arms a timer at the flush interval; the failed flush then asks
+      // for far longer. The pending timer must not fire first.
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(11_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('clears Retry-After on an outcome that is not a retry', async () => {
+      // Only a retriable outcome carries a wait. A stale one left set here would
+      // pin every later flush at the server's old window forever.
+      const outcomes: SendMetricsBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+        { kind: 'fatal', error: new Error('400') },
+      ]
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' })),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+
+      // The wait elapses, the retry lands a 400, and that ends the wait.
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('keeps the Retry-After window when a batch is refused for size', async () => {
+      // `too-large` is a verdict on the body's size — the SDK's own or a 413 —
+      // so it says nothing about the endpoint's rate limit. Ending the wait on
+      // it lets the next refusal install a fresh window, pushing the retry out
+      // past the deadline the endpoint actually named — observed here as the
+      // flush landing at 300s rather than at 310s.
+      const outcomes: SendMetricsBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+        { kind: 'too-large' },
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+      ]
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' })),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 1000 }, instance)
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      await metrics.flush()
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+
+      await vi.advanceTimersByTimeAsync(295_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(4)
+    })
+
+    it('closes the window at the ceiling for a host out-pacing it', async () => {
+      // Each refusal sliding the deadline would keep `_nextFlushDelay` pinned at
+      // the full window, so the flush cadence would never recover.
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 30_000 })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 60_000 }, instance)
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      // Sampled: whether a given moment falls inside a window is timing
+      // dependent, but it must fall outside one sometimes.
+      let sawWindowClosed = false
+      for (let i = 0; i < 70; i++) {
+        await vi.advanceTimersByTimeAsync(5000)
+        // Sampled before the flush: a flush that finds the window closed opens
+        // a fresh one, so sampling after it would always look open.
+        if ((metrics as any)._retryAfter.remainingMs() === 0) {
+          sawWindowClosed = true
+        }
+        metrics.count('orders_created', 1)
+        await metrics.flush()
+      }
+      expect(sawWindowClosed).toBe(true)
+    })
+
+    it('does not install a Retry-After that lands after reset', async () => {
+      let settle: ((outcome: SendMetricsBatchOutcome) => void) | undefined
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn(
+          (): Promise<SendMetricsBatchOutcome> =>
+            new Promise((resolve) => {
+              settle = resolve
+            })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 1000 }, instance)
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      metrics.reset()
+
+      // Settle first: the capture that arms the next timer must not find a
+      // window belonging to the client that was just torn down.
+      settle?.({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      await vi.advanceTimersByTimeAsync(0)
+
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('drops a Retry-After wait on reset', async () => {
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+
+      metrics.reset()
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not restart a served-out wait for a sample captured mid-retry', async () => {
+      // A deadline, not a duration. The retry's timer has already fired, so the
+      // capture below is the one that arms the next timer — and it sees the
+      // wait still set, because the send it belongs to has not settled. Holding
+      // a duration here re-arms for the whole window a second time and leaves
+      // the sample 300s behind on an endpoint that has already recovered.
+      let settleRetry: ((outcome: SendMetricsBatchOutcome) => void) | undefined
+      let call = 0
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> => {
+          call++
+          if (call === 1) {
+            return Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+          }
+          return new Promise<SendMetricsBatchOutcome>((resolve) => {
+            settleRetry = resolve
+          })
+        }),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      // The wait elapses and the retry goes out, but hangs.
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+
+      metrics.count('orders_created', 1)
+      settleRetry?.({ kind: 'ok' })
+      await vi.advanceTimersByTimeAsync(0)
+
+      // One interval, not another window.
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('keeps its own interval when the endpoint asks for less', async () => {
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve({ kind: 'retry-later', error: new Error('503'), retryAfterMs: 10 })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('sends on an explicit flush inside the window', async () => {
+      // Lifecycle drains (RN background, shutdown) must not become no-ops for
+      // the length of a window; metrics has no `force` escape hatch.
+      const outcomes: SendMetricsBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+      ]
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve(outcomes.shift() ?? { kind: 'ok' })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      metrics.count('orders_created', 1)
+      await metrics.flush()
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('ends the wait when a later failure names none', async () => {
+      const outcomes: SendMetricsBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 },
+        { kind: 'retry-later', error: new Error('503') },
+      ]
+      const instance = createMockInstance({
+        _sendMetricsBatch: vi.fn((): Promise<SendMetricsBatchOutcome> =>
+          Promise.resolve(outcomes.shift() ?? { kind: 'ok' })
+        ),
+      })
+      const metrics = createMetrics({ flushIntervalMs: 10_000 }, instance)
+      metrics.count('orders_created', 1)
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(2)
+
+      // Off the 300s window and back on our own backoff: two consecutive
+      // failures, so one doubling of the interval rather than another 300s.
+      await vi.advanceTimersByTimeAsync(20_000)
+      expect(instance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+    })
+
     it('does not send when the window is empty', async () => {
       createMetrics({ flushIntervalMs: 5000 })
       await vi.advanceTimersByTimeAsync(15000)
@@ -519,4 +779,48 @@ describe('PostHogMetrics', () => {
       expect(optedOutInstance._sendMetricsBatch).not.toHaveBeenCalled()
     })
   })
+
+  it('does not hold a later capture at a window a successful flush already closed', async () => {
+    let sends = 0
+    mockInstance._sendMetricsBatch = vi.fn(async (): Promise<SendMetricsBatchOutcome> => {
+      sends += 1
+      return sends === 1 ? { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 } : { kind: 'ok' }
+    })
+    const metrics = createMetrics({ flushIntervalMs: 5000 })
+
+    metrics.count('a', 1)
+    await vi.advanceTimersByTimeAsync(5000)
+    await metrics.flush()
+    const closedAt = Date.now()
+
+    metrics.count('b', 1)
+    await vi.advanceTimersByTimeAsync(5000)
+
+    expect(mockInstance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+    expect(Date.now() - closedAt).toBeLessThanOrEqual(5000)
+  })
+
+  it('releases a series captured mid-flush once that flush closes the window', async () => {
+    const metrics = createMetrics({ flushIntervalMs: 10_000 })
+    let sends = 0
+    mockInstance._sendMetricsBatch = vi.fn(async (): Promise<SendMetricsBatchOutcome> => {
+      sends += 1
+      if (sends === 1) {
+        return { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }
+      }
+      await Promise.resolve()
+      if (sends === 2) {
+        metrics.count('mid', 1)
+      }
+      return { kind: 'ok' }
+    })
+
+    metrics.count('a', 1)
+    await vi.advanceTimersByTimeAsync(10_000)
+    await vi.advanceTimersByTimeAsync(10_000)
+    await metrics.flush()
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(mockInstance._sendMetricsBatch).toHaveBeenCalledTimes(3)
+  })
 })
diff --git a/packages/core/src/metrics/index.ts b/packages/core/src/metrics/index.ts
index cb86a0336e..8043ccf054 100644
--- a/packages/core/src/metrics/index.ts
+++ b/packages/core/src/metrics/index.ts
@@ -9,7 +9,10 @@ import type {
   OtlpNumberDataPoint,
 } from '@posthog/types'
 import type { Logger } from '../types'
-import { isArray, safeSetTimeout } from '../utils'
+import { isArray } from '../utils'
+import { FlushTimer } from '../utils/flush-timer'
+import { RetryAfterWindow } from '../utils/retry-after'
+import { MAX_FLUSH_BACKOFF_MS, NO_JITTER, backoffDelayMs, drawJitter } from '../utils/backoff'
 import { toOtlpKeyValueList } from '../utils/otlp-any-value'
 import {
   DEFAULT_HISTOGRAM_BOUNDS,
@@ -62,7 +65,11 @@ interface SeriesState {
  */
 export class PostHogMetrics {
   private _series = new Map<string, SeriesState>()
-  private _flushTimer?: ReturnType<typeof safeSetTimeout>
+  private readonly _flushTimer = new FlushTimer(() =>
+    this.flush().catch((e) => {
+      this._logger.error('Metrics flush failed:', e)
+    })
+  )
   // Serializes flushes — a manual flush() during an in-flight timer flush
   // queues behind it instead of racing it for the same window.
   private _flushPromise: Promise<void> | null = null
@@ -73,6 +80,9 @@ export class PostHogMetrics {
   // types under one name produces charts that blend both series.
   private _typeByName = new Map<string, MetricType>()
   private _typeCollisionWarned = new Set<string>()
+  private _retryAfter = new RetryAfterWindow()
+  private _consecutiveFlushFailures = 0
+  private _flushJitter = NO_JITTER
   // Bumped by reset(). A flush that was in flight when reset() ran (e.g. it
   // lost a shutdown race) sees a stale generation when its send settles and
   // discards its window instead of merging it back and re-arming the timer.
@@ -136,7 +146,10 @@ export class PostHogMetrics {
   /** Clears the flush timer, drops the current window, and invalidates in-flight flushes. */
   reset(): void {
     this._generation++
-    this._clearFlushTimer()
+    this._retryAfter.reset()
+    this._consecutiveFlushFailures = 0
+    this._flushJitter = NO_JITTER
+    this._flushTimer.clear()
     this._series = new Map()
     this._flushPromise = null
     this._seriesCapWarned = false
@@ -284,26 +297,35 @@ export class PostHogMetrics {
     return result
   }
 
+  // Every capture calls this, so a pending timer is left alone — re-arming on
+  // each one would push the flush out for as long as metrics keep arriving.
   private _armFlushTimer(): void {
-    if (this._flushTimer) {
+    if (this._flushTimer.pending) {
       return
     }
-    this._flushTimer = safeSetTimeout(() => {
-      this._flushTimer = undefined
-      this.flush().catch((e) => {
-        this._logger.error('Metrics flush failed:', e)
-      })
-    }, this._config.flushIntervalMs)
+    this._flushTimer.arm(this._nextFlushDelay())
   }
 
-  private _clearFlushTimer(): void {
-    if (this._flushTimer) {
-      clearTimeout(this._flushTimer)
-      this._flushTimer = undefined
-    }
+  // A floor, not a replacement: the header never retries us sooner than our own
+  // backoff would have. The backoff doubles per consecutive failure so a
+  // sustained outage is not retried every interval, and the jitter is drawn
+  // once per failure so a fleet refused together does not return together.
+  private _nextFlushDelay(): number {
+    return Math.max(
+      backoffDelayMs(
+        this._config.flushIntervalMs,
+        this._consecutiveFlushFailures,
+        this._flushJitter,
+        MAX_FLUSH_BACKOFF_MS
+      ),
+      this._retryAfter.remainingMs()
+    )
   }
 
   private async _doFlush(): Promise<void> {
+    // A flush retires the pending timer, so a delay armed for a window this
+    // flush may close cannot outlive it.
+    this._flushTimer.clear()
     if (this._series.size === 0) {
       return
     }
@@ -323,6 +345,21 @@ export class PostHogMetrics {
       // reconfigured, so this window is dropped whatever the outcome was.
       return
     }
+    this._retryAfter.record(outcome)
+    // Before any delay is taken from it, so the timer armed below is measured
+    // against this outcome rather than the one before it.
+    if (outcome.kind === 'retry-later') {
+      this._consecutiveFlushFailures++
+      this._flushJitter = drawJitter()
+    } else {
+      this._consecutiveFlushFailures = 0
+      this._flushJitter = NO_JITTER
+    }
+    // Outright, not through the ratchet: a timer a mid-flight capture armed is
+    // measured against a window this outcome may just have closed.
+    if (this._flushTimer.pending) {
+      this._flushTimer.arm(this._nextFlushDelay())
+    }
     switch (outcome.kind) {
       case 'ok':
         return
@@ -331,7 +368,7 @@ export class PostHogMetrics {
         // the next flush instead of being lost — and re-arm the timer, since
         // with no new captures nothing else would schedule that flush.
         this._mergeWindowBack(window)
-        this._armFlushTimer()
+        this._flushTimer.armNoEarlierThan(this._nextFlushDelay())
         return
       case 'too-large':
         this._logger.warn('Metrics batch exceeded the server size limit and was dropped')
diff --git a/packages/core/src/metrics/types.ts b/packages/core/src/metrics/types.ts
index 7df8be3b1f..d5bbc28668 100644
--- a/packages/core/src/metrics/types.ts
+++ b/packages/core/src/metrics/types.ts
@@ -20,8 +20,16 @@ import type { BeforeSendMetricFn, MetricAttributeValue, OtlpMetricsPayload } fro
 /** Same tagged outcome shape as `SendLogsBatchOutcome` — one policy for both signals. */
 export type SendMetricsBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'retry-later'; error: unknown }
-  | { kind: 'too-large' }
+  | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
+  | {
+      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 73565120ec..f35fe5ca67 100644
--- a/packages/core/src/posthog-core-stateless.ts
+++ b/packages/core/src/posthog-core-stateless.ts
@@ -41,6 +41,7 @@ import {
   getEventUuid,
   safeJsonStringify,
 } from './utils'
+import { parseRetryAfterMs } from './utils/retry-after'
 import { uuidv7 } from './vendor/uuidv7'
 import {
   ErrorPropertiesBuilder,
@@ -70,6 +71,20 @@ class PostHogFetchHttpError extends Error {
     return this.response.status
   }
 
+  /**
+   * The response's `Retry-After` as milliseconds from now, when it sent a usable
+   * one, clamped to `MAX_RETRY_AFTER_MS`.
+   */
+  get retryAfterMs(): number | undefined {
+    try {
+      return parseRetryAfterMs(this.response.headers?.get('retry-after'))
+    } catch {
+      // `headers.get` is injected transport code; a throwing one must not turn a
+      // retriable failure into an unhandled rejection.
+      return undefined
+    }
+  }
+
   get bodyReadTimedOut(): boolean {
     return this._bodyReadTimedOut
   }
@@ -206,6 +221,46 @@ function isRetryableFlagsFetchError(
   return code !== 'ECONNREFUSED'
 }
 
+/**
+ * Ceiling on what the SDK will put on the wire: a body over it is reported as
+ * too large without a request being made, and a batch of one that still exceeds
+ * it is dropped. The ingestion service decompresses a `Content-Encoding: gzip`
+ * request before it applies its own `MAX_REQUEST_BODY_SIZE_BYTES`, so the size
+ * that has to stay under the limit is the uncompressed one measured here.
+ *
+ * Set to the largest limit any known deployment configures — 10 MiB, what the
+ * ingestion service runs with — rather than the 2 MB the service falls back to
+ * when nothing configures it. The ceiling only earns its place by refusing a
+ * body that no deployment would have accepted: at 2 MB it would instead refuse
+ * bodies the service takes today, dropping records with no `413` to show for
+ * them. Deployments configured lower, and proxies in front of them, are covered
+ * by the `413` path, which stays the primary mechanism.
+ */
+const OTLP_MAX_BODY_BYTES = 10 * 1024 * 1024
+
+/**
+ * A request body's size on the wire. `Buffer` where it exists, `TextEncoder`
+ * elsewhere.
+ *
+ * Total by construction: it runs on hosts that define only part of the web
+ * platform — `Blob` in particular is absent on some server runtimes — and a
+ * size that cannot be measured is reported as `0`, leaving the body to be sent
+ * rather than turning a missing global into a failed export.
+ */
+function byteLengthOf(body: string | Blob | Uint8Array): number {
+  try {
+    if (typeof body !== 'string') {
+      return body instanceof Uint8Array ? body.byteLength : body.size
+    }
+    if (typeof Buffer !== 'undefined') {
+      return Buffer.byteLength(body, STRING_FORMAT)
+    }
+    return new TextEncoder().encode(body).length
+  } catch {
+    return 0
+  }
+}
+
 export function isPostHogFetchContentTooLargeError(err: unknown): err is PostHogFetchHttpError & { status: 413 } {
   return typeof err === 'object' && err instanceof PostHogFetchHttpError && err.status === 413
 }
@@ -233,8 +288,16 @@ function isPostHogEventProperties(value: JsonType | undefined): value is PostHog
  */
 export type SendLogsBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'too-large' }
-  | { kind: 'retry-later'; error: unknown }
+  | {
+      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 }
 
 /**
@@ -244,8 +307,16 @@ export type SendLogsBatchOutcome =
  */
 type SendOtlpBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'too-large' }
-  | { kind: 'retry-later'; error: unknown }
+  | {
+      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 }
 
 export enum QuotaLimitedFeature {
@@ -1327,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)[] = []
@@ -1665,13 +1745,44 @@ export abstract class PostHogCoreStateless {
       return { kind: 'fatal', error: new Error('The client is disabled') }
     }
 
-    const serialized = JSON.stringify(payload)
+    // Serialised behind a guard: a payload too big to hold as one string throws
+    // `RangeError` here, which escapes the tagged-outcome contract and leaves the
+    // caller retrying a batch it can never send. Reported as too-large so it takes
+    // the same halve-and-isolate path as a batch that serialises but is oversized.
+    let serialized: string
+    try {
+      serialized = JSON.stringify(payload)
+    } catch (error) {
+      this.logMsgIfDebug(() =>
+        console.warn(`[PostHog] Could not serialize a ${path} batch; reporting it as too large`, error)
+      )
+      return { kind: 'too-large', measuredLocally: true }
+    }
+
+    // Measured on the uncompressed payload: the endpoint decompresses the body
+    // and applies its limit to what comes out, so one that gzips small is still
+    // refused on its decompressed size. A batch the endpoint cannot accept is
+    // reported without being sent — and before it is compressed — so the caller
+    // halves it, and ultimately isolates and drops the one oversized record,
+    // without spending a request or a gzip pass on each attempt.
+    const payloadBytes = byteLengthOf(serialized)
+    if (payloadBytes > OTLP_MAX_BODY_BYTES) {
+      this.logMsgIfDebug(() =>
+        console.warn(
+          `[PostHog] Not sending a ${path} batch of ${payloadBytes} bytes: the endpoint accepts at most ${OTLP_MAX_BODY_BYTES}`
+        )
+      )
+      return { kind: 'too-large', measuredLocally: true }
+    }
+
     const url =
       auth === 'bearer'
         ? `${this.host}/i/v1/${path}`
         : `${this.host}/i/v1/${path}?token=${encodeURIComponent(this.apiKey)}`
 
     const gzippedPayload = !this.disableCompression ? await this.compressPayload(serialized) : null
+    const body = gzippedPayload || serialized
+
     const fetchOptions: PostHogFetchOptions = {
       method: 'POST',
       headers: {
@@ -1680,7 +1791,7 @@ export abstract class PostHogCoreStateless {
         ...(auth === 'bearer' && { Authorization: `Bearer ${this.apiKey}` }),
         ...(gzippedPayload !== null && { 'Content-Encoding': 'gzip' }),
       },
-      body: gzippedPayload || serialized,
+      body,
     }
 
     try {
@@ -1693,6 +1804,12 @@ export abstract class PostHogCoreStateless {
             if (isPostHogFetchContentTooLargeError(err)) {
               return false
             }
+            if (err instanceof PostHogFetchHttpError && err.retryAfterMs !== undefined) {
+              // The endpoint named a wait. This loop retries on a fixed short
+              // delay, so retrying here would spend every attempt inside the
+              // window; hand it to the queue's backoff instead.
+              return false
+            }
             return isPostHogFetchRetryableError(err)
           },
         }
@@ -1703,7 +1820,8 @@ export abstract class PostHogCoreStateless {
         return { kind: 'too-large' }
       }
       if (isPostHogFetchRetryableError(err)) {
-        return { kind: 'retry-later', error: err }
+        const retryAfterMs = err instanceof PostHogFetchHttpError ? err.retryAfterMs : undefined
+        return { kind: 'retry-later', error: err, ...(retryAfterMs !== undefined && { retryAfterMs }) }
       }
       return { kind: 'fatal', error: err }
     }
@@ -1751,25 +1869,7 @@ export abstract class PostHogCoreStateless {
     requestTimeout?: number
   ): Promise<T | void> {
     const body = options.body ? options.body : ''
-    let reqByteLength = -1
-    try {
-      if (body instanceof Blob) {
-        reqByteLength = body.size
-      } else if (body instanceof Uint8Array) {
-        reqByteLength = body.byteLength
-      } else {
-        reqByteLength = Buffer.byteLength(body, STRING_FORMAT)
-      }
-    } catch {
-      if (body instanceof Blob) {
-        reqByteLength = body.size
-      } else if (body instanceof Uint8Array) {
-        reqByteLength = body.byteLength
-      } else {
-        const encoded = new TextEncoder().encode(body)
-        reqByteLength = encoded.length
-      }
-    }
+    const reqByteLength = byteLengthOf(body)
 
     const retriableOptions = { ...this._retryOptions, ...retryOptions }
     let attempt = 0
diff --git a/packages/core/src/traces/index.spec.ts b/packages/core/src/traces/index.spec.ts
index 07bcdf9e86..ac05721be2 100644
--- a/packages/core/src/traces/index.spec.ts
+++ b/packages/core/src/traces/index.spec.ts
@@ -70,6 +70,9 @@ describe('PostHogTraces', () => {
     mockInstance = createMockInstance()
     logger = createMockLogger()
     context = {}
+    // Retry delays carry jitter; pinned to its midpoint so every timing
+    // assertion here measures the backoff itself and cannot flake.
+    vi.spyOn(Math, 'random').mockReturnValue(0.5)
   })
 
   describe('startSpan', () => {
@@ -1932,6 +1935,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({
@@ -2625,6 +2655,436 @@ describe('PostHogTraces', () => {
     })
   })
 
+  describe('Retry-After', () => {
+    it('waits at least as long as the endpoint asked', async () => {
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('throttled'),
+        retryAfterMs: 90_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 1000, maxExportBatchSize: 1 })
+      traces.startSpan('held').end()
+
+      // One attempt, then a wait longer than the 30s exponential cap would give.
+      while (mockInstance._sendTracesBatch.mock.calls.length < 1) {
+        await vi.advanceTimersByTimeAsync(1000)
+      }
+      await vi.advanceTimersByTimeAsync(60_000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(31_000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('keeps its own backoff when the endpoint asks for less', async () => {
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('down'),
+        retryAfterMs: 10,
+      })
+      const traces = createTraces({ flushIntervalMs: 5000, maxExportBatchSize: 1 })
+      traces.startSpan('held').end()
+
+      while (mockInstance._sendTracesBatch.mock.calls.length < 1) {
+        await vi.advanceTimersByTimeAsync(1000)
+      }
+      // A 10ms Retry-After must not turn the retry loop into a hot loop: the
+      // next attempt still waits out the queue's own backoff, not 10ms.
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(5000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('waits out Retry-After even when a span ends mid-flush', async () => {
+      // The span that ends while the send is in flight arms a timer at the plain
+      // interval; the 429 then asks for far longer. The earlier timer must not
+      // fire first, or the SDK sends inside the window it was told to skip.
+      mockInstance._sendTracesBatch.mockImplementation(() => {
+        traces.startSpan('arrived mid-flush').end()
+        return Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      })
+      const traces = createTraces({ flushIntervalMs: 1000 })
+      traces.startSpan('first').end()
+
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+    })
+
+    it('clears Retry-After on an outcome that is not a retry', async () => {
+      // Only a retriable outcome carries a wait. A stale one left set here would
+      // pin every later flush at the server's old window.
+      const outcomes: SendTracesBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('throttled'), retryAfterMs: 300_000 },
+        { kind: 'fatal', error: new Error('400') },
+      ]
+      mockInstance._sendTracesBatch.mockImplementation(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const traces = createTraces({ flushIntervalMs: 1000, maxExportBatchSize: 1 })
+      traces.startSpan('first').end()
+
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      // The wait elapses, the retry lands a 400, and that ends the wait.
+      await vi.advanceTimersByTimeAsync(300_000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+
+      traces.startSpan('second').end()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(3)
+    })
+
+    it('stops honouring a stale Retry-After after a success', async () => {
+      let attempt = 0
+      mockInstance._sendTracesBatch.mockImplementation(() => {
+        attempt++
+        return Promise.resolve(
+          attempt === 1 ? { kind: 'retry-later', error: new Error('throttled'), retryAfterMs: 120_000 } : { kind: 'ok' }
+        )
+      })
+      const traces = createTraces({ flushIntervalMs: 1000, maxExportBatchSize: 1 })
+      traces.startSpan('first').end()
+      await vi.advanceTimersByTimeAsync(200_000)
+
+      traces.startSpan('second').end()
+      await traces.flush()
+      expect(sentSpans().map((s) => s.name)).toContain('second')
+
+      // The wait is gone, so a later failure falls back to the plain interval.
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('down') })
+      traces.startSpan('third').end()
+      const before = mockInstance._sendTracesBatch.mock.calls.length
+      await vi.advanceTimersByTimeAsync(2000)
+      expect(mockInstance._sendTracesBatch.mock.calls.length).toBeGreaterThan(before)
+    })
+
+    it('still drains on an explicit flush inside the window', async () => {
+      // An explicit flush is a lifecycle or teardown boundary with no later
+      // attempt — on a serverless host the retry timer is unref'd and dies with
+      // the isolate — so the window must not turn it into a no-op.
+      const outcomes: SendTracesBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('throttled'), retryAfterMs: 300_000 },
+      ]
+      mockInstance._sendTracesBatch.mockImplementation(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      await traces.flush()
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      // The endpoint has recovered, and the caller asked explicitly.
+      await traces.flush()
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+      expect(sentSpans().map((s) => s.name)).toContain('first')
+    })
+
+    it('does not spend the head batch retry budget inside the window', async () => {
+      // The batch is dropped after MAX_RETRIES_PER_BATCH consecutive failures.
+      // At a host's flush cadence that budget would burn out long before the
+      // window the endpoint asked for, losing the spans with it.
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('throttled'),
+        retryAfterMs: 300_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      for (let i = 0; i < 12; i++) {
+        await traces.flush()
+        await vi.advanceTimersByTimeAsync(10_000)
+      }
+
+      expect(logger.warn).not.toHaveBeenCalledWith(expect.stringContaining('Dropping'))
+      expect(mockInstance._sendTracesBatch.mock.calls.length).toBeGreaterThan(1)
+
+      // The budget is deferred, not disabled. Once the window elapses the batch
+      // must retire — a deadline that slid forward on each refusal would keep
+      // the window open forever and strand everything behind the head batch.
+      for (let i = 0; i < 12; i++) {
+        await vi.advanceTimersByTimeAsync(310_000)
+        await traces.flush()
+      }
+      expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('failed 8 times in a row'))
+    })
+
+    it('charges the budget for a timer-path failure after the wait was served out', async () => {
+      // The exemption is narrow: an attempt made *early*, inside a window that
+      // has not elapsed. A retry the timer fired at the deadline has served the
+      // wait, so it must count — otherwise the budget never advances and a
+      // permanently refused head batch pins everything behind it.
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('429'),
+        retryAfterMs: 60_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 1000 })
+      traces.startSpan('first').end()
+
+      for (let i = 0; i < 12; i++) {
+        await vi.advanceTimersByTimeAsync(61_000)
+      }
+
+      expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('failed 8 times in a row'))
+    })
+
+    it('does not let a host out-pacing the window stall the retry budget', async () => {
+      // flush() more often than the window is long. Each refusal slides the
+      // deadline, up to five minutes from where the window was installed; the
+      // in-window sends are uncharged, so each window costs one charge and eight
+      // of them retire the head batch rather than letting it hold the queue.
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('429'),
+        retryAfterMs: 30_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 10_000, maxQueueSize: 5, maxExportBatchSize: 2 })
+
+      for (let i = 0; i < 500; i++) {
+        traces.startSpan(`span-${i}`).end()
+        await traces.flush()
+        await vi.advanceTimersByTimeAsync(5_000)
+      }
+
+      expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('failed 8 times in a row'))
+    })
+
+    it('still charges the retry budget for failures outside any window', async () => {
+      // The budget must keep working when the endpoint names no wait, or a
+      // permanently failing head batch pins the queue forever. It takes the
+      // eight backoff windows the timer would have waited — roughly three
+      // minutes at this interval — however often the caller drains in between.
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('503') })
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      for (let i = 0; i < 25; i++) {
+        await traces.flush()
+        await vi.advanceTimersByTimeAsync(10_000)
+      }
+
+      expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Dropping'))
+    })
+
+    it('does not spend the retry budget on a caller draining faster than the backoff', async () => {
+      // A serverless host calls `flush()` per invocation. Charging every refusal
+      // retired the batch in eight calls and no elapsed time at all, so a blip
+      // the timer would have ridden out cost the spans instead of a request.
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('503') })
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      for (let i = 0; i < 20; i++) {
+        await traces.flush()
+      }
+
+      expect((traces as any)._queue).toHaveLength(1)
+      expect(logger.warn).not.toHaveBeenCalledWith(expect.stringContaining('Dropping'))
+    })
+
+    it('does not charge a caller-driven flush inside a window a later refusal extended', async () => {
+      // The charge point is set from the window as it stood at the first
+      // refusal. An in-window refusal then extends the window past it, and a
+      // flush between the two deadlines is still inside the endpoint's wait.
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('429'),
+        retryAfterMs: 60_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      await traces.flush()
+      expect((traces as any)._headBatchFailures).toBe(1)
+
+      await vi.advanceTimersByTimeAsync(50_000)
+      await traces.flush()
+      await vi.advanceTimersByTimeAsync(10_000)
+      const sends = mockInstance._sendTracesBatch.mock.calls.length
+      await traces.flush()
+
+      expect(mockInstance._sendTracesBatch.mock.calls.length).toBe(sends + 1)
+      expect((traces as any)._headBatchFailures).toBe(1)
+    })
+
+    it('charges once per window however many attempts share it', async () => {
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('503') })
+      const traces = createTraces({ flushIntervalMs: 10_000 })
+      traces.startSpan('first').end()
+
+      await traces.flush()
+      expect((traces as any)._headBatchFailures).toBe(1)
+
+      await traces.flush()
+      await traces.flush()
+      expect((traces as any)._headBatchFailures).toBe(1)
+
+      await vi.advanceTimersByTimeAsync(30_000)
+      await traces.flush()
+      expect((traces as any)._headBatchFailures).toBe(2)
+    })
+
+    it('gives a fresh batch a fresh budget after an opt-out cleared the queue', async () => {
+      // The head batch leaves with the queue, so its budget must leave too:
+      // otherwise the next batch inherits a spent retry count and a charge
+      // deadline still in the future, and its first refusal goes uncharged.
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('503') })
+      const traces = createTraces({ flushIntervalMs: 10_000, maxExportBatchSize: 1 })
+      traces.startSpan('before').end()
+      await traces.flush()
+      expect((traces as any)._headBatchFailures).toBe(1)
+
+      mockInstance.optedOut = true
+      await traces.flush()
+
+      // Asserted here rather than through a later refusal: the charge deadline
+      // the first failure installed is still in the future either way, so the
+      // next refusal is uncharged and the count reads 1 with or without the reset.
+      expect((traces as any)._headBatchFailures).toBe(0)
+      expect((traces as any)._headBatchChargeableAt).toBe(0)
+    })
+
+    it('gives a fresh batch a fresh window after the head is retired', async () => {
+      // The deadline belongs to the batch, not the queue: a new head must be
+      // chargeable straight away or it inherits the last one's served wait.
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'fatal', error: new Error('400') })
+      const traces = createTraces({ flushIntervalMs: 10_000, maxExportBatchSize: 1 })
+      traces.startSpan('first').end()
+      await traces.flush()
+
+      mockInstance._sendTracesBatch.mockResolvedValue({ kind: 'retry-later', error: new Error('503') })
+      traces.startSpan('second').end()
+      await traces.flush()
+
+      expect((traces as any)._headBatchFailures).toBe(1)
+    })
+
+    it('does not let a backward clock step hold the retry budget open', async () => {
+      // The deadline is wall clock. An NTP correction or a resumed VM would
+      // otherwise keep the window "open" for the size of the step — and with
+      // in-window refusals uncharged, the head batch would never retire.
+      mockInstance._sendTracesBatch.mockResolvedValue({
+        kind: 'retry-later',
+        error: new Error('429'),
+        retryAfterMs: 60_000,
+      })
+      const traces = createTraces({ flushIntervalMs: 1000 })
+      traces.startSpan('first').end()
+      await traces.flush()
+
+      // Step the clock back an hour and let simulated time run from there.
+      vi.setSystemTime(Date.now() - 3_600_000)
+
+      // Asserted on the queue, not the drop warning: `_recordDrop` paces that
+      // warning off the same clock and would suppress it here.
+      for (let i = 0; i < 12; i++) {
+        await traces.flush()
+        await vi.advanceTimersByTimeAsync(61_000)
+      }
+      expect((traces as any)._queue).toHaveLength(0)
+    })
+
+    it('keeps the Retry-After window when a batch is refused for size', async () => {
+      // `too-large` is a verdict on the body's size — the SDK's own or a 413 —
+      // so it says nothing about the endpoint's rate limit and must not end the
+      // wait.
+      let first = true
+      mockInstance._sendTracesBatch.mockImplementation(() => {
+        const outcome: SendTracesBatchOutcome = first
+          ? { kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 }
+          : { kind: 'too-large' }
+        first = false
+        return Promise.resolve(outcome)
+      })
+      const traces = createTraces({ flushIntervalMs: 1000, maxExportBatchSize: 1 })
+      traces.startSpan('first').end()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      // Batches of one the endpoint cannot accept: the spans are dropped.
+      traces.startSpan('second').end()
+      await traces.flush()
+      const sends = mockInstance._sendTracesBatch.mock.calls.length
+
+      traces.startSpan('third').end()
+      await vi.advanceTimersByTimeAsync(10_000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(sends)
+    })
+
+    it('stops the drain when it retires a batch inside the window', async () => {
+      // Retiring the head batch clears the failure counters, so the drain would
+      // otherwise carry straight on to the next batch — inside the wait the very
+      // refusal that retired it had just installed. One extra send still lands at
+      // that instant: `flush()`'s outer loop is deliberately not window-aware.
+      const sentAt: number[] = []
+      mockInstance._sendTracesBatch.mockImplementation(() => {
+        sentAt.push(Date.now())
+        return Promise.resolve({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 60_000 })
+      })
+      const traces = createTraces({ flushIntervalMs: 1000, maxExportBatchSize: 1 })
+      for (let i = 0; i < 4; i++) {
+        traces.startSpan(`span-${i}`).end()
+      }
+
+      // Every attempt lands after the previous window elapsed, so every refusal
+      // is charged and the eighth retires the head batch.
+      for (let i = 0; i < 12; i++) {
+        await vi.advanceTimersByTimeAsync(61_000)
+      }
+
+      const perInstant = new Map<number, number>()
+      for (const at of sentAt) {
+        perInstant.set(at, (perInstant.get(at) ?? 0) + 1)
+      }
+      expect(Math.max(...perInstant.values())).toBe(2)
+    })
+
+    it('does not install a Retry-After that lands after reset', async () => {
+      // The generation check must precede the window assignment, or a 429 that
+      // settles after teardown pins a wait on the fresh client.
+      let settle: ((outcome: SendTracesBatchOutcome) => void) | undefined
+      mockInstance._sendTracesBatch.mockImplementation(
+        () =>
+          new Promise((resolve) => {
+            settle = resolve
+          })
+      )
+      const traces = createTraces({ flushIntervalMs: 1000 })
+      traces.startSpan('before-reset').end()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      traces.reset()
+      traces.startSpan('after-reset').end()
+
+      settle?.({ kind: 'retry-later', error: new Error('429'), retryAfterMs: 300_000 })
+      await vi.advanceTimersByTimeAsync(0)
+
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+    })
+
+    it('drops a Retry-After wait on reset', async () => {
+      const outcomes: SendTracesBatchOutcome[] = [
+        { kind: 'retry-later', error: new Error('throttled'), retryAfterMs: 300_000 },
+      ]
+      mockInstance._sendTracesBatch.mockImplementation(() => Promise.resolve(outcomes.shift() ?? { kind: 'ok' }))
+      const traces = createTraces({ flushIntervalMs: 1000 })
+      traces.startSpan('first').end()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(1)
+
+      traces.reset()
+      traces.startSpan('second').end()
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(mockInstance._sendTracesBatch).toHaveBeenCalledTimes(2)
+    })
+  })
+
   describe('live span bounds', () => {
     it('returns an inert handle once maxLiveSpans spans are live', async () => {
       const traces = createTraces({ maxLiveSpans: 2 })
diff --git a/packages/core/src/traces/index.ts b/packages/core/src/traces/index.ts
index bf25098deb..6b2fac63ae 100644
--- a/packages/core/src/traces/index.ts
+++ b/packages/core/src/traces/index.ts
@@ -24,18 +24,18 @@ import { parseTraceparent, sanitizeTracestate, traceparentHeader } from './trace
 import { clampEndTime, resolveStartTime, resolveSuppliedTime, sanitizeName, toEpochMs } from './sanitize'
 import { assignUserAttributes } from '../utils/json-utils'
 import { buildOtlpSpan, buildOtlpTracesPayload, buildTracesResourceAttributes } from './otlp'
-import { isPromise, safeSetTimeout } from '../utils'
+import { isPromise } from '../utils'
+import { FlushTimer } from '../utils/flush-timer'
+import { RetryAfterWindow } from '../utils/retry-after'
+import { MAX_FLUSH_BACKOFF_MS, NO_JITTER, backoffDelayMs, drawJitter } from '../utils/backoff'
 
 // Retriable failures on the same head batch before it is dropped, so a stuck
 // batch cannot pin the queue while fresher spans are refused at the cap. The
-// budget counts attempts, not elapsed time: on the timer path the backoff
-// spreads them over minutes, while a host that calls `flush()` per request
-// spends them as fast as the requests arrive.
+// budget counts backoff windows rather than attempts: a host that drains on
+// every request would otherwise retire a batch in milliseconds, spending on
+// its own call rate what the timer path spends over minutes.
 const MAX_RETRIES_PER_BATCH = 8
 
-const MAX_FLUSH_BACKOFF_EXPONENT = 6
-const MAX_FLUSH_BACKOFF_MS = 30_000
-
 type SpanCallback<T> = (span: Span) => T
 
 /** Monotonic where the platform has one, wall clock otherwise. Both are ms, and a platform never switches. */
@@ -166,7 +166,7 @@ interface ParentContext {
  */
 export class PostHogTraces {
   private _queue: SpanRecord[] = []
-  private _flushTimer?: ReturnType<typeof safeSetTimeout>
+  private readonly _flushTimer = new FlushTimer(() => this._flushInBackground())
   // Serializes flushes: a second caller joins the first instead of double-sending the head.
   private _flushPromise: Promise<number> | null = null
   // A trigger no-ops while a background drain is already pending.
@@ -177,12 +177,17 @@ export class PostHogTraces {
   private _lastDropWarningAt = 0
   private _dropReasons = new Set<string>()
   private _consecutiveFlushFailures = 0
+  private _flushJitter = NO_JITTER
+  private _retryAfter = new RetryAfterWindow()
   // Separate from the backoff counter: this one belongs to whatever batch is at
   // the head, and resets whenever that batch is removed or shrunk.
   private _headBatchFailures = 0
   // Read only while a budget is in flight, so the head cannot grow to sweep in
   // fresh spans and drop them on a budget they never spent.
   private _headBatchSize = 0
+  // When the head batch may next be charged, on `clockNow`'s basis: one failure
+  // per backoff window, whoever drove the attempt.
+  private _headBatchChargeableAt = 0
   // Bumped by reset(); a pass whose generation is stale abandons the queue.
   private _generation = 0
   // Live-span accounting: span id -> monotonic start. Ids and numbers only,
@@ -327,7 +332,22 @@ export class PostHogTraces {
    *
    * A pass reports spans removed — queue length can't stand in, since a send
    * concurrent with an arrival leaves it unchanged.
+   *
+   * An open `Retry-After` window does not stop this: an explicit flush, or one a
+   * host runs to keep a request alive, sends whatever is queued — a serverless
+   * isolate may not be around for the armed timer to fire. The window is honoured
+   * by not charging such an attempt against the head batch's retry budget, so the
+   * 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) {
@@ -346,7 +366,7 @@ export class PostHogTraces {
   }
 
   private _startFlush(): Promise<number> {
-    this._clearFlushTimer()
+    this._flushTimer.clear()
     // Deferred by a microtask so the slot below is installed before the pass
     // reads anything: `_flushInner` runs synchronously as far as its first
     // await, and a resource-attribute getter or `toJSON` that ends a span in
@@ -364,7 +384,7 @@ export class PostHogTraces {
         if (this._flushPromise === promise) {
           this._flushPromise = null
         }
-        this._armFlushTimerIfQueued()
+        this._armFlushTimerIfQueuedNoEarlierThan()
       })
     this._flushPromise = promise
     return promise
@@ -372,7 +392,7 @@ export class PostHogTraces {
 
   /** Clears the queue and timer. Used on shutdown and between tests. */
   reset(): void {
-    this._clearFlushTimer()
+    this._flushTimer.clear()
     if (this._queue.length) {
       // Critical, and said here rather than counted: this is the last chance to
       // say anything about these spans, the drop warning is gated behind `debug`
@@ -393,7 +413,15 @@ export class PostHogTraces {
     this._dropReasons.clear()
     this._lastDropWarningAt = 0
     this._consecutiveFlushFailures = 0
+    this._flushJitter = NO_JITTER
+    this._retryAfter.reset()
+    this._resetHeadBatchBudget()
+  }
+
+  /** Called wherever the head batch leaves or changes shape, so its budget goes with it. */
+  private _resetHeadBatchBudget(): void {
     this._headBatchFailures = 0
+    this._headBatchChargeableAt = 0
   }
 
   /**
@@ -536,10 +564,14 @@ export class PostHogTraces {
       this._logger.debug('Span queue notification failed', error)
     }
 
-    // Not while a flush is failing: the queue stays above the batch size for the
-    // whole outage, so every further span end would re-POST immediately and the
-    // retry backoff would never apply.
-    if (this._queue.length >= this._maxExportBatchSize && !this._consecutiveFlushFailures) {
+    // Not while a flush is failing or the endpoint has asked us to wait: the
+    // queue stays above the batch size for the whole outage, so every further
+    // span end would re-POST immediately and the retry backoff would never apply.
+    if (
+      this._queue.length >= this._maxExportBatchSize &&
+      !this._consecutiveFlushFailures &&
+      !this._retryAfter.isOpen()
+    ) {
       this._flushInBackground()
     } else {
       this._armFlushTimerIfQueued()
@@ -792,6 +824,9 @@ export class PostHogTraces {
     }
     const discarded = this._queue.length
     this._queue = []
+    // The head batch left with the queue, so its budget goes too — anything
+    // queued after consent returns is a different batch.
+    this._resetHeadBatchBudget()
     this._recordDrop(discarded, 'the user has opted out')
     this._warnAboutDrops()
     return discarded
@@ -820,6 +855,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 {
@@ -836,7 +875,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)
 
@@ -845,10 +884,16 @@ export class PostHogTraces {
           this._queue.splice(0, size)
           remaining -= size
           removed += size
-          this._headBatchFailures = 0
+          this._resetHeadBatchBudget()
           continue
         }
 
+        // Read before the send, so the budget below charges this attempt against
+        // the window it was actually made under. A send inside an open
+        // `Retry-After` window is caller-driven and exempt from the wait; a later
+        // refusal can extend that window past the charge point, so both are read.
+        const chargeable = clockNow() >= this._headBatchChargeableAt && !this._retryAfter.isOpen()
+
         const outcome = await this._instance._sendTracesBatch(
           buildOtlpTracesPayload(spans, resourceAttributes, scopeName, scopeVersion, this._logger)
         )
@@ -858,9 +903,12 @@ export class PostHogTraces {
           return removed
         }
 
+        this._retryAfter.record(outcome)
+
         if (outcome.kind === 'ok') {
           this._consecutiveFlushFailures = 0
-          this._headBatchFailures = 0
+          this._flushJitter = NO_JITTER
+          this._resetHeadBatchBudget()
           this._queue.splice(0, size)
           remaining -= size
           removed += size
@@ -877,24 +925,37 @@ export class PostHogTraces {
             this._queue.splice(0, 1)
             remaining -= 1
             removed += 1
-            this._recordDrop(1, 'the ingestion endpoint rejected it as too large')
+            this._recordDrop(1, 'it is too large for the ingestion endpoint')
             this._consecutiveFlushFailures = 0
-            this._headBatchFailures = 0
+            this._flushJitter = NO_JITTER
+            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._headBatchFailures = 0
-          this._logger.debug(`Batch too large; retrying the same spans in batches of ${this._maxExportBatchSize}`)
+          this._resetHeadBatchBudget()
+          this._logger.debug(`Batch too large; retrying the same spans in batches of ${halved}`)
           continue
         }
 
         if (outcome.kind === 'retry-later') {
           this._consecutiveFlushFailures++
-          this._headBatchFailures++
+          this._flushJitter = drawJitter()
+          // One charge per backoff window: a refusal arriving before the window
+          // the last charge bought has elapsed is the same refusal seen again,
+          // not new evidence against the batch.
           this._headBatchSize = size
+          if (chargeable) {
+            this._headBatchFailures++
+            this._headBatchChargeableAt = clockNow() + this._nextFlushDelay()
+          }
           if (this._headBatchFailures < MAX_RETRIES_PER_BATCH) {
             // Keep the spans queued; the flush timer picks them up again.
             this._logger.debug('Span export failed; retrying on the next flush', outcome.error)
@@ -906,8 +967,17 @@ export class PostHogTraces {
           remaining -= size
           removed += size
           this._consecutiveFlushFailures = 0
-          this._headBatchFailures = 0
+          this._flushJitter = NO_JITTER
+          this._resetHeadBatchBudget()
           this._recordDrop(size, `the ingestion endpoint failed ${MAX_RETRIES_PER_BATCH} times in a row`)
+          if (this._retryAfter.isOpen()) {
+            // Retiring the batch cleared the failure counters, so this drain
+            // would carry straight on to the next batch inside the endpoint's
+            // wait. Ending the pass here costs one send rather than the rest of
+            // the queue: `flush()` sees a non-zero count and loops, so the next
+            // batch still goes out, one pass later.
+            return removed
+          }
           continue
         }
 
@@ -917,7 +987,8 @@ export class PostHogTraces {
         remaining -= size
         removed += size
         this._consecutiveFlushFailures = 0
-        this._headBatchFailures = 0
+        this._flushJitter = NO_JITTER
+        this._resetHeadBatchBudget()
         this._recordDrop(size, 'the ingestion endpoint rejected the batch')
       }
 
@@ -948,32 +1019,42 @@ export class PostHogTraces {
         this._backgroundFlush = undefined
         // A trigger that arrived while this drain was finishing found the guard
         // set and the queue empty, so neither path armed a timer.
-        this._armFlushTimerIfQueued()
+        this._armFlushTimerIfQueuedNoEarlierThan()
       })
   }
 
+  // Every span end can reach this, so a pending timer is left alone rather than
+  // pushing the flush out.
   private _armFlushTimerIfQueued(): void {
-    if (this._flushTimer || !this._queue.length) {
+    if (this._flushTimer.pending || !this._queue.length) {
       return
     }
-    this._flushTimer = safeSetTimeout(() => {
-      this._flushTimer = undefined
-      this._flushInBackground()
-    }, this._nextFlushDelay())
+    this._flushTimer.arm(this._nextFlushDelay())
   }
 
-  // Retry delay: base interval, doubling, capped at 30s — never below an interval
-  // a host configured above the cap.
-  private _nextFlushDelay(): number {
-    const exponent = Math.min(Math.max(0, this._consecutiveFlushFailures - 1), MAX_FLUSH_BACKOFF_EXPONENT)
-    const delay = this._config.flushIntervalMs * 2 ** exponent
-    return Math.min(delay, Math.max(MAX_FLUSH_BACKOFF_MS, this._config.flushIntervalMs))
+  // Both floors, so a timer a span end armed at the plain interval gives way to
+  // a longer one.
+  private _armFlushTimerIfQueuedNoEarlierThan(): void {
+    if (!this._queue.length) {
+      return
+    }
+    this._flushTimer.armNoEarlierThan(this._nextFlushDelay())
   }
 
-  private _clearFlushTimer(): void {
-    if (this._flushTimer) {
-      clearTimeout(this._flushTimer)
-      this._flushTimer = undefined
-    }
+  // Retry delay: base interval, doubling, capped at 30s — never below an interval
+  // a host configured above the cap. The jitter is drawn once per failure, so the
+  // timer and the retry-budget charge point are measured against the same delay.
+  private _nextFlushDelay(): number {
+    // A floor, not a replacement: the header never retries us sooner than our
+    // own backoff would have.
+    return Math.max(
+      backoffDelayMs(
+        this._config.flushIntervalMs,
+        this._consecutiveFlushFailures,
+        this._flushJitter,
+        MAX_FLUSH_BACKOFF_MS
+      ),
+      this._retryAfter.remainingMs()
+    )
   }
 }
diff --git a/packages/core/src/traces/types.ts b/packages/core/src/traces/types.ts
index 6d95c1b29a..f22ccc7301 100644
--- a/packages/core/src/traces/types.ts
+++ b/packages/core/src/traces/types.ts
@@ -29,8 +29,16 @@ import type {
 /** Same tagged outcome shape as `SendLogsBatchOutcome` — one policy for all three signals. */
 export type SendTracesBatchOutcome =
   | { kind: 'ok' }
-  | { kind: 'retry-later'; error: unknown }
-  | { kind: 'too-large' }
+  | { kind: 'retry-later'; error: unknown; retryAfterMs?: number }
+  | {
+      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/core/src/utils/backoff.ts b/packages/core/src/utils/backoff.ts
new file mode 100644
index 0000000000..934291db81
--- /dev/null
+++ b/packages/core/src/utils/backoff.ts
@@ -0,0 +1,44 @@
+/** Doublings the retry delay may grow by before it stops growing. */
+export const MAX_FLUSH_BACKOFF_EXPONENT = 6
+
+/**
+ * Ceiling on the SDK's own retry delay, which the logs and traces contracts
+ * both state as "exponential backoff capped at ~30s". A host that configured a
+ * longer flush interval keeps it: the cap is there to stop the doubling running
+ * away, not to flush more often than asked.
+ */
+export const MAX_FLUSH_BACKOFF_MS = 30_000
+
+/**
+ * How far a delay may be moved either side of its computed value. Clients that
+ * fail together otherwise retry together, and arrive at the endpoint as one
+ * burst each time it comes back — which is what OTel asks jitter to prevent.
+ *
+ * A quarter is enough to spread a fleet without a retry landing so early that
+ * it beats the interval the host configured, or so late that a recovered
+ * endpoint sits idle.
+ */
+const JITTER = 0.25
+
+/** A multiplier in `[1 - JITTER, 1 + JITTER]`, drawn once per failure by the caller. */
+export function drawJitter(): number {
+  return 1 - JITTER + Math.random() * JITTER * 2
+}
+
+/** No jitter, for the delay a queue uses when nothing has failed. */
+export const NO_JITTER = 1
+
+/**
+ * The delay before retrying an export, `baseMs` doubled once per failure past
+ * the first and capped at `maxMs`.
+ *
+ * `jitter` is applied to the SDK's own delay only. A `Retry-After` the endpoint
+ * sent is a floor underneath it, applied by the caller: spreading a fleet must
+ * never move a retry earlier than the endpoint asked for.
+ */
+export function backoffDelayMs(baseMs: number, failures: number, jitter: number, maxMs?: number): number {
+  const exponent = Math.min(Math.max(0, failures - 1), MAX_FLUSH_BACKOFF_EXPONENT)
+  const delay = baseMs * 2 ** exponent
+  const capped = maxMs === undefined ? delay : Math.min(delay, Math.max(maxMs, baseMs))
+  return Math.round(capped * jitter)
+}
diff --git a/packages/core/src/utils/flush-timer.ts b/packages/core/src/utils/flush-timer.ts
new file mode 100644
index 0000000000..a0ac9ae338
--- /dev/null
+++ b/packages/core/src/utils/flush-timer.ts
@@ -0,0 +1,50 @@
+import { safeSetTimeout } from './index'
+
+/**
+ * The pending flush timer of an export queue, and when it is due.
+ *
+ * The queues arm this from two kinds of place: one that must not push a pending
+ * flush further out (every capture reaches it), and one that must not pull a
+ * flush back in front of a wait the endpoint asked for. Holding the deadline
+ * next to the handle is what lets the second kind compare against the first.
+ */
+export class FlushTimer {
+  private _timer?: ReturnType<typeof safeSetTimeout>
+  private _firesAt = 0
+
+  /** @param _onFire runs when the timer elapses, after the handle is released. */
+  constructor(private readonly _onFire: () => void) {}
+
+  /** Whether a flush is already scheduled. */
+  get pending(): boolean {
+    return !!this._timer
+  }
+
+  /** Schedules a flush in `delayMs`, replacing any timer already pending. */
+  arm(delayMs: number): void {
+    this.clear()
+    this._firesAt = Date.now() + delayMs
+    this._timer = safeSetTimeout(() => {
+      this._timer = undefined
+      this._onFire()
+    }, delayMs)
+  }
+
+  /**
+   * Arms only if it moves the flush later, so a timer armed for a longer wait
+   * survives a caller asking for a shorter one.
+   */
+  armNoEarlierThan(delayMs: number): void {
+    if (this._timer && Date.now() + delayMs <= this._firesAt) {
+      return
+    }
+    this.arm(delayMs)
+  }
+
+  clear(): void {
+    if (this._timer) {
+      clearTimeout(this._timer)
+      this._timer = undefined
+    }
+  }
+}
diff --git a/packages/core/src/utils/retry-after.ts b/packages/core/src/utils/retry-after.ts
new file mode 100644
index 0000000000..3751af30f3
--- /dev/null
+++ b/packages/core/src/utils/retry-after.ts
@@ -0,0 +1,135 @@
+/**
+ * Longest `Retry-After` the SDK will wait. A rate-limit window can legitimately
+ * be long, but nothing upstream of the SDK bounds this header — it is as likely
+ * to come from a proxy or CDN as from PostHog — and an unbounded value would
+ * strand a queue for hours, so the wait is capped and the retry happens early.
+ */
+export const MAX_RETRY_AFTER_MS = 5 * 60_000
+
+/**
+ * How far the wall clock may run behind the moment a window was installed before
+ * the window is discarded. NTP corrections on a healthy host are milliseconds,
+ * so a step past this is a clock change the deadline can no longer measure.
+ */
+const CLOCK_STEP_TOLERANCE_MS = 5_000
+
+/**
+ * `Retry-After` as milliseconds from now. The header is either delta-seconds or
+ * an HTTP-date; anything else, a date already in the past, or a non-positive
+ * delta yields `undefined` so the caller keeps its own backoff. The result is
+ * capped at `MAX_RETRY_AFTER_MS`.
+ */
+export function parseRetryAfterMs(value: unknown, now: number = Date.now()): number | undefined {
+  if (typeof value !== 'string' || !value) {
+    return undefined
+  }
+  const raw = value.trim()
+  // `headers.get` joins repeated headers with ", ", so a CDN and a load
+  // balancer that each append one yield "60, 120". Read the first, which the
+  // outermost hop set — but only for delta-seconds, since an HTTP-date carries
+  // a comma of its own ("Wed, 21 Oct 2015 07:28:00 GMT").
+  const trimmed = /^\d+\s*,/.test(raw) ? raw.slice(0, raw.indexOf(',')).trim() : raw
+  // Integer seconds. Not parseFloat: "10 minutes" must not read as 10 seconds.
+  const seconds = /^\d+$/.test(trimmed) ? Number(trimmed) : Number.NaN
+  if (!Number.isFinite(seconds) && /^[+-]?[\d.]+$/.test(trimmed)) {
+    // Numeric but not delta-seconds, so it is malformed. `Date.parse` reads
+    // "-5", "+5" and "5.5" as dates in 2001 rather than rejecting them, which
+    // on a device whose clock predates that would surface as a real wait.
+    return undefined
+  }
+  const ms = Number.isFinite(seconds) ? seconds * 1000 : Date.parse(trimmed) - now
+  if (!Number.isFinite(ms) || ms <= 0) {
+    return undefined
+  }
+  return Math.min(ms, MAX_RETRY_AFTER_MS)
+}
+
+/** The part of an export outcome the window reads. */
+export type RetryAfterOutcome = {
+  kind: 'ok' | 'retry-later' | 'too-large' | 'fatal'
+  retryAfterMs?: number
+}
+
+/**
+ * The wait an ingestion endpoint asked for, shared by the logs, metrics and
+ * traces export queues.
+ *
+ * Held as an absolute deadline rather than a duration, so a timer armed while
+ * the wait is already part-served counts down the remainder instead of
+ * restarting it.
+ */
+export class RetryAfterWindow {
+  private _until = 0
+  private _installedAt = 0
+
+  /**
+   * Folds one export outcome into the window.
+   *
+   * A refusal naming a longer wait than the one being served pushes the
+   * deadline out, but never past `MAX_RETRY_AFTER_MS` from where the window was
+   * first installed. That ceiling is the SDK's own policy rather than OTLP's,
+   * which asks for the header to be honoured and names no cap: a host flushing
+   * faster than the window is long would otherwise refresh the deadline forever
+   * and never recover, since logs gates its size trigger and `onReconnect` on
+   * the window and metrics re-arms its timer from it. The cost is that a wait
+   * longer than the ceiling is served short.
+   *
+   * The deadline is never pulled in, so a shorter header cannot cut a wait the
+   * endpoint already asked for.
+   */
+  record(outcome: RetryAfterOutcome): void {
+    if (outcome.kind === 'too-large') {
+      // A verdict on the body's size, whether the SDK's own or the endpoint's
+      // 413, and either way silent on the endpoint's rate limit.
+      return
+    }
+    if (outcome.kind !== 'retry-later') {
+      // A stale deadline left set here would pin every later send at a window
+      // the endpoint has moved on from.
+      this.reset()
+      return
+    }
+    if (!outcome.retryAfterMs) {
+      // A refusal that names no wait — a network error, a timeout, a
+      // header-less 503 — does not revoke one the endpoint already named.
+      return
+    }
+    // Read before `_installedAt` is used: a spent window resets it, and a
+    // backward clock step is caught here rather than extending off a stale one.
+    const open = this.isOpen()
+    const now = Date.now()
+    const asked = Math.min(outcome.retryAfterMs, MAX_RETRY_AFTER_MS)
+    if (!open) {
+      this._installedAt = now
+      this._until = now + asked
+      return
+    }
+    this._until = Math.max(this._until, Math.min(now + asked, this._installedAt + MAX_RETRY_AFTER_MS))
+  }
+
+  /**
+   * Milliseconds left of the wait, `0` when none is open. Closes a window it
+   * finds spent, so a later backward clock step cannot bring it back.
+   */
+  remainingMs(): number {
+    const now = Date.now()
+    if (now < this._installedAt - CLOCK_STEP_TOLERANCE_MS) {
+      this.reset()
+      return 0
+    }
+    const remaining = Math.min(MAX_RETRY_AFTER_MS, Math.max(0, this._until - now))
+    if (remaining === 0) {
+      this.reset()
+    }
+    return remaining
+  }
+
+  isOpen(): boolean {
+    return this.remainingMs() > 0
+  }
+
+  reset(): void {
+    this._until = 0
+    this._installedAt = 0
+  }
+}
diff --git a/packages/node/references/posthog-node-references-latest.json b/packages/node/references/posthog-node-references-latest.json
index 227784b49b..7b452faf35 100644
--- a/packages/node/references/posthog-node-references-latest.json
+++ b/packages/node/references/posthog-node-references-latest.json
@@ -4092,21 +4092,21 @@
       "name": "SendLogsBatchOutcome",
       "properties": [],
       "path": "../core/src/posthog-core-stateless.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SendMetricsBatchOutcome",
       "name": "SendMetricsBatchOutcome",
       "properties": [],
       "path": "../core/src/metrics/types.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SendTracesBatchOutcome",
       "name": "SendTracesBatchOutcome",
       "properties": [],
       "path": "../core/src/traces/types.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SentryIntegrationOptions",
diff --git a/packages/node/src/__tests__/traces.spec.ts b/packages/node/src/__tests__/traces.spec.ts
index a27b233b34..e6a3e72c07 100644
--- a/packages/node/src/__tests__/traces.spec.ts
+++ b/packages/node/src/__tests__/traces.spec.ts
@@ -555,6 +555,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 8194d73224..1f56641f64 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<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) {
diff --git a/packages/react-native/references/posthog-react-native-references-latest.json b/packages/react-native/references/posthog-react-native-references-latest.json
index 4da958d6bc..4a873d27d1 100644
--- a/packages/react-native/references/posthog-react-native-references-latest.json
+++ b/packages/react-native/references/posthog-react-native-references-latest.json
@@ -3733,7 +3733,7 @@
           "name": "flushIntervalMs"
         },
         {
-          "description": "Max records per outbound POST. Keeps each request under the server's 2 MB cap. On a 413 response, the SDK halves this value, retries the same records, then ramps back up by 1 per healthy send. A 413 on a single-record batch drops the record (it's larger than the server can accept regardless of batch size). Default: 50 (RN) / 100 (browser).",
+          "description": "Max records per outbound POST. Keeps each request under the server's request body cap. On a 413 response, the SDK halves this value, retries the same records, then ramps back up by 1 per healthy send. A 413 on a single-record batch drops the record (it's larger than the server can accept regardless of batch size). Default: 50 (RN) / 100 (browser).",
           "type": "number",
           "name": "maxBatchRecordsPerPost"
         },
@@ -4453,21 +4453,21 @@
       "name": "SendLogsBatchOutcome",
       "properties": [],
       "path": "../core/src/posthog-core-stateless.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SendMetricsBatchOutcome",
       "name": "SendMetricsBatchOutcome",
       "properties": [],
       "path": "../core/src/metrics/types.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SendTracesBatchOutcome",
       "name": "SendTracesBatchOutcome",
       "properties": [],
       "path": "../core/src/traces/types.ts",
-      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n} | {\n    kind: 'too-large';\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
+      "example": "{\n    kind: 'ok';\n} | {\n    kind: 'retry-later';\n    error: unknown;\n    retryAfterMs?: number;\n} | {\n    kind: 'too-large';\n    measuredLocally?: boolean;\n} | {\n    kind: 'fatal';\n    error: unknown;\n}"
     },
     {
       "id": "SeverityLevel",