diff --git a/.changeset/android-v4-major.md b/.changeset/android-v4-major.md index d23ecc864..2a74fb04b 100644 --- a/.changeset/android-v4-major.md +++ b/.changeset/android-v4-major.md @@ -21,7 +21,7 @@ - **Breaking:** rename `PostHogConfig.maxRetries` to `pushNotificationSubscriptionMaxRetries`, because it only limits push-notification subscription retries - **Breaking:** set `$lib` and `$lib_version` from the `PostHog-Sdk-Info` request header; a value set in properties, super properties or `beforeSend` is removed before sending - **Breaking:** send `$session_id` and `$window_id` as top-level `session_id` and `window_id` fields; a non-string value is dropped, with a log when `debug` is on -- **Breaking:** disable the SDK at setup when `projectToken` is a personal API key (`phx_...`), as for an empty token +- **Breaking:** disable the SDK at setup when `projectToken` is a personal API key (`phx_...`), as for an empty token, and print a warning for both even when `debug` is off - **Breaking:** add an `options` argument to `capture`, `screen`, `identify`, `setPersonProperties`, `alias`, `group` and `captureException` for per-event capture options such as `cookieless_mode`, sent apart from the properties and editable in `beforeSend` through `PostHogEvent.options`; Java callers pass the trailing argument or `null`, and classes that implement `PostHogInterface` add the parameter - **Breaking:** move `identify(distinctId, userProperties, userPropertiesSetOnce)` from `PostHogCoreInterface` to `PostHogStatelessInterface`; `PostHogInterface` declares `identify` with the trailing `options` parameter - **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 @@ -31,7 +31,9 @@ - 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` -- debug logs for capture requests name the event count and the response status instead of printing the request and response bodies, and name each event PostHog drops with the reason it gives +- debug logs for capture requests name the event count and the response status instead of printing the request and response bodies, and count the events PostHog drops, the duplicate copies it drops and the events it asks to retry, in one line per batch, and log the status when PostHog rejects a whole batch +- **Breaking:** drop an event whose name is empty before sending it, including one `beforeSend` renames to `""` +- **Breaking:** let `userProperties`, `userPropertiesSetOnce` and `groups` win key by key over `$set`, `$set_once` and `$groups` maps passed in the same call's properties, instead of being replaced by them, and merge a `$groups` map in properties into the groups set with `group()` - add `captureAiCompression`, `captureAiMaxQueueSize`, `captureAiTimeout` and `captureAiMaxEventBytes` to `PostHogConfig` to tune AI requests separately from other events; the defaults are gzip, 100 queued events (and at most 50 MiB), 60 seconds and 8 MiB + 64 KiB per serialized event - cap a request at about 5 MiB of events, for analytics and AI events, and send as soon as that much is queued; an event larger than that is sent on its own - Add `onFeatureFlags()` to get notified after every feature flags load (on the main thread on Android), returning a `PostHogFeatureFlagsSubscription` you can `unsubscribe()` diff --git a/.changeset/server-keep-on-feature-flags.md b/.changeset/server-keep-on-feature-flags.md index c610af9dc..57116ce82 100644 --- a/.changeset/server-keep-on-feature-flags.md +++ b/.changeset/server-keep-on-feature-flags.md @@ -4,4 +4,6 @@ - Move `PostHogOnFeatureFlags` into `posthog-server`; its API is unchanged - Bump the `posthog` core dependency to 7.0 -- Disable the client at setup when `projectToken` is a personal API key (`phx_...`), as for an empty token +- Disable the client at setup when `projectToken` is a personal API key (`phx_...`), as for an empty token, and print why for both even when `debug` is off +- Drop events with an empty name before sending +- Merge `userProperties`, `userPropertiesSetOnce` and `groups` key by key over a `$set`, `$set_once` or `$groups` map in `properties` diff --git a/docs/migration-3.x-to-4.0.md b/docs/migration-3.x-to-4.0.md index 527599bbd..da6314251 100644 --- a/docs/migration-3.x-to-4.0.md +++ b/docs/migration-3.x-to-4.0.md @@ -256,7 +256,7 @@ Capture v1 requests send the project token in an `Authorization: Bearer `), every `PostHog-*` header, `Content-Type`, `Content-Encoding` and `User-Agent` through unchanged. PostHog answers `401` without `Authorization` and `400` without the others, and the events are dropped. If your proxy strips other credentials, keep `Authorization` values that start with `Bearer phc_`, or the exact `Bearer ` if your project token doesn't start with `phc_`. - Accept request bodies up to 20 MB. AI requests are often larger than 1 MB, the default limit in nginx and ingress-nginx. A `413` from the proxy makes the SDK halve the batch until it fits, and drops an event that doesn't fit on its own. +- Pass PostHog's response body through unchanged. A 2xx response whose body isn't PostHog's `{"results": {...}}` drops the events instead of retrying them. PostHog's managed reverse proxy already does this. See the [proxy guides](https://posthog.com/docs/advanced/proxy). @@ -284,17 +285,22 @@ If a proxy redirects capture requests to another host, point `host` at the final ## Event validation Capture v1 checks each event on its own and drops only the invalid ones, so the rest of the batch is delivered. -The SDK doesn't check these before sending. PostHog drops the event, and the SDK removes it from the queue. -With `debug` on, the SDK logs the name and uuid of each dropped event and the reason PostHog gives, such as `misrouted_event`, never its properties. + +The SDK drops an event whose name is empty before sending it, for example `capture("")` or one `beforeSend` renames to `""`. +With `debug` on, it logs this once. + +The SDK doesn't check the rest before sending. PostHog drops the event, and the SDK removes it from the queue. +With `debug` on, the SDK logs one line per batch with how many events PostHog dropped, how many duplicate copies it dropped and how many it asked to retry, never the event names or properties. Debug logs for capture requests name the event count and the response status, and no longer print the request and response bodies. -- An event with an empty event name is dropped, for example `capture("")`. - An event with an empty or whitespace-only distinct ID is dropped, for example one `beforeSend` sets to `""`. - An event whose name is over 200 UTF-8 bytes is dropped. Legacy capture had no limit on event name length. - An event whose distinct ID is over 200 UTF-8 bytes is dropped. Legacy capture shortened the ID to 200 characters and kept the event. In 3.x, an empty event name or distinct ID made legacy capture reject the whole batch it was in. +A `NaN` or infinite number in the properties is dropped from the event, and the event is sent without it. + The distinct ID can come from `identify()`, the `distinctId` argument of `capture` or `beforeSend`. If you identify users with a long ID, such as a JWT or a long email address, shorten or hash it before you upgrade. Otherwise every event for that user is dropped. @@ -311,7 +317,7 @@ An event's properties are merged in this order, and later values win: 1. properties the SDK sets itself, such as `$os_name`, `$app_version`, `$screen_name`, `$is_identified`, `$session_id` and the session replay debug properties (`$recording_status`, `$sdk_debug_*`) 2. registered super properties (`register`) -3. the properties you pass to `capture` or another event-sending method; `userProperties`, `userPropertiesSetOnce` and `groups` are placed as `$set`, `$set_once` and `$groups` here +3. the properties you pass to `capture` or another event-sending method; `userProperties`, `userPropertiesSetOnce` and `groups` are placed as `$set`, `$set_once` and `$groups` here, and when you also pass that key in properties they win key by key; a `$groups` map in properties adds to the groups set with `group()` instead of replacing them 4. `beforeSend` In 3.x, the SDK's own values won over super properties, and `$is_identified`, `$process_person_profile` and the session replay debug properties also won over the properties you pass. @@ -424,6 +430,8 @@ On Android, listeners run on the main thread. On the JVM, they run on a backgrou `posthog-server` isn't having a major release, and its API is unchanged: it keeps its own `apiKey`, `getFeatureFlagPayload` and `remoteConfig`, and `com.posthog.PostHogOnFeatureFlags` now ships in `posthog-server` itself. None of the API changes above apply to it. -A personal API key (`phx_...`) in `projectToken` now disables the client at setup, as an empty token does. +A personal API key (`phx_...`) in `projectToken` now disables the client at setup, as an empty token does, and both print why even when `debug` is off. + +Events with an empty name are dropped before sending. `userProperties`, `userPropertiesSetOnce` and `groups` now merge key by key over a `$set`, `$set_once` or `$groups` map in `properties`. Its next release depends on core 7.0, so update it if your project also uses core 7.0 or `posthog-android` 4.0. Older `posthog-server` versions aren't compatible with core 7.0 at runtime. diff --git a/posthog/src/main/java/com/posthog/PostHog.kt b/posthog/src/main/java/com/posthog/PostHog.kt index 2a067889b..766e47378 100644 --- a/posthog/src/main/java/com/posthog/PostHog.kt +++ b/posthog/src/main/java/com/posthog/PostHog.kt @@ -320,14 +320,14 @@ public class PostHog private constructor( if (config.logger is PostHogNoOpLogger) PostHogPrintLogger(config) else config.logger if (config.projectToken.isEmpty()) { - config.logger.log( + config.logger.logWarning( "PostHog SDK is disabled because the project token is required and was empty after trimming whitespace.", ) return } if (config.projectToken.startsWith("phx_")) { - config.logger.log( + config.logger.logWarning( "PostHog SDK is disabled because the project token is a personal API key (phx_...), " + "use your project token (phc_...).", ) @@ -977,9 +977,20 @@ public class PostHog private constructor( } // Caller-supplied properties win over registered super properties and SDK-set values. + val currentGroups = props["\$groups"] as? Map<*, *> properties?.let { props.putAll(it) } + if (appendSharedProps) { + putTypedValuesOverCaller( + props, + properties, + userProperties, + userPropertiesSetOnce, + groups.takeIf { appendGroups }, + currentGroups, + ) + } // only Session replay needs distinct_id also in the props // remove after https://github.com/PostHog/posthog/pull/18954 gets merged @@ -1426,6 +1437,9 @@ public class PostHog private constructor( aiLane: Boolean, onQueued: () -> Unit = {}, ): UUID? { + if (isEmptyNameDropped(postHogEvent)) { + return null + } // Reevaluate if this is a snapshot event because the event might have been updated by the beforeSend hook val isSnapshotEvent = !aiLane && postHogEvent.event == PostHogEventName.SNAPSHOT.event // if this is a $snapshot event and $session_id is missing, don't process then event diff --git a/posthog/src/main/java/com/posthog/PostHogStateless.kt b/posthog/src/main/java/com/posthog/PostHogStateless.kt index eb5673141..8b59710b8 100644 --- a/posthog/src/main/java/com/posthog/PostHogStateless.kt +++ b/posthog/src/main/java/com/posthog/PostHogStateless.kt @@ -19,6 +19,7 @@ import java.util.Date import java.util.UUID import java.util.concurrent.ExecutorService import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicBoolean public open class PostHogStateless protected constructor( private val queueExecutor: ExecutorService = @@ -35,6 +36,7 @@ public open class PostHogStateless protected constructor( protected val setupLock: Any = Any() protected val optOutLock: Any = Any() + private val warnedEmptyEventName = AtomicBoolean(false) private var featureFlagsCalled: PostHogFeatureFlagCalledCache? = null // Read on the capture path from any caller thread without setupLock. @@ -60,14 +62,14 @@ public open class PostHogStateless protected constructor( if (config.logger is PostHogNoOpLogger) PostHogPrintLogger(config) else config.logger if (config.projectToken.isEmpty()) { - config.logger.log( + config.logger.logWarning( "PostHog SDK is disabled because the project token is required and was empty after trimming whitespace.", ) return } if (config.projectToken.startsWith("phx_")) { - config.logger.log( + config.logger.logWarning( "PostHog SDK is disabled because the project token is a personal API key (phx_...), " + "use your project token (phc_...).", ) @@ -216,13 +218,39 @@ public open class PostHogStateless protected constructor( props.putAll(it) } + val currentGroups = props["\$groups"] as? Map<*, *> properties?.let { props.putAll(it) } + putTypedValuesOverCaller(props, properties, userProperties, userPropertiesSetOnce, groups.takeIf { appendGroups }, currentGroups) return props } + /** + * A typed per-call value wins key by key over the same map in the caller's [properties]: + * `userProperties` over `$set`, `userPropertiesSetOnce` over `$set_once` and `groups` over `$groups`. + * A caller `$groups` adds to [currentGroups] (the stored and per-call groups) instead of replacing them. + */ + internal fun putTypedValuesOverCaller( + props: MutableMap, + properties: Map?, + userProperties: Map?, + userPropertiesSetOnce: Map?, + groups: Map?, + currentGroups: Map<*, *>? = null, + ) { + for ((key, typed) in listOf("\$set" to userProperties, "\$set_once" to userPropertiesSetOnce, "\$groups" to groups)) { + val caller = properties?.get(key) as? Map<*, *> ?: continue + val merged = LinkedHashMap() + if (key == "\$groups") currentGroups?.let { merged.putAll(it) } + if (typed == null && merged.isEmpty()) continue + merged.putAll(caller) + typed?.let { merged.putAll(it) } + props[key] = merged + } + } + protected fun mergeGroups(givenGroups: Map?): Map? { val preferences = getPreferences() @@ -294,6 +322,9 @@ public open class PostHogStateless protected constructor( config?.logger?.log(message) return } + if (isEmptyNameDropped(postHogEvent)) { + return + } queue?.add(postHogEvent) } catch (e: Throwable) { @@ -366,6 +397,20 @@ public open class PostHogStateless protected constructor( return eventChecked } + /** + * Capture drops an event with an empty name, so the SDK drops it before queueing, after + * beforeSend too. Logs only the first drop, with no event name. + */ + internal fun isEmptyNameDropped(event: PostHogEvent): Boolean { + if (event.event.isNotEmpty()) { + return false + } + if (warnedEmptyEventName.compareAndSet(false, true)) { + config?.logger?.log("Dropping an event with an empty name.") + } + return true + } + public override fun optIn() { if (!isEnabled()) { return diff --git a/posthog/src/main/java/com/posthog/internal/PostHogApi.kt b/posthog/src/main/java/com/posthog/internal/PostHogApi.kt index c455a7874..08d30483a 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogApi.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogApi.kt @@ -51,6 +51,7 @@ public class PostHogApi( private const val CAPTURE_V1_ANALYTICS_PATH = "/i/v1/analytics/events" private const val CAPTURE_V1_AI_PATH = "/i/v1/ai/events" private const val CAPTURE_V1_MAX_REDIRECTS = 5 + private const val CAPTURE_V1_MAX_PENDING_RETRIES = 16 private const val CAPTURE_V1_MAX_RESULTS_BYTES = 1024L * 1024L } @@ -137,9 +138,22 @@ public class PostHogApi( private val captureV1Lock = Any() - // The identity of the last capture v1 request per endpoint path whose events are still to be - // retried. In memory only: a new request id after an app restart is fine, the server only logs it. - private val captureV1PendingRetry = mutableMapOf() + // The identity of each capture v1 request whose events are still to be retried, keyed by endpoint + // path and uuid set, so a send of other events in between keeps it. Holds the most recent + // CAPTURE_V1_MAX_PENDING_RETRIES: an entry stays when the queue drops its events instead of + // retrying them. In memory only: a new request id after an app restart is fine, the server only logs it. + // A plain map trimmed by hand: a LinkedHashMap subclass gets bridges to Java 8 Map methods missing on API 23. + private val captureV1PendingRetry = LinkedHashMap>, CaptureV1RequestIdentity>() + + private fun putCaptureV1PendingRetry( + key: Pair>, + identity: CaptureV1RequestIdentity, + ) { + captureV1PendingRetry[key] = identity + if (captureV1PendingRetry.size > CAPTURE_V1_MAX_PENDING_RETRIES) { + captureV1PendingRetry.remove(captureV1PendingRetry.keys.first()) + } + } private val theHost: String get() { @@ -167,15 +181,17 @@ public class PostHogApi( * Sends analytics events to capture v1 (`POST /i/v1/analytics/events`). * * Returns the uuids of the events the 2xx response asked to retry, with the `Retry-After` it - * sent; every other event in [events] was delivered or dropped and must not be sent again. A 2xx - * whose per-event results can't be read, including a transfer that failed after the status - * arrived, counts as delivered: the server already stored its events and a resend would store - * them again. + * sent; every other event in [events] was delivered or dropped and must not be sent again. + * + * A 2xx whose body can't be read as `{"results": {...}}`, including a transfer that failed after + * the status arrived, is a terminal failure of the whole batch: it throws a [PostHogApiError] + * with that 2xx status, which the queue doesn't retry. * * Only a 307 or 308 to the origin of [PostHogConfig.host] is followed, at most 5 times, with * the same body and headers; any other redirect is a terminal [PostHogApiError]. * - * Throws [PostHogApiError] for a non-2xx status and [IOException] for a transport failure. + * Throws [PostHogApiError] for a non-2xx status or unreadable results, and [IOException] for a + * transport failure before the status. */ @Throws(PostHogApiError::class, IOException::class) internal fun captureV1(events: List): SendResult = postCaptureV1(CAPTURE_V1_ANALYTICS_PATH, events, captureClient) @@ -246,7 +262,7 @@ public class PostHogApi( val target = response.header("Location")?.let { location -> request.url.resolve(location) } if (redirects >= CAPTURE_V1_MAX_REDIRECTS || target == null || !isConfiguredOrigin(target)) { config.logger.log("Capture did not follow a ${response.code} redirect.") - clearCaptureV1Retry(path) + clearCaptureV1Retry(path, identity) throw PostHogApiError(response.code, response.message, response.body, parseRetryAfter(response)) } // Same origin, so the body and every header are safe to resend. @@ -267,9 +283,9 @@ public class PostHogApi( uuids: Set, ): CaptureV1RequestIdentity { synchronized(captureV1Lock) { - val pending = captureV1PendingRetry[path] + val pending = captureV1PendingRetry[path to uuids] val identity = - if (pending != null && pending.uuids == uuids) { + if (pending != null) { pending.copy(attempt = pending.attempt + 1) } else { CaptureV1RequestIdentity( @@ -279,20 +295,24 @@ public class PostHogApi( createdAt = formatISO8601Date(config.dateProvider.currentDate()), ) } - captureV1PendingRetry[path] = identity + putCaptureV1PendingRetry(path to uuids, identity) return identity } } - private fun clearCaptureV1Retry(path: String) { + private fun clearCaptureV1Retry( + path: String, + identity: CaptureV1RequestIdentity, + ) { synchronized(captureV1Lock) { - captureV1PendingRetry.remove(path) + captureV1PendingRetry.remove(path to identity.uuids) } } /** * Reads the per-event results of a 2xx and records which events, if any, the next request - * retries under the same identity. A non-2xx is thrown; a terminal one also drops the identity. + * retries under the same identity. A non-2xx is thrown; a terminal one also drops the identity, + * as does a 2xx without readable results, which is thrown as a terminal failure. */ @Throws(PostHogApiError::class) private fun captureV1Result( @@ -303,44 +323,52 @@ public class PostHogApi( ): SendResult { if (!response.isSuccessful) { if (!isCaptureV1RetriableStatusCode(response.code)) { - clearCaptureV1Retry(path) + clearCaptureV1Retry(path, identity) } throw PostHogApiError(response.code, response.message, response.body, parseRetryAfter(response)) } + // Counts only: the body may name event uuids. val results = readCaptureV1Results(response) if (results == null) { - config.logger.log( - "Capture returned ${response.code} without readable per-event results, treating ${events.size} events as delivered.", - ) + clearCaptureV1Retry(path, identity) + val message = "Capture returned ${response.code} without readable per-event results, dropping ${events.size} events." + config.logger.log(message) + throw PostHogApiError(response.code, message, null) } val retryUuids = mutableSetOf() - var dropped = 0 - // Events missing from the results count as delivered. - for (event in events) { - val uuid = event.uuid?.toString() ?: continue - val entry = results?.get(uuid) as? Map<*, *> ?: continue + // Drop counts by the server's `details` tag, an opaque string. + val droppedByDetails = sortedMapOf() + var duplicatesDropped = 0 + // Events missing from the results count as delivered. The uuids are read once, so a repeated + // uuid's later copies (`duplicates_dropped`) are counted, not given the kept copy's verdict. + for (uuid in events.mapNotNullTo(LinkedHashSet()) { it.uuid?.toString() }) { + val entry = results[uuid] as? Map<*, *> ?: continue when (entry["result"]) { "retry" -> retryUuids.add(uuid) "drop" -> { - dropped++ - // The event name, uuid and the server's reason (such as misrouted_event or - // ai_event_too_big), never the properties. - config.logger.log("Capture dropped event ${event.event} ($uuid): ${entry["details"] ?: "no details"}.") + val details = entry["details"] as? String ?: "no details" + droppedByDetails[details] = (droppedByDetails[details] ?: 0) + 1 } } + // Clamped to the batch size, so a malformed count can't overflow the sum. + duplicatesDropped += (entry["duplicates_dropped"] as? Number)?.toLong()?.coerceIn(0L, events.size.toLong())?.toInt() ?: 0 } - if (dropped > 0 || retryUuids.isNotEmpty()) { - config.logger.log("Capture: $dropped dropped, ${retryUuids.size} to retry.") + // Counts and reasons only: one line per batch, never per event. + val dropped = droppedByDetails.values.sum() + if (dropped > 0 || duplicatesDropped > 0 || retryUuids.isNotEmpty()) { + val reasons = if (dropped > 0) droppedByDetails.entries.joinToString(", ", " (", ")") { "${it.key}: ${it.value}" } else "" + config.logger.log( + "Capture $path: $dropped dropped$reasons, $duplicatesDropped duplicate copies dropped, ${retryUuids.size} to retry.", + ) } synchronized(captureV1Lock) { - if (retryUuids.isEmpty()) { - captureV1PendingRetry.remove(path) - } else { - captureV1PendingRetry[path] = identity.copy(uuids = retryUuids) + captureV1PendingRetry.remove(path to identity.uuids) + if (retryUuids.isNotEmpty()) { + putCaptureV1PendingRetry(path to retryUuids, identity.copy(uuids = retryUuids)) } } return SendResult(retryUuids, parseRetryAfter(response)) diff --git a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt index 86e015253..269c10c50 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogQueue.kt @@ -510,7 +510,7 @@ public class PostHogQueue( } } } catch (e: PostHogApiError) { - deleteFiles = deleteFilesIfAPIError(e, batchLimits, records.size, config.logger, spec.isRetriableStatusCode) + deleteFiles = deleteFilesIfAPIError(e, batchLimits, records.size, config.logger, spec.isRetriableStatusCode, spec.recordsLabel) // only re-throw if retriable (files kept), so executeWithRetry // can track retryCount and apply backoff @@ -806,6 +806,7 @@ internal fun deleteFilesIfAPIError( actualBatchSize: Int, logger: PostHogLogger, isRetriableStatusCode: (Int) -> Boolean = ::isCaptureV1RetriableStatusCode, + recordsLabel: String = "records", ): Boolean { // workaround due to png images exceed our max. limit in kafka if (e.statusCode == 413 && actualBatchSize > 1 && batchLimits.cap > 1) { @@ -823,5 +824,6 @@ internal fun deleteFilesIfAPIError( return false } + logger.log("Flushing failed with ${e.statusCode}, dropping $actualBatchSize $recordsLabel.") return true } diff --git a/posthog/src/main/java/com/posthog/internal/PostHogSendCachedEventsIntegration.kt b/posthog/src/main/java/com/posthog/internal/PostHogSendCachedEventsIntegration.kt index 0457cc456..8f353c30e 100644 --- a/posthog/src/main/java/com/posthog/internal/PostHogSendCachedEventsIntegration.kt +++ b/posthog/src/main/java/com/posthog/internal/PostHogSendCachedEventsIntegration.kt @@ -77,8 +77,15 @@ internal class PostHogSendCachedEventsIntegration( val iterator = legacy.iterator() var eventsCount = 0 + var batchBytes = 0L while (iterator.hasNext()) { val eventBytes = iterator.next() + // Checked before adding, as the queue does: a non-empty batch is sent before + // it would pass the capture v1 byte target, so order is kept and an event + // over the target ships alone. The event stays at the head for the next batch. + if (events.isNotEmpty() && batchBytes + eventBytes.size > CAPTURE_V1_MAX_BATCH_BYTES) { + break + } try { val inputStream = config.encryption?.decrypt(eventBytes.inputStream()) ?: eventBytes.inputStream() @@ -88,6 +95,7 @@ internal class PostHogSendCachedEventsIntegration( events.add(normalizeLegacyEvent(event)) eventsBytes.add(eventBytes) eventsCount++ + batchBytes += eventBytes.size } ?: run { removeFileSafely(iterator) } diff --git a/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt b/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt index b2b0cd2fe..c9bd0764d 100644 --- a/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt +++ b/posthog/src/test/java/com/posthog/PostHogStatelessTest.kt @@ -277,7 +277,7 @@ internal class PostHogStatelessTest { sut.setup(config) assertFalse(sut.isEnabledPublic()) - assertTrue(mockLogger.messages.any { it.contains("PostHog SDK is disabled because the project token is required") }) + assertTrue(mockLogger.warnings.any { it.contains("PostHog SDK is disabled because the project token is required") }) } @Test @@ -289,7 +289,7 @@ internal class PostHogStatelessTest { sut.setup(config) assertFalse(sut.isEnabledPublic()) - assertTrue(mockLogger.messages.any { it.contains("PostHog SDK is disabled because the project token is a personal API key") }) + assertTrue(mockLogger.warnings.any { it.contains("PostHog SDK is disabled because the project token is a personal API key") }) } @Test @@ -476,6 +476,49 @@ internal class PostHogStatelessTest { assertEquals(mapOf("company" to "acme"), event.properties!!["\$groups"]) } + @Test + fun `captureStateless drops an event with an empty name`() { + val mockQueue = MockQueue() + sut = createStatelessInstance() + config = createConfig() + + sut.setup(config) + sut.setMockQueue(mockQueue) + + sut.captureStateless(event = "", distinctId = "user123") + + assertTrue(mockQueue.events.isEmpty()) + } + + @Test + fun `captureStateless merges typed values over the same maps in properties per key`() { + val mockQueue = MockQueue() + sut = createStatelessInstance() + config = createConfig() + + sut.setup(config) + sut.setMockQueue(mockQueue) + + sut.captureStateless( + event = "test_event", + distinctId = "user123", + properties = + mapOf( + "\$set" to mapOf("name" to "caller", "plan" to "free"), + "\$set_once" to mapOf("first" to "caller", "source" to "ad"), + "\$groups" to mapOf("company" to "caller", "team" to "core"), + ), + userProperties = mapOf("name" to "typed"), + userPropertiesSetOnce = mapOf("first" to "typed"), + groups = mapOf("company" to "typed"), + ) + + val properties = mockQueue.events.single().properties!! + assertEquals(mapOf("name" to "typed", "plan" to "free"), properties["\$set"]) + assertEquals(mapOf("first" to "typed", "source" to "ad"), properties["\$set_once"]) + assertEquals(mapOf("company" to "typed", "team" to "core"), properties["\$groups"]) + } + private class IgnoredExceptionStub(message: String) : RuntimeException(message) @Test diff --git a/posthog/src/test/java/com/posthog/PostHogTest.kt b/posthog/src/test/java/com/posthog/PostHogTest.kt index 652549236..e231e0029 100644 --- a/posthog/src/test/java/com/posthog/PostHogTest.kt +++ b/posthog/src/test/java/com/posthog/PostHogTest.kt @@ -277,7 +277,7 @@ internal class PostHogTest { ) assertTrue(sut.isOptOut()) - assertTrue(logger.messages.any { it.contains("PostHog SDK is disabled because the project token is required") }) + assertTrue(logger.warnings.any { it.contains("PostHog SDK is disabled because the project token is required") }) sut.close() } @@ -298,7 +298,7 @@ internal class PostHogTest { ) assertTrue(sut.isOptOut()) - assertTrue(logger.messages.any { it.contains("PostHog SDK is disabled because the project token is a personal API key") }) + assertTrue(logger.warnings.any { it.contains("PostHog SDK is disabled because the project token is a personal API key") }) sut.close() } @@ -2171,6 +2171,81 @@ internal class PostHogTest { sut.close() } + @Test + fun `a caller groups map adds to stored groups, and typed groups win per key`() { + val http = mockHttp() + val prefs = PostHogMemoryPreferences() + prefs.setValue(GROUPS, mapOf("org" to "stored", "company" to "stored")) + val sut = getSut(http.url("/").toString(), preloadFeatureFlags = false, cachePreferences = prefs, reloadFeatureFlags = false) + + sut.capture( + EVENT, + DISTINCT_ID, + properties = mapOf("\$groups" to mapOf("company" to "caller", "team" to "t1")), + groups = mapOf("team" to "typed"), + ) + + queueExecutor.shutdownAndAwaitTermination() + + val properties = serializer.deserialize(http.takeRequest().body.unGzip().reader()).batch.first().properties!! + assertEquals(mapOf("org" to "stored", "company" to "caller", "team" to "typed"), properties["\$groups"]) + + sut.close() + } + + @Test + fun `typed per-call values win key by key over the same maps in properties`() { + val http = mockHttp() + val sut = getSut(http.url("/").toString(), preloadFeatureFlags = false) + + sut.capture( + EVENT, + DISTINCT_ID, + properties = + mapOf( + "\$set" to mapOf("plan" to "free", "seen" to true), + "\$set_once" to mapOf("first" to "a", "source" to "ad"), + "\$groups" to mapOf("company" to "caller", "team" to "t1"), + ), + userProperties = mapOf("plan" to "pro"), + userPropertiesSetOnce = mapOf("first" to "b"), + groups = mapOf("company" to "typed"), + ) + + queueExecutor.shutdownAndAwaitTermination() + + val properties = serializer.deserialize(http.takeRequest().body.unGzip().reader()).batch.first().properties!! + assertEquals(mapOf("plan" to "pro", "seen" to true), properties["\$set"]) + assertEquals(mapOf("first" to "b", "source" to "ad"), properties["\$set_once"]) + assertEquals(mapOf("company" to "typed", "team" to "t1"), properties["\$groups"]) + + sut.close() + } + + @Test + fun `drops an event whose name is empty, also after beforeSend`() { + val http = mockHttp() + val sut = + getSut(http.url("/").toString(), preloadFeatureFlags = false, beforeSend = { + if (it.event == "blanked") it.copy(event = "") else it + }) + val logger = TestLogger() + config.logger = logger + + sut.capture("") + sut.capture("blanked") + sut.capture(EVENT) + + queueExecutor.shutdownAndAwaitTermination() + + assertEquals(1, http.requestCount) + assertEquals(EVENT, serializer.deserialize(http.takeRequest().body.unGzip().reader()).batch.single().event) + assertEquals(1, logger.messages.count { it == "Dropping an event with an empty name." }) + assertTrue(logger.messages.none { it.contains("was dropped") }) + + sut.close() + } + @Test fun `captures an event with a custom timestamp`() { val http = mockHttp() diff --git a/posthog/src/test/java/com/posthog/internal/PostHogCaptureV1Test.kt b/posthog/src/test/java/com/posthog/internal/PostHogCaptureV1Test.kt index 1239204e9..e69f652d7 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogCaptureV1Test.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogCaptureV1Test.kt @@ -116,7 +116,7 @@ internal class PostHogCaptureV1Test { @Test fun `posts to the capture v1 endpoint with the required headers`() { - val http = httpServers.mockHttp() + val http = httpServers.mockHttp(response = resultsResponse()) clock.setCurrentDate(parseISO8601Date("2024-05-01T10:20:30.456Z")!!) api(http.url("/").toString()).captureV1(listOf(event())) @@ -136,7 +136,7 @@ internal class PostHogCaptureV1Test { @Test fun `captureAi posts to the AI endpoint with the same headers and body shape`() { - val http = httpServers.mockHttp() + val http = httpServers.mockHttp(response = resultsResponse()) clock.setCurrentDate(parseISO8601Date("2024-05-01T10:20:30.456Z")!!) val event = event("\$ai_generation") @@ -187,7 +187,7 @@ internal class PostHogCaptureV1Test { @Test fun `the capture Bearer token replaces a custom Authorization header`() { - val http = httpServers.mockHttp() + val http = httpServers.mockHttp(response = resultsResponse()) val config = config(http.url("/").toString()).apply { requestHeaders = mapOf("Authorization" to "Bearer proxy", "X-Proxy" to "yes") @@ -202,7 +202,7 @@ internal class PostHogCaptureV1Test { @Test fun `body carries created_at and the batch without api_key or sent_at`() { - val http = httpServers.mockHttp() + val http = httpServers.mockHttp(response = resultsResponse()) clock.setCurrentDate(parseISO8601Date("2024-05-01T10:20:30.456Z")!!) val event = event() @@ -331,13 +331,13 @@ internal class PostHogCaptureV1Test { assertEquals(setOf(retry.uuid.toString()), result.retryIds) assertEquals(7, result.retryAfterSeconds) - assertTrue(logger.messages.contains("Capture: 1 dropped, 1 to retry.")) - assertTrue(logger.messages.contains("Capture dropped event dropped (${dropped.uuid}): no details.")) - assertTrue(logger.messages.none { it.contains(retry.uuid.toString()) }) + assertTrue( + logger.messages.contains("Capture /i/v1/analytics/events: 1 dropped (no details: 1), 0 duplicate copies dropped, 1 to retry."), + ) } @Test - fun `logs each dropped event with the details capture gives, without its properties`() { + fun `logs one line per batch with the endpoint and drop reasons, without event names, uuids or properties`() { val misrouted = event("button_clicked", properties = mapOf("prompt" to "secret prompt")) val tooBig = event("\$ai_generation", properties = mapOf("prompt" to "secret prompt")) val http = @@ -352,24 +352,66 @@ internal class PostHogCaptureV1Test { api(http.url("/").toString()).captureAi(listOf(misrouted, tooBig)) - assertTrue(logger.messages.contains("Capture dropped event button_clicked (${misrouted.uuid}): misrouted_event.")) - assertTrue(logger.messages.contains("Capture dropped event \$ai_generation (${tooBig.uuid}): ai_event_too_big.")) - assertTrue(logger.messages.contains("Capture: 2 dropped, 0 to retry.")) - assertTrue(logger.messages.none { it.contains("secret prompt") }) + assertEquals( + listOf("Capture /i/v1/ai/events: 2 dropped (ai_event_too_big: 1, misrouted_event: 1), 0 duplicate copies dropped, 0 to retry."), + logger.messages.filter { it.startsWith("Capture /") }, + ) + assertTrue( + logger.messages.none { + it.contains("secret prompt") || it.contains("button_clicked") || + it.contains(misrouted.uuid.toString()) || it.contains(tooBig.uuid.toString()) + }, + ) } @Test - fun `a 2xx without readable results counts as delivered`() { - val http = httpServers.mockHttp(response = MockResponse().setBody("not json")) + fun `counts the duplicate copies capture dropped`() { + val kept = event() + val copy = kept.copy() + val http = + httpServers.mockHttp( + response = MockResponse().setBody("{\"results\":{\"${kept.uuid}\":{\"result\":\"ok\",\"duplicates_dropped\":1}}}"), + ) - val result = api(http.url("/").toString()).captureV1(listOf(event())) + val result = api(http.url("/").toString()).captureV1(listOf(kept, copy)) assertTrue(result.retryIds.isEmpty()) - assertTrue(logger.messages.any { it.contains("without readable per-event results, treating 1 events as delivered") }) + assertTrue(logger.messages.contains("Capture /i/v1/analytics/events: 0 dropped, 1 duplicate copies dropped, 0 to retry.")) + } + + @Test + fun `clamps each duplicate count to the batch size`() { + val first = event() + val second = event() + val http = + httpServers.mockHttp( + response = + MockResponse().setBody( + "{\"results\":{\"${first.uuid}\":{\"result\":\"ok\",\"duplicates_dropped\":99999999999999999999}," + + "\"${second.uuid}\":{\"result\":\"ok\",\"duplicates_dropped\":-5}}}", + ), + ) + + api(http.url("/").toString()).captureV1(listOf(first, second)) + + assertTrue(logger.messages.contains("Capture /i/v1/analytics/events: 0 dropped, 2 duplicate copies dropped, 0 to retry.")) + } + + @Test + fun `a 2xx without readable results is a terminal failure`() { + for (body in listOf("not json", "{}", "{\"results\":[]}")) { + val http = httpServers.mockHttp(response = MockResponse().setBody(body)) + + val error = assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event())) } + + assertEquals(200, error.statusCode) + assertFalse(isCaptureV1RetriableStatusCode(error.statusCode)) + } + assertTrue(logger.messages.any { it.contains("without readable per-event results, dropping 1 events") }) } @Test - fun `a 2xx whose transfer fails after the status counts as delivered`() { + fun `a 2xx whose transfer fails after the status is a terminal failure`() { val http = httpServers.mockHttp( response = @@ -378,13 +420,11 @@ internal class PostHogCaptureV1Test { .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY), ) - val result = api(http.url("/").toString()).captureV1(listOf(event())) - - assertTrue(result.retryIds.isEmpty()) + assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event())) } } @Test - fun `a 2xx with results over 1 MiB counts as delivered`() { + fun `a 2xx with results over 1 MiB is a terminal failure`() { val event = event() val padding = "x".repeat(1024 * 1024) val http = @@ -392,13 +432,11 @@ internal class PostHogCaptureV1Test { response = MockResponse().setBody("{\"results\":{\"${event.uuid}\":{\"result\":\"retry\"}},\"pad\":\"$padding\"}"), ) - val result = api(http.url("/").toString()).captureV1(listOf(event)) - - assertTrue(result.retryIds.isEmpty()) + assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event)) } } @Test - fun `a 2xx whose results trickle past the read deadline counts as delivered`() { + fun `a 2xx whose results trickle past the read deadline is a terminal failure`() { val event = event() val padding = "x".repeat(16 * 1024) // throttles the request read too, so allow a whole small request per period @@ -411,9 +449,7 @@ internal class PostHogCaptureV1Test { ) val sut = api(http.url("/").toString()).apply { captureV1ResultsReadDeadlineMs = 300 } - val result = sut.captureV1(listOf(event)) - - assertTrue(result.retryIds.isEmpty()) + assertThrows(PostHogApiError::class.java) { sut.captureV1(listOf(event)) } } @Test @@ -450,7 +486,7 @@ internal class PostHogCaptureV1Test { @Test fun `a 503 with Retry-After 0 is not resent by the HTTP client`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(503).setHeader("Retry-After", "0")) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event())) } @@ -484,7 +520,7 @@ internal class PostHogCaptureV1Test { fun `a retriable status and a network error keep the request identity`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(503)) http.enqueue(MockResponse().setSocketPolicy(SocketPolicy.NO_RESPONSE)) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) val events = listOf(event()) // a short read timeout turns the unanswered request into an IOException OkHttp doesn't retry itself val config = @@ -504,10 +540,10 @@ internal class PostHogCaptureV1Test { @Test fun `a dropped connection is not resent by the HTTP client`() { - val http = httpServers.mockHttp(response = MockResponse().setBody("")) + val http = httpServers.mockHttp(response = resultsResponse()) // the second request reuses the pooled connection, which OkHttp would otherwise retry on a new one http.enqueue(MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST)) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) val sut = api(http.url("/").toString()) sut.captureV1(listOf(event())) @@ -545,10 +581,31 @@ internal class PostHogCaptureV1Test { assertNotEquals(first.headers["PostHog-Request-Id"], second.headers["PostHog-Request-Id"]) } + @Test + fun `sending other events in between keeps the pending retry identity`() { + val retried = event() + val other = event("other") + val http = httpServers.mockHttp(response = resultsResponse(retried.uuid!! to "retry")) + http.enqueue(resultsResponse(other.uuid!! to "ok")) + http.enqueue(resultsResponse(retried.uuid!! to "ok")) + val sut = api(http.url("/").toString()) + + sut.captureV1(listOf(retried)) + sut.captureV1(listOf(other)) + sut.captureV1(listOf(retried)) + + val (first, between, resent) = List(3) { http.takeRequest() } + assertEquals("1", between.headers["PostHog-Attempt"]) + assertNotEquals(first.headers["PostHog-Request-Id"], between.headers["PostHog-Request-Id"]) + assertEquals("2", resent.headers["PostHog-Attempt"]) + assertEquals(first.headers["PostHog-Request-Id"], resent.headers["PostHog-Request-Id"]) + assertEquals(body(first)["created_at"], body(resent)["created_at"]) + } + @Test fun `a terminal status starts a new request identity on the next send`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(400)) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) val events = listOf(event()) val sut = api(http.url("/").toString()) @@ -577,7 +634,7 @@ internal class PostHogCaptureV1Test { fun `follows a same-origin 307 with the same body and headers`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(307).setHeader("Location", "/proxied/events")) http.enqueue(MockResponse().setResponseCode(308).setHeader("Location", "/final/events")) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) api(http.url("/").toString()).captureV1(listOf(event())) @@ -595,7 +652,7 @@ internal class PostHogCaptureV1Test { @Test fun `does not follow a 302`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(302).setHeader("Location", "/elsewhere")) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) val error = assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event())) } @@ -605,7 +662,7 @@ internal class PostHogCaptureV1Test { @Test fun `does not follow a 307 to another origin`() { - val other = httpServers.mockHttp() + val other = httpServers.mockHttp(response = resultsResponse()) val http = httpServers.mockHttp( response = MockResponse().setResponseCode(307).setHeader("Location", other.url("/events").toString()), @@ -622,7 +679,7 @@ internal class PostHogCaptureV1Test { @Test fun `stops after five redirects`() { val http = httpServers.mockHttp(total = 6, response = MockResponse().setResponseCode(307).setHeader("Location", "/again")) - http.enqueue(MockResponse().setBody("")) + http.enqueue(resultsResponse()) val error = assertThrows(PostHogApiError::class.java) { api(http.url("/").toString()).captureV1(listOf(event())) } @@ -751,6 +808,20 @@ internal class PostHogCaptureV1Test { assertEquals(1, sut.dequeList.size) } + @Test + fun `queue drops a batch whose 2xx results can't be read without resending it`() { + val http = httpServers.mockHttp(response = MockResponse().setBody("not json")) + http.enqueue(resultsResponse()) + val sut = queue(http) + + sut.add(event()) + executor.awaitExecution() + + assertTrue(sut.dequeList.isEmpty()) + assertEquals(1, http.requestCount) + assertFalse(logger.messages.any { it.contains("successfully") }) + } + @Test fun `queued events round-trip their options through disk`() { val http = httpServers.mockHttp(response = MockResponse().setResponseCode(503)) diff --git a/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt index 670f95e88..58b7348ad 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogQueueTest.kt @@ -5,6 +5,7 @@ import com.posthog.PostHogConfig import com.posthog.PostHogEvent import com.posthog.PostHogEventName import com.posthog.TestHttpServers +import com.posthog.TestLogger import com.posthog.awaitExecution import com.posthog.generateEvent import com.posthog.internal.errortracking.ThrowableCoercer @@ -958,8 +959,21 @@ internal class PostHogQueueTest { val e = PostHogApiError(400, "", null) val config = PostHogConfig(PROJECT_TOKEN) val limits = initialBatchLimits(config) + val logger = TestLogger() - assertTrue(deleteFilesIfAPIError(e, limits, actualBatchSize = limits.cap, logger = config.logger)) + assertTrue(deleteFilesIfAPIError(e, limits, actualBatchSize = limits.cap, logger = logger)) + assertEquals(listOf("Flushing failed with 400, dropping ${limits.cap} records."), logger.messages) + } + + @Test + fun `the drop log names the records it drops`() { + val e = PostHogApiError(400, "", null) + val config = PostHogConfig(PROJECT_TOKEN) + val logger = TestLogger() + + deleteFilesIfAPIError(e, initialBatchLimits(config), actualBatchSize = 3, logger = logger, recordsLabel = "AI events") + + assertEquals(listOf("Flushing failed with 400, dropping 3 AI events."), logger.messages) } @Test diff --git a/posthog/src/test/java/com/posthog/internal/PostHogSendCachedLegacyEventsIntegrationTest.kt b/posthog/src/test/java/com/posthog/internal/PostHogSendCachedLegacyEventsIntegrationTest.kt index a6f527d6e..78b239036 100644 --- a/posthog/src/test/java/com/posthog/internal/PostHogSendCachedLegacyEventsIntegrationTest.kt +++ b/posthog/src/test/java/com/posthog/internal/PostHogSendCachedLegacyEventsIntegrationTest.kt @@ -163,17 +163,33 @@ internal class PostHogSendCachedLegacyEventsIntegrationTest { @Test fun `sends events from the legacy queue in batches`() { val prefix = writeLegacyFile(listOf(event, event)) - val http = servers.mockHttp(2) + val http = servers.mockHttp(2, MockResponse().setBody("{\"results\":{}}")) getSut(prefix, http.url("/").toString(), maxBatchSize = 1).install(mock()) executor.shutdownAndAwaitTermination() assertEquals(emptyList(), retained(prefix)) assertEquals(2, http.requestCount) } + @Test + fun `closes a batch before it would pass 5 MiB and sends an oversized event alone, in order`() { + fun sized( + name: String, + bytes: Int, + ) = event.replace("testEvent", name).replace("testValue", "a".repeat(bytes)) + val mib = 1024 * 1024 + val prefix = writeLegacyFile(listOf(sized("one", 2 * mib), sized("two", 2 * mib), sized("big", 6 * mib), sized("small", 1))) + val http = servers.mockHttp(3, MockResponse().setBody("{\"results\":{}}")) + getSut(prefix, http.url("/").toString()).install(mock()) + executor.shutdownAndAwaitTermination() + assertEquals(emptyList(), retained(prefix)) + val batches = (1..http.requestCount).map { sentEvents(assertNotNull(http.takeRequest(5, TimeUnit.SECONDS))).map { it["event"] } } + assertEquals(listOf(listOf("one", "two"), listOf("big"), listOf("small")), batches) + } + @Test fun `send a valid event and discard a broken event`() { val prefix = writeLegacyFile(listOf("invalid event", event)) - val http = servers.mockHttp(2) + val http = servers.mockHttp(2, MockResponse().setBody("{\"results\":{}}")) getSut(prefix, http.url("/").toString(), maxBatchSize = 1).install(mock()) executor.shutdownAndAwaitTermination() assertEquals(emptyList(), retained(prefix))