diff --git a/.changeset/android-v4-major.md b/.changeset/android-v4-major.md index 47de85ba5..703e9280d 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; 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 9ed5b3d8d..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, 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. diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt index f82e7abb7..31dbefd32 100644 --- a/posthog/src/main/java/com/posthog/PostHog.kt +++ b/posthog/src/main/java/com/posthog/PostHog.kt @@ -160,7 +160,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 @@ -1189,18 +1189,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, + 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 } @@ -1433,12 +1441,18 @@ 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 - onQueued() + // 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) { + onQueued() + } } // Notify surveys integration about the event surveysHandler?.onEvent(event, mergedProperties) diff --git a/posthog/src/main/java/com/posthog/PostHogConfig.kt b/posthog/src/main/java/com/posthog/PostHogConfig.kt index bab75d8c1..dfe5908ed 100644 --- a/posthog/src/main/java/com/posthog/PostHogConfig.kt +++ b/posthog/src/main/java/com/posthog/PostHogConfig.kt @@ -438,7 +438,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. */ diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt index fd1213473..86e015253 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt @@ -42,6 +42,15 @@ 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 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> = 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( } } + /** + * 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 accepted = + 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. + if (pendingTasks >= pending.size) return@synchronized true + pendingTasks++ + // Submitted under the lock so tasks queue in the order their records were added, and a + // flush queued after this call returns runs after this record is written. + 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) + } + } + true + } catch (e: Throwable) { + // RejectedExecutionException once the queue is shut down + pendingTasks-- + if (pending.remove(entry)) { + pendingBytes -= bytes + } + false + } + } + if (dropped > 0) { + config.logger.log("Too many ${spec.recordsLabel} are waiting to be written, the oldest $dropped are dropped.") + } + return accepted + } + private fun flushIfOverThreshold(isFatal: Boolean) { if (isAboveThreshold(batchLimits.flushAt)) { flushBatch(isFatal) @@ -702,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/PostHogCaptureAiTest.kt b/posthog/src/test/java/com/posthog/PostHogCaptureAiTest.kt index 938316749..dbc158f2d 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,22 @@ internal class PostHogCaptureAiTest { @Suppress("UNCHECKED_CAST") private fun Map.properties(): Map = this["properties"] as Map + @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() diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt index c3023d1b7..0630c616c 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogQueueByteLimitsTest.kt +++ b/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)