fix: drop the oldest AI events while too many wait to be written to disk

.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; 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`

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, 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.

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<PostHogEvent>? = null
+    private var aiQueue: PostHogQueue<PostHogEvent>? = 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 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.
      */
-    private fun isOverAiEventLimit(event: PostHogEvent): Boolean {
+    private fun queueAiEvent(
+        aiQueue: PostHogQueue<PostHogEvent>,
+        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, the AI queue is shut down.")
             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()
                 }
             }

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, the oldest waiting events
+     * are dropped. A value of 0 or less uses the default.
      *
      * Default: 100.
      */

posthog/src/main/java/com/posthog/internal/PostHogQueue.kt

@@ -42,6 +42,15 @@ public class PostHogQueue<Record>(
 
     // The size each file was counted with, so a file deleted from the cache dir is still subtracted in full.
     private val fileBytes = HashMap<File, Long>()
+
+    // 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 val pending: ArrayDeque<Pair<Record, Long>> = 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
@@ -231,6 +240,65 @@ public class PostHogQueue<Record>(
         }
     }
 
+    /**
+     * 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,
+        bytes: Long,
+    ): Boolean {
+        val maxQueueBytes = spec.maxQueueBytes
+        if (maxQueueBytes == null) {
+            add(record)
+            return true
+        }
+        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
+            while (pending.size > 1 && (pending.size > maxRecords || pendingBytes > maxQueueBytes)) {
+                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 {
+            // Each task writes the oldest waiting record.
+            executor.execute {
+                val next =
+                    synchronized(pendingLock) {
+                        pendingTasks--
+                        pending.removeFirstOrNull()?.also { pendingBytes -= it.second }
+                    }
+                if (next != null) {
+                    flushRecordSync(next.first)
+                }
+            }
+        } catch (e: Throwable) {
+            // RejectedExecutionException once the queue is shut down
+            synchronized(pendingLock) {
+                pendingTasks--
+                if (pending.remove(entry)) {
+                    pendingBytes -= bytes
+                }
+            }
+            return false
+        }
+        return true
+    }
+
     private fun flushIfOverThreshold(isFatal: Boolean) {
         if (isAboveThreshold(batchLimits.flushAt)) {
             flushBatch(isFatal)
@@ -702,6 +770,10 @@ public class PostHogQueue<Record>(
     internal val queuedBytesForTesting: Long
         @PostHogVisibleForTesting
         get() = synchronized(dequeLock) { queuedBytes }
+
+    internal val pendingTasksForTesting: Int
+        @PostHogVisibleForTesting
+        get() = synchronized(pendingLock) { pendingTasks }
 }
 
 internal class BatchLimits(

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,22 @@ internal class PostHogCaptureAiTest {
     @Suppress("UNCHECKED_CAST")
     private fun Map<String, Any?>.properties(): Map<String, Any?> = this["properties"] as Map<String, Any?>
 
+    @Test
+    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)
+        aiExecutor.execute { release.await() }
+
+        assertNotNull(sut.captureAi("\$ai_generation"))
+        assertNotNull(sut.captureAi("\$ai_generation"))
+        assertNotNull(sut.captureAi("\$ai_generation"))
+        assertTrue(logger.messages.any { it.contains("the oldest 1 are dropped") })
+
+        release.countDown()
+        aiExecutor.awaitExecution()
+    }
+
     @Test
     fun `captureAi posts to the AI endpoint and returns the uuid on the wire`() {
         val http = httpServers.mockHttp()

posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt

@@ -9,6 +9,7 @@ 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
@@ -117,6 +118,29 @@ internal class PostHogQueueByteLimitsTest {
         assertEquals(0L, sut.queuedBytesForTesting)
     }
 
+    @Test
+    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: a is dropped
+        assertTrue(sut.addMeasured(record("c", 40), 40))
+        assertTrue(sut.addMeasured(record("d", 20), 20))
+        // 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") })
+        // one executor task per waiting record, not per add
+        assertEquals(3, sut.pendingTasksForTesting)
+        release.countDown()
+        executor.awaitExecution()
+
+        assertEquals(listOf("c", "d", "e"), queuedIds(sut))
+    }
+
     @Test
     fun `a record over the byte limit is sent alone and keeps its place`() {
         val sut = queue(maxBatchBytes = 100)