From c3f94922583b6b985b22e1609ad7a81896be05e0 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 28 Oct 2026 18:39:20 -0400 Subject: [PATCH 1/3] fix: drop an AI event while too many wait to be written to disk One executor both writes and sends AI events, so during a slow upload new multi-MiB events waited in memory with no limit. captureAi now returns null while captureAiMaxQueueSize events or 50 MiB are waiting, the same limits the disk queue keeps. Co-Authored-By: Claude Opus 5.5 --- .changeset/android-v4-major.md | 2 +- docs/migration-3.x-to-4.0.md | 2 +- posthog/src/main/java/com/posthog/PostHog.kt | 38 ++++++++----- .../main/java/com/posthog/PostHogConfig.kt | 4 +- .../main/java/com/posthog/PostHogInterface.kt | 3 +- .../java/com/posthog/internal/PostHogQueue.kt | 53 ++++++++++++++++++- .../java/com/posthog/PostHogCaptureAiTest.kt | 18 +++++++ .../internal/PostHogQueueByteLimitsTest.kt | 24 +++++++++ 8 files changed, 126 insertions(+), 18 deletions(-) diff --git a/.changeset/android-v4-major.md b/.changeset/android-v4-major.md index 4bc23ca33..891e1da0f 100644 --- a/.changeset/android-v4-major.md +++ b/.changeset/android-v4-major.md @@ -26,7 +26,7 @@ - **Breaking:** registered super properties now override the properties the SDK sets itself, such as `$os_name`; super and event properties also override `$is_identified`, `$process_person_profile` and the session replay debug properties, which the SDK no longer re-applies after the caller's values - **Breaking:** the SDK sets the `process_person_profile` option to `false` only when person processing is off, and no longer adds a `$process_person_profile` property to every event - **Breaking:** keep the uuid of its own capture on the event `beforeSend` returns, so a uuid set in `beforeSend`, for example with `copy(uuid = …)`, is replaced -- **Breaking:** add `captureAi` to send AI events such as `$ai_generation` to PostHog's AI endpoint (`/i/v1/ai/events`), which accepts events whose properties are up to 8 MiB, on its own disk queue; it returns the event uuid, or `null` when the event isn't queued, and classes that implement `PostHogInterface` add the method +- **Breaking:** add `captureAi` to send AI events such as `$ai_generation` to PostHog's AI endpoint (`/i/v1/ai/events`), which accepts events whose properties are up to 8 MiB, on its own disk queue; it returns the event uuid, or `null` when the event isn't queued, including while `captureAiMaxQueueSize` AI events or 50 MiB are still waiting to be written to disk, and classes that implement `PostHogInterface` add the method - use the current distinct ID when the `distinctId` argument to `capture` is empty or whitespace-only, instead of dropping the event - **Breaking:** skip deep link query parameters whose name starts with `$` in the `Deep Link Opened` event, so a link can't set a PostHog property such as `$process_person_profile` - stop storing or matching the person-properties deduplication hash for a `$set` sent with `process_person_profile: false` diff --git a/docs/migration-3.x-to-4.0.md b/docs/migration-3.x-to-4.0.md index 9ed5b3d8d..883dccd6a 100644 --- a/docs/migration-3.x-to-4.0.md +++ b/docs/migration-3.x-to-4.0.md @@ -164,7 +164,7 @@ PostHog.captureAi( ) ``` -- `captureAi` returns the event's uuid, or `null` when the event isn't queued: the SDK is disabled or opted out, `beforeSend` dropped it, or it's over the size limit. +- `captureAi` returns the event's uuid, or `null` when the event isn't queued: the SDK is disabled or opted out, `beforeSend` dropped it, it's over the size limit, or 100 AI events (`captureAiMaxQueueSize`) or 50 MiB are still waiting to be written to disk, for example during a slow upload. - An AI event's properties can be up to 8 MiB. The SDK drops a serialized event over 8 MiB + 64 KiB, measured before `encryption`, and logs the event name and size, never its content. Set `captureAiMaxEventBytes` to lower that limit; a larger value uses 8 MiB + 64 KiB and logs a warning. - PostHog drops an event sent with `captureAi` whose name doesn't start with `$ai_`. `capture` never sends events to the AI endpoint, even ones named `$ai_*`. - AI requests time out after 60 seconds without network activity, or `captureAiTimeout`. Other requests keep OkHttp's 10-second defaults. diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt index ea0a756d9..d91e55675 100644 --- a/posthog/src/main/java/com/posthog/PostHog.kt +++ b/posthog/src/main/java/com/posthog/PostHog.kt @@ -135,7 +135,7 @@ public class PostHog private constructor( // The AI lane never shares a queue, executor or retry state with analytics events. @Volatile - private var aiQueue: PostHogQueueInterface? = null + private var aiQueue: PostHogQueue? = null private var pushSubscriptionManager: PostHogPushSubscriptionManager? = null @@ -1070,18 +1070,26 @@ public class PostHog private constructor( } /** - * Whether [event], serialized and before [PostHogConfig.encryption], is over the AI event limit. - * Runs on the caller's thread so captureAi can return null; writing stops at the limit, but the - * properties JSON tree is built in full first. Logs the name and size only: AI properties can - * hold prompts and media. + * Hands [event] to [aiQueue] unless, serialized and before [PostHogConfig.encryption], it's over + * the AI event limit or too many AI events are still waiting to be written. Runs on the caller's + * thread so captureAi can return null; writing stops at the limit, but the properties JSON tree is + * built in full first. Logs the name and size only: AI properties can hold prompts and media. */ - private fun isOverAiEventLimit(event: PostHogEvent): Boolean { + private fun queueAiEvent( + aiQueue: PostHogQueue, + event: PostHogEvent, + ): Boolean { val config = config ?: return false val limit = config.captureAiMaxStoredEventBytes - if (config.serializedSize(event, stopAfter = limit) <= limit) { + val bytes = config.serializedSize(event, stopAfter = limit) + if (bytes > limit) { + config.logger.log("Dropping AI event ${event.event} of more than $limit bytes, over the limit.") + return false + } + if (!aiQueue.addMeasured(event, bytes)) { + config.logger.log("Dropping AI event ${event.event} of $bytes bytes, too many AI events are waiting to be written.") return false } - config.logger.log("Dropping AI event ${event.event} of more than $limit bytes, over the limit.") return true } @@ -1216,12 +1224,16 @@ public class PostHog private constructor( // beforeSend may have opted out if (isOptedOut()) { config?.logger?.log("PostHog is in OptOut state.") - } else if (aiLane && isOverAiEventLimit(postHogEvent)) { - // dropped here rather than in the queue, so captureAi returns null } else if (targetQueue != null) { - targetQueue.add(postHogEvent) - queued = true - if (claimedReplayDebugBundle) { + // AI events are measured and admitted here rather than in the queue, so captureAi returns null + queued = + if (aiLane) { + queueAiEvent(aiQueue ?: return null, postHogEvent) + } else { + targetQueue.add(postHogEvent) + true + } + if (queued && claimedReplayDebugBundle) { commitReplayDebugPropertiesWindow() } } diff --git a/posthog/src/main/java/com/posthog/PostHogConfig.kt b/posthog/src/main/java/com/posthog/PostHogConfig.kt index 6c47c6096..c3248cb5a 100644 --- a/posthog/src/main/java/com/posthog/PostHogConfig.kt +++ b/posthog/src/main/java/com/posthog/PostHogConfig.kt @@ -437,7 +437,9 @@ public open class PostHogConfig( * are dropped. * * AI events can be megabytes, so this is lower than [maxQueueSize]. The queue also keeps at most - * 50 MiB of AI events, whatever this is set to. A value of 0 or less uses the default. + * 50 MiB of AI events, whatever this is set to. The same limits apply to AI events still waiting + * to be written to disk, for example during a slow upload: over them, [PostHog.captureAi] drops + * the new event and returns null. A value of 0 or less uses the default. * * Default: 100. */ diff --git a/posthog/src/main/java/com/posthog/PostHogInterface.kt b/posthog/src/main/java/com/posthog/PostHogInterface.kt index 108335003..21cd10b32 100644 --- a/posthog/src/main/java/com/posthog/PostHogInterface.kt +++ b/posthog/src/main/java/com/posthog/PostHogInterface.kt @@ -139,7 +139,8 @@ public interface PostHogInterface : PostHogCoreInterface { * @param timestamp the event timestamp override. UTC is preferred; the equivalent instant is serialized in UTC. If omitted, the current time is used * @param options settings that tell PostHog how to process this event, see `options` in [capture] * @return the event's uuid once it's handed to the queue, or null when it isn't: the SDK is - * disabled or opted out, `beforeSend` dropped it, or it's over the size limit + * disabled or opted out, `beforeSend` dropped it, it's over the size limit, or + * [PostHogConfig.captureAiMaxQueueSize] events or 50 MiB are still waiting to be written to disk */ public fun captureAi( event: String, diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt index fd1213473..f31040802 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt @@ -42,6 +42,11 @@ public class PostHogQueue( // The size each file was counted with, so a file deleted from the cache dir is still subtracted in full. private val fileBytes = HashMap() + + // Records handed to the executor by addMeasured and not yet written to disk. Guarded by pendingLock. + private val pendingLock = Any() + private var pendingRecords = 0 + private var pendingBytes = 0L private val timerLock = Any() @PostHogVisibleForTesting @@ -209,9 +214,16 @@ public class PostHogQueue( private fun flushRecordSync( record: Record, isFatal: Boolean = false, + onWritten: () -> Unit = {}, ) { ensureCachedRecordsLoaded() - if (addRecordSync(record)) { + val added = + try { + addRecordSync(record) + } finally { + onWritten() + } + if (added) { removeRecordSync() // this is best effort since we dont know if theres // enough time to flush records to the wire @@ -231,6 +243,45 @@ public class PostHogQueue( } } + /** + * Adds [record], [bytes] long, unless the records still waiting to be written already hold + * [EndpointSpec.maxQueueSize] records or would go over [EndpointSpec.maxQueueBytes]. The disk keeps + * no more than that, so waiting in memory with more can't deliver more, while a slow send on the + * executor would let large records pile up. Returns whether the record was handed to the executor. + */ + internal fun addMeasured( + record: Record, + bytes: Long, + ): Boolean { + val maxQueueBytes = spec.maxQueueBytes + if (maxQueueBytes == null) { + add(record) + return true + } + val maxRecords = spec.maxQueueSize(config).coerceAtLeast(1) + synchronized(pendingLock) { + if (pendingRecords >= maxRecords || pendingBytes + bytes > maxQueueBytes) { + return false + } + pendingRecords++ + pendingBytes += bytes + } + val written = { + synchronized(pendingLock) { + pendingRecords-- + pendingBytes -= bytes + } + } + try { + executor.execute { flushRecordSync(record, onWritten = written) } + } catch (e: Throwable) { + // RejectedExecutionException once the queue is shut down + written() + return false + } + return true + } + private fun flushIfOverThreshold(isFatal: Boolean) { if (isAboveThreshold(batchLimits.flushAt)) { flushBatch(isFatal) diff --git a/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt b/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt index 938316749..cbae97439 100644 --- a/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt +++ b/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt @@ -12,6 +12,7 @@ import java.io.FilterInputStream import java.io.FilterOutputStream import java.io.InputStream import java.io.OutputStream +import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit import kotlin.test.AfterTest @@ -87,6 +88,23 @@ internal class PostHogCaptureAiTest { @Suppress("UNCHECKED_CAST") private fun Map.properties(): Map = this["properties"] as Map + @Test + fun `captureAi returns null while captureAiMaxQueueSize events wait to be written`() { + val http = httpServers.mockHttp(total = 3) + val sut = getSut(http) { captureAiMaxQueueSize = 2 } + val release = CountDownLatch(1) + aiExecutor.execute { release.await() } + + assertNotNull(sut.captureAi("\$ai_generation")) + assertNotNull(sut.captureAi("\$ai_generation")) + assertNull(sut.captureAi("\$ai_generation")) + assertTrue(logger.messages.any { it.contains("too many AI events are waiting to be written") }) + + release.countDown() + aiExecutor.awaitExecution() + assertNotNull(sut.captureAi("\$ai_generation")) + } + @Test fun `captureAi posts to the AI endpoint and returns the uuid on the wire`() { val http = httpServers.mockHttp() diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt index c3023d1b7..f51e7d090 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt @@ -9,10 +9,12 @@ import org.junit.Rule import org.junit.rules.TemporaryFolder import java.io.File import java.util.UUID +import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import kotlin.test.AfterTest import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertTrue /** @@ -117,6 +119,28 @@ internal class PostHogQueueByteLimitsTest { assertEquals(0L, sut.queuedBytesForTesting) } + @Test + fun `addMeasured refuses a record while those waiting to be written are at the limits`() { + val sut = queue(maxQueueBytes = 100, maxQueueSize = 3) + val release = CountDownLatch(1) + executor.execute { release.await() } + + assertTrue(sut.addMeasured(record("a", 40), 40)) + assertTrue(sut.addMeasured(record("b", 40), 40)) + // 120 bytes waiting + assertFalse(sut.addMeasured(record("c", 40), 40)) + assertTrue(sut.addMeasured(record("d", 20), 20)) + // 3 records waiting + assertFalse(sut.addMeasured(record("e", 1), 1)) + release.countDown() + executor.awaitExecution() + + assertTrue(sut.addMeasured(record("f", 40), 40)) + executor.awaitExecution() + // the disk limits then evict the oldest, as for any record + assertEquals(listOf("b", "d", "f"), queuedIds(sut)) + } + @Test fun `a record over the byte limit is sent alone and keeps its place`() { val sut = queue(maxBatchBytes = 100) From 80759d29d134a08135a4b394c5c2d78f5c14807f Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 28 Oct 2026 19:46:24 -0400 Subject: [PATCH 2/3] fix: drop the oldest waiting AI events instead of the new one While a slow upload holds the AI executor, captureAi now keeps the new event and drops the oldest events still waiting to be written once captureAiMaxQueueSize events or 50 MiB wait, the same as the disk queue already does. captureAi returns null again only for disabled, opted-out, beforeSend-dropped or oversized events. Co-Authored-By: Claude Opus 5.5 --- .changeset/android-v4-major.md | 2 +- docs/migration-3.x-to-4.0.md | 2 +- posthog/src/main/java/com/posthog/PostHog.kt | 4 +- .../main/java/com/posthog/PostHogConfig.kt | 4 +- .../main/java/com/posthog/PostHogInterface.kt | 3 +- .../java/com/posthog/internal/PostHogQueue.kt | 57 +++++++++++-------- .../java/com/posthog/PostHogCaptureAiTest.kt | 7 +-- .../internal/PostHogQueueByteLimitsTest.kt | 18 +++--- 8 files changed, 50 insertions(+), 47 deletions(-) diff --git a/.changeset/android-v4-major.md b/.changeset/android-v4-major.md index 891e1da0f..0e560342f 100644 --- a/.changeset/android-v4-major.md +++ b/.changeset/android-v4-major.md @@ -26,7 +26,7 @@ - **Breaking:** registered super properties now override the properties the SDK sets itself, such as `$os_name`; super and event properties also override `$is_identified`, `$process_person_profile` and the session replay debug properties, which the SDK no longer re-applies after the caller's values - **Breaking:** the SDK sets the `process_person_profile` option to `false` only when person processing is off, and no longer adds a `$process_person_profile` property to every event - **Breaking:** keep the uuid of its own capture on the event `beforeSend` returns, so a uuid set in `beforeSend`, for example with `copy(uuid = …)`, is replaced -- **Breaking:** add `captureAi` to send AI events such as `$ai_generation` to PostHog's AI endpoint (`/i/v1/ai/events`), which accepts events whose properties are up to 8 MiB, on its own disk queue; it returns the event uuid, or `null` when the event isn't queued, including while `captureAiMaxQueueSize` AI events or 50 MiB are still waiting to be written to disk, and classes that implement `PostHogInterface` add the method +- **Breaking:** add `captureAi` to send AI events such as `$ai_generation` to PostHog's AI endpoint (`/i/v1/ai/events`), which accepts events whose properties are up to 8 MiB, on its own disk queue; it returns the event uuid, or `null` when the event isn't queued; past `captureAiMaxQueueSize` AI events or 50 MiB, including events still waiting to be written to disk, the oldest are dropped, and classes that implement `PostHogInterface` add the method - use the current distinct ID when the `distinctId` argument to `capture` is empty or whitespace-only, instead of dropping the event - **Breaking:** skip deep link query parameters whose name starts with `$` in the `Deep Link Opened` event, so a link can't set a PostHog property such as `$process_person_profile` - stop storing or matching the person-properties deduplication hash for a `$set` sent with `process_person_profile: false` diff --git a/docs/migration-3.x-to-4.0.md b/docs/migration-3.x-to-4.0.md index 883dccd6a..577679f7e 100644 --- a/docs/migration-3.x-to-4.0.md +++ b/docs/migration-3.x-to-4.0.md @@ -164,7 +164,7 @@ PostHog.captureAi( ) ``` -- `captureAi` returns the event's uuid, or `null` when the event isn't queued: the SDK is disabled or opted out, `beforeSend` dropped it, it's over the size limit, or 100 AI events (`captureAiMaxQueueSize`) or 50 MiB are still waiting to be written to disk, for example during a slow upload. +- `captureAi` returns the event's uuid, or `null` when the event isn't queued: the SDK is disabled or opted out, `beforeSend` dropped it, or it's over the size limit. Once 100 AI events (`captureAiMaxQueueSize`) or 50 MiB are queued, including events still waiting to be written to disk during a slow upload, the oldest are dropped. - An AI event's properties can be up to 8 MiB. The SDK drops a serialized event over 8 MiB + 64 KiB, measured before `encryption`, and logs the event name and size, never its content. Set `captureAiMaxEventBytes` to lower that limit; a larger value uses 8 MiB + 64 KiB and logs a warning. - PostHog drops an event sent with `captureAi` whose name doesn't start with `$ai_`. `capture` never sends events to the AI endpoint, even ones named `$ai_*`. - AI requests time out after 60 seconds without network activity, or `captureAiTimeout`. Other requests keep OkHttp's 10-second defaults. diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt index d91e55675..20d8fafa7 100644 --- a/posthog/src/main/java/com/posthog/PostHog.kt +++ b/posthog/src/main/java/com/posthog/PostHog.kt @@ -1071,7 +1071,7 @@ public class PostHog private constructor( /** * Hands [event] to [aiQueue] unless, serialized and before [PostHogConfig.encryption], it's over - * the AI event limit or too many AI events are still waiting to be written. Runs on the caller's + * the AI event limit or the queue is shut down. Runs on the caller's * thread so captureAi can return null; writing stops at the limit, but the properties JSON tree is * built in full first. Logs the name and size only: AI properties can hold prompts and media. */ @@ -1087,7 +1087,7 @@ public class PostHog private constructor( return false } if (!aiQueue.addMeasured(event, bytes)) { - config.logger.log("Dropping AI event ${event.event} of $bytes bytes, too many AI events are waiting to be written.") + config.logger.log("Dropping AI event ${event.event} of $bytes bytes, the AI queue is shut down.") return false } return true diff --git a/posthog/src/main/java/com/posthog/PostHogConfig.kt b/posthog/src/main/java/com/posthog/PostHogConfig.kt index c3248cb5a..734e0cb15 100644 --- a/posthog/src/main/java/com/posthog/PostHogConfig.kt +++ b/posthog/src/main/java/com/posthog/PostHogConfig.kt @@ -438,8 +438,8 @@ public open class PostHogConfig( * * AI events can be megabytes, so this is lower than [maxQueueSize]. The queue also keeps at most * 50 MiB of AI events, whatever this is set to. The same limits apply to AI events still waiting - * to be written to disk, for example during a slow upload: over them, [PostHog.captureAi] drops - * the new event and returns null. A value of 0 or less uses the default. + * to be written to disk, for example during a slow upload: over them, the oldest waiting events + * are dropped. A value of 0 or less uses the default. * * Default: 100. */ diff --git a/posthog/src/main/java/com/posthog/PostHogInterface.kt b/posthog/src/main/java/com/posthog/PostHogInterface.kt index 21cd10b32..108335003 100644 --- a/posthog/src/main/java/com/posthog/PostHogInterface.kt +++ b/posthog/src/main/java/com/posthog/PostHogInterface.kt @@ -139,8 +139,7 @@ public interface PostHogInterface : PostHogCoreInterface { * @param timestamp the event timestamp override. UTC is preferred; the equivalent instant is serialized in UTC. If omitted, the current time is used * @param options settings that tell PostHog how to process this event, see `options` in [capture] * @return the event's uuid once it's handed to the queue, or null when it isn't: the SDK is - * disabled or opted out, `beforeSend` dropped it, it's over the size limit, or - * [PostHogConfig.captureAiMaxQueueSize] events or 50 MiB are still waiting to be written to disk + * disabled or opted out, `beforeSend` dropped it, or it's over the size limit */ public fun captureAi( event: String, diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt index f31040802..55ea9219b 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt @@ -43,9 +43,10 @@ public class PostHogQueue( // The size each file was counted with, so a file deleted from the cache dir is still subtracted in full. private val fileBytes = HashMap() - // Records handed to the executor by addMeasured and not yet written to disk. Guarded by pendingLock. + // Records added by addMeasured and not yet taken by the executor to be written to disk, oldest + // first, with their sizes. Guarded by pendingLock. private val pendingLock = Any() - private var pendingRecords = 0 + private val pending: ArrayDeque> = ArrayDeque() private var pendingBytes = 0L private val timerLock = Any() @@ -214,16 +215,9 @@ public class PostHogQueue( private fun flushRecordSync( record: Record, isFatal: Boolean = false, - onWritten: () -> Unit = {}, ) { ensureCachedRecordsLoaded() - val added = - try { - addRecordSync(record) - } finally { - onWritten() - } - if (added) { + if (addRecordSync(record)) { removeRecordSync() // this is best effort since we dont know if theres // enough time to flush records to the wire @@ -244,10 +238,10 @@ public class PostHogQueue( } /** - * Adds [record], [bytes] long, unless the records still waiting to be written already hold - * [EndpointSpec.maxQueueSize] records or would go over [EndpointSpec.maxQueueBytes]. The disk keeps - * no more than that, so waiting in memory with more can't deliver more, while a slow send on the - * executor would let large records pile up. Returns whether the record was handed to the executor. + * Adds [record], [bytes] long. While a slow send holds the executor, records wait in memory before + * they're written; once more than [EndpointSpec.maxQueueSize] records or [EndpointSpec.maxQueueBytes] + * wait, the oldest waiting ones are dropped, as the disk queue drops its oldest. The newest record + * is always kept. Returns false only when the executor no longer accepts work. */ internal fun addMeasured( record: Record, @@ -259,24 +253,37 @@ public class PostHogQueue( return true } val maxRecords = spec.maxQueueSize(config).coerceAtLeast(1) + val entry = record to bytes + var dropped = 0 synchronized(pendingLock) { - if (pendingRecords >= maxRecords || pendingBytes + bytes > maxQueueBytes) { - return false - } - pendingRecords++ + pending.addLast(entry) pendingBytes += bytes - } - val written = { - synchronized(pendingLock) { - pendingRecords-- - pendingBytes -= bytes + while (pending.size > 1 && (pending.size > maxRecords || pendingBytes > maxQueueBytes)) { + pendingBytes -= pending.removeFirst().second + dropped++ } } + if (dropped > 0) { + config.logger.log("Too many ${spec.recordsLabel} are waiting to be written, the oldest $dropped are dropped.") + } try { - executor.execute { flushRecordSync(record, onWritten = written) } + // One task per record; a task whose record was dropped finds an earlier-added one or nothing. + executor.execute { + val next = + synchronized(pendingLock) { + pending.removeFirstOrNull()?.also { pendingBytes -= it.second } + } + if (next != null) { + flushRecordSync(next.first) + } + } } catch (e: Throwable) { // RejectedExecutionException once the queue is shut down - written() + synchronized(pendingLock) { + if (pending.remove(entry)) { + pendingBytes -= bytes + } + } return false } return true diff --git a/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt b/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt index cbae97439..dbc158f2d 100644 --- a/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt +++ b/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt @@ -89,7 +89,7 @@ internal class PostHogCaptureAiTest { private fun Map.properties(): Map = this["properties"] as Map @Test - fun `captureAi returns null while captureAiMaxQueueSize events wait to be written`() { + fun `captureAi keeps the newest events while captureAiMaxQueueSize events wait to be written`() { val http = httpServers.mockHttp(total = 3) val sut = getSut(http) { captureAiMaxQueueSize = 2 } val release = CountDownLatch(1) @@ -97,12 +97,11 @@ internal class PostHogCaptureAiTest { assertNotNull(sut.captureAi("\$ai_generation")) assertNotNull(sut.captureAi("\$ai_generation")) - assertNull(sut.captureAi("\$ai_generation")) - assertTrue(logger.messages.any { it.contains("too many AI events are waiting to be written") }) + assertNotNull(sut.captureAi("\$ai_generation")) + assertTrue(logger.messages.any { it.contains("the oldest 1 are dropped") }) release.countDown() aiExecutor.awaitExecution() - assertNotNull(sut.captureAi("\$ai_generation")) } @Test diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt index f51e7d090..a01efcd42 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt @@ -14,7 +14,6 @@ import java.util.concurrent.Executors import kotlin.test.AfterTest import kotlin.test.Test import kotlin.test.assertEquals -import kotlin.test.assertFalse import kotlin.test.assertTrue /** @@ -120,25 +119,24 @@ internal class PostHogQueueByteLimitsTest { } @Test - fun `addMeasured refuses a record while those waiting to be written are at the limits`() { + fun `addMeasured drops the oldest waiting records once those waiting to be written are over the limits`() { val sut = queue(maxQueueBytes = 100, maxQueueSize = 3) val release = CountDownLatch(1) executor.execute { release.await() } assertTrue(sut.addMeasured(record("a", 40), 40)) assertTrue(sut.addMeasured(record("b", 40), 40)) - // 120 bytes waiting - assertFalse(sut.addMeasured(record("c", 40), 40)) + // 120 bytes waiting: a is dropped + assertTrue(sut.addMeasured(record("c", 40), 40)) assertTrue(sut.addMeasured(record("d", 20), 20)) - // 3 records waiting - assertFalse(sut.addMeasured(record("e", 1), 1)) + // 4 records waiting: b is dropped + assertTrue(sut.addMeasured(record("e", 1), 1)) + // dropped while waiting, before the disk limits ever see them + assertEquals(2, logger.messages.count { it.contains("blobs are waiting to be written, the oldest 1 are dropped") }) release.countDown() executor.awaitExecution() - assertTrue(sut.addMeasured(record("f", 40), 40)) - executor.awaitExecution() - // the disk limits then evict the oldest, as for any record - assertEquals(listOf("b", "d", "f"), queuedIds(sut)) + assertEquals(listOf("c", "d", "e"), queuedIds(sut)) } @Test From cf53ab69a865a88c3d261491bb43cc212b569a81 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 28 Oct 2026 21:18:11 -0400 Subject: [PATCH 3/3] fix: queue no more AI write tasks than events waiting to be written Dropping a waiting AI event no longer leaves its executor task behind for nothing: a dropped event's task writes a newer one instead, so the executor never holds more tasks than events are waiting. Co-Authored-By: Claude Opus 5.5 --- .../java/com/posthog/internal/PostHogQueue.kt | 16 +++++++++++++++- .../internal/PostHogQueueByteLimitsTest.kt | 2 ++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt index 55ea9219b..25feffd00 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt @@ -48,6 +48,9 @@ public class PostHogQueue( private val pendingLock = Any() private val pending: ArrayDeque> = ArrayDeque() private var pendingBytes = 0L + + // Executor tasks queued to write pending records, never more than pending holds. Guarded by pendingLock. + private var pendingTasks = 0 private val timerLock = Any() @PostHogVisibleForTesting @@ -255,6 +258,7 @@ public class PostHogQueue( val maxRecords = spec.maxQueueSize(config).coerceAtLeast(1) val entry = record to bytes var dropped = 0 + val schedule: Boolean synchronized(pendingLock) { pending.addLast(entry) pendingBytes += bytes @@ -262,15 +266,20 @@ public class PostHogQueue( pendingBytes -= pending.removeFirst().second dropped++ } + // A dropped record's task writes a newer one instead, so tasks never outnumber records. + schedule = pendingTasks < pending.size + if (schedule) pendingTasks++ } if (dropped > 0) { config.logger.log("Too many ${spec.recordsLabel} are waiting to be written, the oldest $dropped are dropped.") } + if (!schedule) return true try { - // One task per record; a task whose record was dropped finds an earlier-added one or nothing. + // Each task writes the oldest waiting record. executor.execute { val next = synchronized(pendingLock) { + pendingTasks-- pending.removeFirstOrNull()?.also { pendingBytes -= it.second } } if (next != null) { @@ -280,6 +289,7 @@ public class PostHogQueue( } catch (e: Throwable) { // RejectedExecutionException once the queue is shut down synchronized(pendingLock) { + pendingTasks-- if (pending.remove(entry)) { pendingBytes -= bytes } @@ -760,6 +770,10 @@ public class PostHogQueue( internal val queuedBytesForTesting: Long @PostHogVisibleForTesting get() = synchronized(dequeLock) { queuedBytes } + + internal val pendingTasksForTesting: Int + @PostHogVisibleForTesting + get() = synchronized(pendingLock) { pendingTasks } } internal class BatchLimits( diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt index a01efcd42..0630c616c 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt @@ -133,6 +133,8 @@ internal class PostHogQueueByteLimitsTest { assertTrue(sut.addMeasured(record("e", 1), 1)) // dropped while waiting, before the disk limits ever see them assertEquals(2, logger.messages.count { it.contains("blobs are waiting to be written, the oldest 1 are dropped") }) + // one executor task per waiting record, not per add + assertEquals(3, sut.pendingTasksForTesting) release.countDown() executor.awaitExecution()