From 39a2b3ef38ca99e11193fe93c752fe9a4d91d2e9 Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Fri, 7 Aug 2026 10:13:07 +0200 Subject: [PATCH 01/15] Align OpenFeature track() with Confidence event conventions. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merge per-event OpenFeature context with session context, map tracking details to event data, and allow explicit context overrides in the payload — matching Swift provider behavior. --- Confidence/api/Confidence.api | 2 + .../java/com/spotify/confidence/Confidence.kt | 10 +- .../com/spotify/confidence/EventSender.kt | 10 ++ .../com/spotify/confidence/PayloadMerger.kt | 8 +- .../confidence/EventSenderIntegrationTest.kt | 21 +++- .../spotify/confidence/PayloadMergerTest.kt | 17 +++ .../openfeature/ConfidenceFeatureProvider.kt | 20 ++- .../openfeature/OpenFeatureTrackMapper.kt | 43 +++++++ .../ConfidenceFeatureProviderTrackTest.kt | 118 ++++++++++++++++++ .../openfeature/OpenFeatureTrackMapperTest.kt | 49 ++++++++ 10 files changed, 279 insertions(+), 19 deletions(-) create mode 100644 Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt create mode 100644 Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt create mode 100644 Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt diff --git a/Confidence/api/Confidence.api b/Confidence/api/Confidence.api index 8535a5ed..c9afb419 100644 --- a/Confidence/api/Confidence.api +++ b/Confidence/api/Confidence.api @@ -19,6 +19,7 @@ public final class com/spotify/confidence/Confidence : com/spotify/confidence/Co public fun stop ()V public fun track (Lcom/spotify/confidence/Producer;)V public fun track (Ljava/lang/String;Ljava/util/Map;)V + public fun track (Ljava/lang/String;Ljava/util/Map;Ljava/util/Map;)V public synthetic fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/Contextual; public fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/EventSender; } @@ -407,6 +408,7 @@ public abstract interface class com/spotify/confidence/EventSender : com/spotify public abstract fun stop ()V public abstract fun track (Lcom/spotify/confidence/Producer;)V public abstract fun track (Ljava/lang/String;Ljava/util/Map;)V + public abstract fun track (Ljava/lang/String;Ljava/util/Map;Ljava/util/Map;)V public abstract fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/EventSender; } diff --git a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt index 3fece006..d756060a 100644 --- a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt +++ b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt @@ -254,7 +254,15 @@ class Confidence internal constructor( eventName: String, data: ConfidenceFieldsType ) { - eventSenderEngine.emit(eventName, data, getContext()) + track(eventName, data, getContext()) + } + + override fun track( + eventName: String, + data: ConfidenceFieldsType, + eventContext: Map + ) { + eventSenderEngine.emit(eventName, data, eventContext) } override fun flush() { diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSender.kt b/Confidence/src/main/java/com/spotify/confidence/EventSender.kt index 0e8ac036..b6de5393 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSender.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSender.kt @@ -11,6 +11,16 @@ interface EventSender : Contextual { data: ConfidenceFieldsType = mapOf() ) + /** + * Store a custom event to be tracked with an explicit evaluation context. + * @param eventContext evaluation context for this event only; does not mutate session context. + */ + fun track( + eventName: String, + data: ConfidenceFieldsType, + eventContext: Map + ) + /** * Track Android-specific events like activities or Track Context updates. * Please note that this method is collecting data in a coroutine scope and will be diff --git a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt index 52449d68..03f44494 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -4,9 +4,11 @@ private typealias ConfidenceStruct = Map internal interface PayloadMerger : (ConfidenceStruct, ConfidenceStruct) -> ConfidenceStruct internal class PayloadMergerImpl : PayloadMerger { override fun invoke(context: ConfidenceStruct, message: ConfidenceStruct): ConfidenceStruct { - if (message.containsKey("context")) { - throw ConfidenceError.InvalidContextInMessage() + return if (message.containsKey("context")) { + // An explicit "context" entry in event data overrides the evaluation context for this event. + message + } else { + message + mapOf("context" to ConfidenceValue.Struct(context)) } - return message + (mapOf("context" to ConfidenceValue.Struct(context))) } } diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt index 12bb21a7..245bb88a 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt @@ -2,7 +2,6 @@ package com.spotify.confidence import android.content.Context import android.content.SharedPreferences -import com.spotify.confidence.ConfidenceError.InvalidContextInMessage import com.spotify.confidence.client.SdkMetadata import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.runBlocking @@ -43,15 +42,29 @@ class EventSenderIntegrationTest { } } - @Test(expected = InvalidContextInMessage::class) - fun context_in_message_throws() = runTest { + @Test + fun context_in_message_overrides_evaluation_context() = runTest { val testDispatcher = UnconfinedTestDispatcher(testScheduler) val confidence = ConfidenceFactory.create( mockContext, clientSecret, dispatcher = testDispatcher ) - confidence.track("test", mapOf("context" to ConfidenceValue.Integer(1))) + confidence.track( + eventName = "test", + data = mapOf("context" to ConfidenceValue.String("override")), + eventContext = mapOf("a" to ConfidenceValue.Integer(1)) + ) + advanceUntilIdle() + val eventStorage = EventStorageImpl(mockContext) + val files = directory.walkFiles().toList() + Assert.assertEquals(1, files.size) + val events = eventStorage.eventsFor(files.first()) + Assert.assertEquals(1, events.size) + Assert.assertEquals( + ConfidenceValue.String("override"), + events.first().payload["context"] + ) } @Test diff --git a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt index 7971b721..e8db8ddc 100644 --- a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt @@ -22,4 +22,21 @@ class PayloadMergerTest { ) ) } + + @Test + fun `context in data overrides evaluation context`() { + val payloadMerger = PayloadMergerImpl() + val context = mapOf("a" to ConfidenceValue.Integer(1), "b" to ConfidenceValue.Integer(2)) + val message = mapOf( + "b" to ConfidenceValue.Integer(3), + "context" to ConfidenceValue.String("override") + ) + val result = payloadMerger(context, message) + assert( + result == mapOf( + "b" to ConfidenceValue.Integer(3), + "context" to ConfidenceValue.String("override") + ) + ) + } } diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt index 73d2438f..50310bb1 100644 --- a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt @@ -113,7 +113,15 @@ class ConfidenceFeatureProvider private constructor( } override fun track(trackingEventName: String, context: EvaluationContext?, details: TrackingEventDetails?) { - confidence.track(trackingEventName, details?.toConfidenceValue() ?: emptyMap()) + val eventContext = mergeEventContext( + sessionContext = confidence.getContext(), + openFeatureContext = context.toTrackContextMap() + ) + confidence.track( + eventName = trackingEventName, + data = details.toTrackingData(), + eventContext = eventContext + ) } private fun generateEvaluation( @@ -157,16 +165,6 @@ class ConfidenceFeatureProvider private constructor( } } -private fun TrackingEventDetails.toConfidenceValue(): Map = mapOf( - "value" to (this.value?.toConfidenceValue() ?: ConfidenceValue.Null) -) + this.structure.asMap().mapValues { it.value.toConfidenceValue() } - -private fun Number.toConfidenceValue(): ConfidenceValue = when (this) { - is Int -> ConfidenceValue.Integer(this) - is Double -> ConfidenceValue.Double(this) - else -> ConfidenceValue.Null -} - internal fun Value.toConfidenceValue(): ConfidenceValue = when (this) { is Value.Structure -> ConfidenceValue.Struct(structure.mapValues { it.value.toConfidenceValue() }) is Value.Boolean -> ConfidenceValue.Boolean(this.boolean) diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt new file mode 100644 index 00000000..4ae965ba --- /dev/null +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt @@ -0,0 +1,43 @@ +package com.spotify.confidence.openfeature + +import com.spotify.confidence.ConfidenceValue +import dev.openfeature.kotlin.sdk.EvaluationContext +import dev.openfeature.kotlin.sdk.TrackingEventDetails + +internal fun mergeEventContext( + sessionContext: Map, + openFeatureContext: Map +): Map { + if (openFeatureContext.isEmpty()) { + return sessionContext + } + return sessionContext + openFeatureContext +} + +internal fun EvaluationContext?.toTrackContextMap(): Map { + if (this == null) { + return emptyMap() + } + val map = mutableMapOf() + val targetingKey = getTargetingKey() + if (targetingKey.isNotEmpty() && !asMap().containsKey("targeting_key")) { + map["targeting_key"] = ConfidenceValue.String(targetingKey) + } + map.putAll(asMap().mapValues { it.value.toConfidenceValue() }) + return map +} + +internal fun TrackingEventDetails?.toTrackingData(): Map { + if (this == null) { + return emptyMap() + } + return mapOf( + "value" to (value?.toConfidenceValue() ?: ConfidenceValue.Null) + ) + structure.asMap().mapValues { it.value.toConfidenceValue() } +} + +private fun Number.toConfidenceValue(): ConfidenceValue = when (this) { + is Int -> ConfidenceValue.Integer(this) + is Double -> ConfidenceValue.Double(this) + else -> ConfidenceValue.Null +} diff --git a/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt new file mode 100644 index 00000000..889b1c34 --- /dev/null +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt @@ -0,0 +1,118 @@ +package com.spotify.confidence.openfeature + +import com.spotify.confidence.Confidence +import com.spotify.confidence.ConfidenceValue +import dev.openfeature.kotlin.sdk.ImmutableContext +import dev.openfeature.kotlin.sdk.ImmutableStructure +import dev.openfeature.kotlin.sdk.TrackingEventDetails +import dev.openfeature.kotlin.sdk.Value +import io.mockk.every +import io.mockk.mockk +import io.mockk.slot +import io.mockk.verify +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test + +class ConfidenceFeatureProviderTrackTest { + @Test + fun trackForwardsMergedContextAndMappedData() { + val confidence = mockk(relaxed = true) + every { confidence.getContext() } returns mapOf("plan" to ConfidenceValue.String("free")) + + val dataSlot = slot>() + val contextSlot = slot>() + every { + confidence.track( + eventName = "Checkout", + data = capture(dataSlot), + eventContext = capture(contextSlot) + ) + } returns Unit + + val provider = ConfidenceFeatureProvider.create(confidence) + val details = TrackingEventDetails( + 499.99, + ImmutableStructure( + "numberOfItems" to Value.Integer(4), + "timeInCheckout" to Value.String("PT3M20S") + ) + ) + val context = ImmutableContext( + targetingKey = "user-1", + attributes = mapOf( + "plan" to Value.String("premium"), + "country" to Value.String("SE") + ) + ) + + provider.track("Checkout", context, details) + + verify { + confidence.track( + eventName = "Checkout", + data = any(), + eventContext = any() + ) + } + assertEquals(ConfidenceValue.Double(499.99), dataSlot.captured["value"]) + assertEquals(ConfidenceValue.Integer(4), dataSlot.captured["numberOfItems"]) + assertEquals(ConfidenceValue.String("premium"), contextSlot.captured["plan"]) + assertEquals(ConfidenceValue.String("SE"), contextSlot.captured["country"]) + assertEquals(ConfidenceValue.String("user-1"), contextSlot.captured["targeting_key"]) + } + + @Test + fun trackWithoutDetailsSendsEmptyData() { + val confidence = mockk(relaxed = true) + every { confidence.getContext() } returns emptyMap() + + val dataSlot = slot>() + every { + confidence.track( + eventName = "PageView", + data = capture(dataSlot), + eventContext = any() + ) + } returns Unit + + val provider = ConfidenceFeatureProvider.create(confidence) + provider.track("PageView", null, null) + + assertTrue(dataSlot.captured.isEmpty()) + } + + @Test + fun trackContextAttributeOverridesMergedEvaluationContext() { + val confidence = mockk(relaxed = true) + every { confidence.getContext() } returns mapOf("plan" to ConfidenceValue.String("free")) + + val dataSlot = slot>() + every { + confidence.track( + eventName = "Checkout", + data = capture(dataSlot), + eventContext = any() + ) + } returns Unit + + val provider = ConfidenceFeatureProvider.create(confidence) + val details = TrackingEventDetails( + null, + ImmutableStructure( + "context" to Value.Structure(mapOf("source" to Value.String("details"))) + ) + ) + + provider.track( + "Checkout", + ImmutableContext(attributes = mapOf("plan" to Value.String("premium"))), + details + ) + + assertEquals( + ConfidenceValue.Struct(mapOf("source" to ConfidenceValue.String("details"))), + dataSlot.captured["context"] + ) + } +} diff --git a/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt b/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt new file mode 100644 index 00000000..87dd86c5 --- /dev/null +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt @@ -0,0 +1,49 @@ +package com.spotify.confidence.openfeature + +import com.spotify.confidence.ConfidenceValue +import dev.openfeature.kotlin.sdk.ImmutableContext +import dev.openfeature.kotlin.sdk.ImmutableStructure +import dev.openfeature.kotlin.sdk.TrackingEventDetails +import dev.openfeature.kotlin.sdk.Value +import org.junit.Assert.assertEquals +import org.junit.Test + +class OpenFeatureTrackMapperTest { + @Test + fun mergeEventContextUsesOpenFeatureValuesOnConflict() { + val merged = mergeEventContext( + sessionContext = mapOf( + "plan" to ConfidenceValue.String("free"), + "visitor_id" to ConfidenceValue.String("v1") + ), + openFeatureContext = mapOf( + "plan" to ConfidenceValue.String("premium"), + "country" to ConfidenceValue.String("SE") + ) + ) + assertEquals(ConfidenceValue.String("premium"), merged["plan"]) + assertEquals(ConfidenceValue.String("v1"), merged["visitor_id"]) + assertEquals(ConfidenceValue.String("SE"), merged["country"]) + } + + @Test + fun trackingDetailsValueAttributeOverridesNumericValue() { + val details = TrackingEventDetails( + 99.77, + ImmutableStructure("value" to Value.String("override")) + ) + val data = details.toTrackingData() + assertEquals(ConfidenceValue.String("override"), data["value"]) + } + + @Test + fun trackContextMapIncludesTargetingKey() { + val context = ImmutableContext( + targetingKey = "user-1", + attributes = mapOf("country" to Value.String("SE")) + ) + val map = context.toTrackContextMap() + assertEquals(ConfidenceValue.String("user-1"), map["targeting_key"]) + assertEquals(ConfidenceValue.String("SE"), map["country"]) + } +} From 16f2401091738c98497f3a0eb859e7fbd920a90d Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Fri, 7 Aug 2026 10:56:35 +0200 Subject: [PATCH 02/15] feat: improve event delivery reliability for tracked events Upload pending batches at startup, flush and upload on stop(), and add an optional periodic flush interval so low-volume events are not left on disk indefinitely. --- .../java/com/spotify/confidence/Confidence.kt | 19 ++- .../spotify/confidence/EventSenderEngine.kt | 100 ++++++++++---- .../EventSenderEngineReliabilityTest.kt | 129 ++++++++++++++++++ .../openfeature/ConfidenceFeatureProvider.kt | 1 + 4 files changed, 213 insertions(+), 36 deletions(-) create mode 100644 Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt diff --git a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt index d756060a..865aa1b2 100644 --- a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt +++ b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt @@ -385,6 +385,7 @@ object ConfidenceFactory { * @param loggingLevel allows to print warnings or debugging information to the local console. * @param timeoutMillis sets a timeout for completing an HTTP call. Defaults to 10 seconds * @param visitorIdContextKey key to use for the visitor id in the context. Defaults to "visitor_id". + * @param eventFlushIntervalMillis optional periodic flush interval in milliseconds. Disabled by default. */ fun create( context: Context, @@ -394,7 +395,8 @@ object ConfidenceFactory { dispatcher: CoroutineDispatcher = Dispatchers.IO, loggingLevel: LoggingLevel = LoggingLevel.WARN, timeoutMillis: Long = 10000, - visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY + visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY, + eventFlushIntervalMillis: Long? = null ): Confidence = create( context = context, clientSecret = clientSecret, @@ -404,7 +406,8 @@ object ConfidenceFactory { loggingLevel = loggingLevel, timeoutMillis = timeoutMillis, visitorIdContextKey = visitorIdContextKey, - resolveBaseUrl = null + resolveBaseUrl = null, + eventFlushIntervalMillis = eventFlushIntervalMillis ) /** @@ -421,7 +424,8 @@ object ConfidenceFactory { dispatcher: CoroutineDispatcher = Dispatchers.IO, loggingLevel: LoggingLevel = LoggingLevel.WARN, timeoutMillis: Long = 10000, - visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY + visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY, + eventFlushIntervalMillis: Long? = null ): Confidence = create( context = context, clientSecret = clientSecret, @@ -431,7 +435,8 @@ object ConfidenceFactory { loggingLevel = loggingLevel, timeoutMillis = timeoutMillis, visitorIdContextKey = visitorIdContextKey, - resolveBaseUrl = getResolveBaseUrl(region, resolveBaseUrl) + resolveBaseUrl = getResolveBaseUrl(region, resolveBaseUrl), + eventFlushIntervalMillis = eventFlushIntervalMillis ) private fun create( @@ -443,7 +448,8 @@ object ConfidenceFactory { loggingLevel: LoggingLevel, timeoutMillis: Long, visitorIdContextKey: String, - resolveBaseUrl: HttpUrl? + resolveBaseUrl: HttpUrl?, + eventFlushIntervalMillis: Long? = null ): Confidence { val debugLogger: DebugLogger? = if (loggingLevel == LoggingLevel.NONE) { null @@ -458,7 +464,8 @@ object ConfidenceFactory { flushPolicies = listOf(minBatchSizeFlushPolicy), sdkMetadata = sdkMetadata, dispatcher = dispatcher, - debugLogger = debugLogger + debugLogger = debugLogger, + flushIntervalMillis = eventFlushIntervalMillis ) val flagApplierClient = FlagApplierClientImpl( clientSecret, diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 877729ec..d665b1da 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -8,10 +8,14 @@ import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking import okhttp3.OkHttpClient import java.io.File @@ -30,7 +34,8 @@ internal class EventSenderEngineImpl( private val clock: Clock = Clock.CalendarBacked.systemUTC(), private val dispatcher: CoroutineDispatcher = Dispatchers.IO, private val sdkMetadata: SdkMetadata, - private val debugLogger: DebugLogger? + private val debugLogger: DebugLogger?, + private val flushIntervalMillis: Long? = null ) : EventSenderEngine { private val writeReqChannel: Channel = Channel() private val sendChannel: Channel = Channel() @@ -43,13 +48,15 @@ internal class EventSenderEngineImpl( debugLogger?.logMessage(message = "EventSenderEngine error: $e", isWarning = true) } } + private var flushIntervalJob: Job? = null + @Volatile + private var isStopped = false init { flushPolicies.add(ManualFlushPolicy) coroutineScope.launch(exceptionHandler) { for (event in writeReqChannel) { if (event.eventDefinition != manualFlushEvent.eventDefinition) { - // skip storing manual flush event eventStorage.writeEvent(event) debugLogger?.logEvent(action = "DiskWrite ", event = event) } @@ -69,36 +76,24 @@ internal class EventSenderEngineImpl( } } - // upload might throw exceptions coroutineScope.launch(exceptionHandler) { for (flush in sendChannel) { - eventStorage.rollover() - val readyFiles = eventStorage.batchReadyFiles() - for (readyFile in readyFiles) { - val events = eventStorage.eventsFor(readyFile) - .map { e -> - EngineEvent( - "eventDefinitions/${e.eventDefinition}", - e.eventTime, - e.payload - ) - } - val batch = EventBatchRequest( - clientSecret = clientSecret, - events = events, - sendTime = clock.currentTime(), - sdk = Sdk(sdkMetadata.sdkId, sdkMetadata.sdkVersion) - ) - runCatching { - val shouldCleanup = uploader.upload(batch) - debugLogger?.logMessage(message = "Uploading events") - if (shouldCleanup) { - readyFile.delete() - } - } + uploadReadyBatches(sealCurrentBatch = true) + } + } + + if (flushIntervalMillis != null && flushIntervalMillis > 0) { + flushIntervalJob = coroutineScope.launch(exceptionHandler) { + while (isActive) { + delay(flushIntervalMillis) + flush() } } } + + coroutineScope.launch(exceptionHandler) { + uploadReadyBatches(sealCurrentBatch = false) + } } override fun onLowMemoryChannel(): Channel> { @@ -111,6 +106,9 @@ internal class EventSenderEngineImpl( data: ConfidenceFieldsType, context: Map ) { + if (isStopped) { + return + } coroutineScope.launch { val payload = payloadMerger(context, data) val event = EngineEvent( @@ -124,6 +122,9 @@ internal class EventSenderEngineImpl( } override fun flush() { + if (isStopped) { + return + } coroutineScope.launch { writeReqChannel.send(manualFlushEvent) debugLogger?.logEvent(action = "Flush ", event = manualFlushEvent) @@ -131,11 +132,46 @@ internal class EventSenderEngineImpl( } override fun stop() { + isStopped = true + flushIntervalJob?.cancel() + runBlocking(dispatcher) { + uploadReadyBatches(sealCurrentBatch = true) + } coroutineScope.cancel() eventStorage.stop() debugLogger?.logMessage(message = "EventSenderEngine closed ") } + private suspend fun uploadReadyBatches(sealCurrentBatch: Boolean) { + if (sealCurrentBatch) { + eventStorage.rollover() + } + val readyFiles = eventStorage.batchReadyFiles() + for (readyFile in readyFiles) { + val events = eventStorage.eventsFor(readyFile) + .map { e -> + EngineEvent( + "eventDefinitions/${e.eventDefinition}", + e.eventTime, + e.payload + ) + } + val batch = EventBatchRequest( + clientSecret = clientSecret, + events = events, + sendTime = clock.currentTime(), + sdk = Sdk(sdkMetadata.sdkId, sdkMetadata.sdkVersion) + ) + runCatching { + val shouldCleanup = uploader.upload(batch) + debugLogger?.logMessage(message = "Uploading events") + if (shouldCleanup) { + readyFile.delete() + } + } + } + } + companion object { private const val SEND_SIG = "FLUSH" private var Instance: EventSenderEngine? = null @@ -145,7 +181,8 @@ internal class EventSenderEngineImpl( sdkMetadata: SdkMetadata, flushPolicies: List = listOf(), dispatcher: CoroutineDispatcher = Dispatchers.IO, - debugLogger: DebugLogger? + debugLogger: DebugLogger?, + flushIntervalMillis: Long? = null ): EventSenderEngine { return Instance ?: run { EventSenderEngineImpl( @@ -155,8 +192,11 @@ internal class EventSenderEngineImpl( flushPolicies = flushPolicies.toMutableList(), dispatcher = dispatcher, sdkMetadata = sdkMetadata, - debugLogger = debugLogger - ) + debugLogger = debugLogger, + flushIntervalMillis = flushIntervalMillis + ).also { + Instance = it + } } } } diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt new file mode 100644 index 00000000..d67d9668 --- /dev/null +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -0,0 +1,129 @@ +package com.spotify.confidence + +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.advanceUntilIdle +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import java.util.Date + +@OptIn(ExperimentalCoroutinesApi::class) +class EventSenderEngineReliabilityTest { + private lateinit var testDispatcher: UnconfinedTestDispatcher + private lateinit var uploader: RecordingEventUploader + private lateinit var storage: RecordingEventStorage + + @Before + fun setUp() { + testDispatcher = UnconfinedTestDispatcher() + uploader = RecordingEventUploader() + storage = RecordingEventStorage() + } + + @Test + fun startupUploadsPendingReadyBatchesWithoutSealingCurrentBatch() = runTest(testDispatcher) { + storage.readyEvents["pending.batch"] = listOf( + EngineEvent("pending", Date(), mapOf()) + ) + storage.currentEvents.add( + EngineEvent("current", Date(), mapOf()) + ) + + EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = uploader, + flushPolicies = mutableListOf(), + dispatcher = testDispatcher, + sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), + debugLogger = null + ) + + advanceUntilIdle() + + assertEquals(listOf("pending"), uploader.uploadedEventNames) + assertEquals(listOf("current"), storage.currentEvents.map { it.eventDefinition }) + } + + @Test + fun stopUploadsCurrentBatch() = runTest(testDispatcher) { + val engine = EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = uploader, + flushPolicies = mutableListOf(), + dispatcher = testDispatcher, + sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), + debugLogger = null + ) + + engine.emit("session-end", mapOf(), mapOf()) + advanceUntilIdle() + engine.stop() + + assertTrue(uploader.uploadedEventNames.contains("session-end")) + } + + @Test + fun periodicFlushIntervalUploadsEvents() = runTest(testDispatcher) { + val engine = EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = uploader, + flushPolicies = mutableListOf(), + dispatcher = testDispatcher, + sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), + debugLogger = null, + flushIntervalMillis = 100 + ) + + engine.emit("interval-event", mapOf(), mapOf()) + advanceUntilIdle() + testScheduler.advanceTimeBy(150) + advanceUntilIdle() + engine.stop() + + assertTrue(uploader.uploadedEventNames.contains("interval-event")) + } + + private class RecordingEventUploader : EventSenderUploader { + val uploadedEventNames = mutableListOf() + + override suspend fun upload(events: EventBatchRequest): Boolean { + uploadedEventNames.addAll(events.events.map { it.eventDefinition.removePrefix("eventDefinitions/") }) + return true + } + } + + private class RecordingEventStorage : EventStorage { + val currentEvents = mutableListOf() + val readyEvents = mutableMapOf>() + + override suspend fun rollover() { + if (currentEvents.isNotEmpty()) { + readyEvents["batch-${readyEvents.size}"] = currentEvents.toList() + currentEvents.clear() + } + } + + override suspend fun writeEvent(event: EngineEvent) { + currentEvents.add(event) + } + + override suspend fun batchReadyFiles(): List { + return readyEvents.keys.map { java.io.File(it) } + } + + override suspend fun eventsFor(file: java.io.File): List { + return readyEvents[file.name].orEmpty() + } + + override fun onLowMemoryChannel() = kotlinx.coroutines.channels.Channel>() + + override fun stop() { + } + } +} diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt index 50310bb1..526bd67f 100644 --- a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt @@ -53,6 +53,7 @@ class ConfidenceFeatureProvider private constructor( } override fun shutdown() { + confidence.flush() } override suspend fun onContextSet( From 927028209a5d0ce5c9a4509d4de7c04de982dd1c Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Fri, 7 Aug 2026 11:03:02 +0200 Subject: [PATCH 03/15] Fix ktlint spacing around @Volatile in EventSenderEngine. --- .../src/main/java/com/spotify/confidence/EventSenderEngine.kt | 1 + 1 file changed, 1 insertion(+) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index d665b1da..62e9a365 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -49,6 +49,7 @@ internal class EventSenderEngineImpl( } } private var flushIntervalJob: Job? = null + @Volatile private var isStopped = false From edbc6af2eff85a5f73fc104bae0a6c95aef93eb8 Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Fri, 7 Aug 2026 11:09:29 +0200 Subject: [PATCH 04/15] Fix EventSenderEngine instance caching regression. Do not assign the companion Instance field; main never cached instances and tests rely on a fresh engine per ConfidenceFactory.create() call. --- .../src/main/java/com/spotify/confidence/EventSenderEngine.kt | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 62e9a365..87baed4c 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -195,9 +195,7 @@ internal class EventSenderEngineImpl( sdkMetadata = sdkMetadata, debugLogger = debugLogger, flushIntervalMillis = flushIntervalMillis - ).also { - Instance = it - } + ) } } } From e4a92b2c403f9e535c9aa6bdf69ea3740c02c17d Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Thu, 13 Aug 2026 13:57:56 +0200 Subject: [PATCH 05/15] Address review findings for track and event reliability Engine reliability fixes: - Bound stop()'s final upload with a 2s timeout so it can never block the caller indefinitely; unsent batches are retried at next startup - Serialize uploadReadyBatches with a mutex so the flush consumer, startup retry and stop() can no longer upload the same batch twice - Skip uploading empty batches, so periodic flushing on an idle app no longer makes an HTTP request per interval - Enqueue emit()/flush() synchronously on a buffered channel and drain it in stop(), so every event accepted before stop() reaches disk - Open the current events file in append mode: a restart previously truncated the prior session's unsealed events API and mapping fixes: - Log a debug warning when event data's "context" field overrides the evaluation context; deprecate the now-unthrown InvalidContextInMessage - Preserve Long/Float tracking values instead of mapping them to Null - Reject non-positive eventFlushIntervalMillis in ConfidenceFactory - Regenerate the API dump (create() overload was hand-edited, HttpError was missing) Test fixes: - Fix reliability tests: UnconfinedTestDispatcher is not a type, and advanceUntilIdle hangs forever on the self-rescheduling interval job - Make the storage fake honor deletion and seal empty batches like the real implementation; assert exact upload counts - Add tests for drain-on-stop, emit-after-stop, idle-interval no-upload and Long/Float value mapping - Reset the shared minBatchSizeFlushPolicy in setup so EventSenderIntegrationTest is no longer order-dependent Co-Authored-By: Claude Fable 5 --- Confidence/api/Confidence.api | 8 +- .../java/com/spotify/confidence/Confidence.kt | 3 + .../com/spotify/confidence/ConfidenceError.kt | 4 + .../spotify/confidence/EventSenderEngine.kt | 60 +++++--- .../com/spotify/confidence/EventStorage.kt | 4 +- .../com/spotify/confidence/PayloadMerger.kt | 8 +- .../EventSenderEngineReliabilityTest.kt | 128 ++++++++++++------ .../confidence/EventSenderIntegrationTest.kt | 7 +- .../openfeature/ConfidenceFeatureProvider.kt | 5 + .../openfeature/OpenFeatureTrackMapper.kt | 9 +- .../openfeature/OpenFeatureTrackMapperTest.kt | 12 ++ 11 files changed, 177 insertions(+), 71 deletions(-) diff --git a/Confidence/api/Confidence.api b/Confidence/api/Confidence.api index c9afb419..c1af1d2c 100644 --- a/Confidence/api/Confidence.api +++ b/Confidence/api/Confidence.api @@ -111,10 +111,10 @@ public final class com/spotify/confidence/ConfidenceError$ParseError : java/lang public final class com/spotify/confidence/ConfidenceFactory { public static final field INSTANCE Lcom/spotify/confidence/ConfidenceFactory; - public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;)Lcom/spotify/confidence/Confidence; - public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;)Lcom/spotify/confidence/Confidence; - public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; - public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; + public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;)Lcom/spotify/confidence/Confidence; + public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;)Lcom/spotify/confidence/Confidence; + public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; + public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; } public final class com/spotify/confidence/ConfidenceFlagEvaluationKt { diff --git a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt index 865aa1b2..7a36130f 100644 --- a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt +++ b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt @@ -451,6 +451,9 @@ object ConfidenceFactory { resolveBaseUrl: HttpUrl?, eventFlushIntervalMillis: Long? = null ): Confidence { + require(eventFlushIntervalMillis == null || eventFlushIntervalMillis > 0) { + "eventFlushIntervalMillis must be positive, or null to disable periodic flushing" + } val debugLogger: DebugLogger? = if (loggingLevel == LoggingLevel.NONE) { null } else { diff --git a/Confidence/src/main/java/com/spotify/confidence/ConfidenceError.kt b/Confidence/src/main/java/com/spotify/confidence/ConfidenceError.kt index f77b8f8a..be7eb1ff 100644 --- a/Confidence/src/main/java/com/spotify/confidence/ConfidenceError.kt +++ b/Confidence/src/main/java/com/spotify/confidence/ConfidenceError.kt @@ -27,5 +27,9 @@ class ConfidenceError { override val message: String ) : Error(message) + @Deprecated( + "No longer thrown: a 'context' field in event data now overrides the " + + "evaluation context for that event instead of failing" + ) class InvalidContextInMessage : Error("Field 'context' is not allowed in event's data") } diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 87baed4c..0dc9b3ad 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -4,6 +4,7 @@ import android.content.Context import com.spotify.confidence.client.Clock import com.spotify.confidence.client.Sdk import com.spotify.confidence.client.SdkMetadata +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope @@ -16,6 +17,9 @@ import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withTimeoutOrNull import okhttp3.OkHttpClient import java.io.File @@ -37,9 +41,15 @@ internal class EventSenderEngineImpl( private val debugLogger: DebugLogger?, private val flushIntervalMillis: Long? = null ) : EventSenderEngine { - private val writeReqChannel: Channel = Channel() + // Buffered so emit()/flush() can enqueue synchronously: every event accepted + // before stop() is drained to disk by the final flush. + private val writeReqChannel: Channel = Channel(Channel.UNLIMITED) private val sendChannel: Channel = Channel() - private val payloadMerger: PayloadMerger = PayloadMergerImpl() + private val payloadMerger: PayloadMerger = PayloadMergerImpl(debugLogger) + + // Serializes read-upload-delete of ready files across the flush consumer, + // the startup retry and stop(), so a batch is never uploaded twice. + private val uploadMutex = Mutex() private val coroutineScope by lazy { CoroutineScope(SupervisorJob() + dispatcher) } @@ -49,13 +59,14 @@ internal class EventSenderEngineImpl( } } private var flushIntervalJob: Job? = null + private val writeJob: Job @Volatile private var isStopped = false init { flushPolicies.add(ManualFlushPolicy) - coroutineScope.launch(exceptionHandler) { + writeJob = coroutineScope.launch(exceptionHandler) { for (event in writeReqChannel) { if (event.eventDefinition != manualFlushEvent.eventDefinition) { eventStorage.writeEvent(event) @@ -110,14 +121,13 @@ internal class EventSenderEngineImpl( if (isStopped) { return } - coroutineScope.launch { - val payload = payloadMerger(context, data) - val event = EngineEvent( - eventDefinition = eventName, - eventTime = clock.currentTime(), - payload = payload - ) - writeReqChannel.send(event) + val payload = payloadMerger(context, data) + val event = EngineEvent( + eventDefinition = eventName, + eventTime = clock.currentTime(), + payload = payload + ) + if (writeReqChannel.trySend(event).isSuccess) { debugLogger?.logEvent(action = "EmitEvent ", event = event) } } @@ -126,8 +136,7 @@ internal class EventSenderEngineImpl( if (isStopped) { return } - coroutineScope.launch { - writeReqChannel.send(manualFlushEvent) + if (writeReqChannel.trySend(manualFlushEvent).isSuccess) { debugLogger?.logEvent(action = "Flush ", event = manualFlushEvent) } } @@ -135,15 +144,22 @@ internal class EventSenderEngineImpl( override fun stop() { isStopped = true flushIntervalJob?.cancel() + // Best effort: drain queued events to disk and attempt one final upload, + // bounded so stop() can never block the caller indefinitely. Batches that + // don't make it are sealed on disk and retried at next startup. runBlocking(dispatcher) { - uploadReadyBatches(sealCurrentBatch = true) + withTimeoutOrNull(STOP_TIMEOUT_MILLIS) { + writeReqChannel.close() + writeJob.join() + uploadReadyBatches(sealCurrentBatch = true) + } } coroutineScope.cancel() eventStorage.stop() debugLogger?.logMessage(message = "EventSenderEngine closed ") } - private suspend fun uploadReadyBatches(sealCurrentBatch: Boolean) { + private suspend fun uploadReadyBatches(sealCurrentBatch: Boolean) = uploadMutex.withLock { if (sealCurrentBatch) { eventStorage.rollover() } @@ -157,24 +173,36 @@ internal class EventSenderEngineImpl( e.payload ) } + if (events.isEmpty()) { + readyFile.delete() + continue + } val batch = EventBatchRequest( clientSecret = clientSecret, events = events, sendTime = clock.currentTime(), sdk = Sdk(sdkMetadata.sdkId, sdkMetadata.sdkVersion) ) - runCatching { + try { val shouldCleanup = uploader.upload(batch) debugLogger?.logMessage(message = "Uploading events") if (shouldCleanup) { readyFile.delete() } + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + debugLogger?.logMessage( + message = "Failed to upload events: $e", + isWarning = true + ) } } } companion object { private const val SEND_SIG = "FLUSH" + private const val STOP_TIMEOUT_MILLIS = 2_000L private var Instance: EventSenderEngine? = null fun instance( context: Context, diff --git a/Confidence/src/main/java/com/spotify/confidence/EventStorage.kt b/Confidence/src/main/java/com/spotify/confidence/EventStorage.kt index cc2f94d8..3ae14d10 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventStorage.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventStorage.kt @@ -14,6 +14,7 @@ import kotlinx.serialization.json.Json import kotlinx.serialization.modules.SerializersModule import kotlinx.serialization.modules.contextual import java.io.File +import java.io.FileOutputStream import java.io.OutputStream internal interface EventStorage { @@ -115,7 +116,8 @@ internal class EventStorageImpl( outputStream?.close() currentFile = latestWriteFile() ?: getFileWithName("events-${System.currentTimeMillis()}") - outputStream = currentFile.outputStream() + // Append so events persisted by a previous session are not truncated + outputStream = FileOutputStream(currentFile, true) } private fun getFileWithName(name: String): File { val directory = context.getDir(DIRECTORY, Context.MODE_PRIVATE) diff --git a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt index 03f44494..18004d5c 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -2,10 +2,16 @@ package com.spotify.confidence private typealias ConfidenceStruct = Map internal interface PayloadMerger : (ConfidenceStruct, ConfidenceStruct) -> ConfidenceStruct -internal class PayloadMergerImpl : PayloadMerger { +internal class PayloadMergerImpl( + private val debugLogger: DebugLogger? = null +) : PayloadMerger { override fun invoke(context: ConfidenceStruct, message: ConfidenceStruct): ConfidenceStruct { return if (message.containsKey("context")) { // An explicit "context" entry in event data overrides the evaluation context for this event. + debugLogger?.logMessage( + message = "Event data contains a 'context' field: it replaces the evaluation context for this event", + isWarning = true + ) message } else { message + mapOf("context" to ConfidenceValue.Struct(context)) diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt index d67d9668..5bde8331 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -1,6 +1,10 @@ package com.spotify.confidence +import com.spotify.confidence.client.SdkMetadata +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.TestDispatcher import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest @@ -12,7 +16,7 @@ import java.util.Date @OptIn(ExperimentalCoroutinesApi::class) class EventSenderEngineReliabilityTest { - private lateinit var testDispatcher: UnconfinedTestDispatcher + private lateinit var testDispatcher: TestDispatcher private lateinit var uploader: RecordingEventUploader private lateinit var storage: RecordingEventStorage @@ -23,6 +27,20 @@ class EventSenderEngineReliabilityTest { storage = RecordingEventStorage() } + private fun engine( + dispatcher: CoroutineDispatcher = testDispatcher, + flushIntervalMillis: Long? = null + ) = EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = uploader, + flushPolicies = mutableListOf(), + dispatcher = dispatcher, + sdkMetadata = SdkMetadata("id", "1.0"), + debugLogger = null, + flushIntervalMillis = flushIntervalMillis + ) + @Test fun startupUploadsPendingReadyBatchesWithoutSealingCurrentBatch() = runTest(testDispatcher) { storage.readyEvents["pending.batch"] = listOf( @@ -32,93 +50,113 @@ class EventSenderEngineReliabilityTest { EngineEvent("current", Date(), mapOf()) ) - EventSenderEngineImpl( - eventStorage = storage, - clientSecret = "secret", - uploader = uploader, - flushPolicies = mutableListOf(), - dispatcher = testDispatcher, - sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), - debugLogger = null - ) + engine() advanceUntilIdle() assertEquals(listOf("pending"), uploader.uploadedEventNames) assertEquals(listOf("current"), storage.currentEvents.map { it.eventDefinition }) + assertTrue(storage.readyEvents.isEmpty()) } @Test fun stopUploadsCurrentBatch() = runTest(testDispatcher) { - val engine = EventSenderEngineImpl( - eventStorage = storage, - clientSecret = "secret", - uploader = uploader, - flushPolicies = mutableListOf(), - dispatcher = testDispatcher, - sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), - debugLogger = null - ) + val engine = engine() engine.emit("session-end", mapOf(), mapOf()) advanceUntilIdle() engine.stop() - assertTrue(uploader.uploadedEventNames.contains("session-end")) + assertEquals(listOf("session-end"), uploader.uploadedEventNames) } @Test - fun periodicFlushIntervalUploadsEvents() = runTest(testDispatcher) { - val engine = EventSenderEngineImpl( - eventStorage = storage, - clientSecret = "secret", - uploader = uploader, - flushPolicies = mutableListOf(), - dispatcher = testDispatcher, - sdkMetadata = com.spotify.confidence.client.SdkMetadata("id", "1.0"), - debugLogger = null, - flushIntervalMillis = 100 - ) + fun stopDrainsQueuedEventsBeforeUploading() { + val engine = engine(dispatcher = Dispatchers.IO) - engine.emit("interval-event", mapOf(), mapOf()) + repeat(50) { engine.emit("event-$it", mapOf(), mapOf()) } + engine.stop() + + assertEquals(50, uploader.uploadedEventNames.size) + } + + @Test + fun emitAfterStopIsIgnored() = runTest(testDispatcher) { + val engine = engine() + + engine.stop() + engine.emit("late-event", mapOf(), mapOf()) advanceUntilIdle() + + assertTrue(uploader.uploadedEventNames.isEmpty()) + } + + @Test + fun periodicFlushIntervalUploadsEventsExactlyOnce() = runTest(testDispatcher) { + val engine = engine(flushIntervalMillis = 100) + + engine.emit("interval-event", mapOf(), mapOf()) + testScheduler.runCurrent() + // advanceUntilIdle would spin forever on the self-rescheduling interval job testScheduler.advanceTimeBy(150) - advanceUntilIdle() + testScheduler.runCurrent() + engine.stop() + + assertEquals(listOf("interval-event"), uploader.uploadedEventNames) + } + + @Test + fun periodicFlushWithoutEventsDoesNotUpload() = runTest(testDispatcher) { + val engine = engine(flushIntervalMillis = 100) + + testScheduler.advanceTimeBy(350) + testScheduler.runCurrent() engine.stop() - assertTrue(uploader.uploadedEventNames.contains("interval-event")) + assertTrue(uploader.uploadedEventNames.isEmpty()) } private class RecordingEventUploader : EventSenderUploader { val uploadedEventNames = mutableListOf() override suspend fun upload(events: EventBatchRequest): Boolean { - uploadedEventNames.addAll(events.events.map { it.eventDefinition.removePrefix("eventDefinitions/") }) + synchronized(uploadedEventNames) { + uploadedEventNames.addAll( + events.events.map { it.eventDefinition.removePrefix("eventDefinitions/") } + ) + } return true } } + // Mimics EventStorageImpl: rollover always seals the current batch (even when + // empty) and uploaded batches disappear when their file is deleted. private class RecordingEventStorage : EventStorage { val currentEvents = mutableListOf() val readyEvents = mutableMapOf>() + private var batchCounter = 0 - override suspend fun rollover() { - if (currentEvents.isNotEmpty()) { - readyEvents["batch-${readyEvents.size}"] = currentEvents.toList() - currentEvents.clear() - } + override suspend fun rollover(): Unit = synchronized(this) { + readyEvents["batch-${batchCounter++}"] = currentEvents.toList() + currentEvents.clear() } - override suspend fun writeEvent(event: EngineEvent) { + override suspend fun writeEvent(event: EngineEvent): Unit = synchronized(this) { currentEvents.add(event) } - override suspend fun batchReadyFiles(): List { - return readyEvents.keys.map { java.io.File(it) } + override suspend fun batchReadyFiles(): List = synchronized(this) { + readyEvents.keys.map { name -> + object : java.io.File(name) { + override fun delete(): Boolean = synchronized(this@RecordingEventStorage) { + readyEvents.remove(name) != null + } + } + } } - override suspend fun eventsFor(file: java.io.File): List { - return readyEvents[file.name].orEmpty() + override suspend fun eventsFor(file: java.io.File): List = synchronized(this) { + readyEvents[file.name].orEmpty() } override fun onLowMemoryChannel() = kotlinx.coroutines.channels.Channel>() diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt index 245bb88a..0317fe43 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt @@ -37,6 +37,9 @@ class EventSenderIntegrationTest { whenever(mockSharedPrefsEdit.putString(any(), any())).thenReturn(mockSharedPrefsEdit) doNothing().whenever(mockSharedPrefsEdit).apply() eventSender = null + // minBatchSizeFlushPolicy is a shared singleton: reset its count so + // events emitted by earlier tests can't trigger a flush in this one + minBatchSizeFlushPolicy.reset() for (file in directory.walkFiles()) { file.delete() } @@ -57,9 +60,7 @@ class EventSenderIntegrationTest { ) advanceUntilIdle() val eventStorage = EventStorageImpl(mockContext) - val files = directory.walkFiles().toList() - Assert.assertEquals(1, files.size) - val events = eventStorage.eventsFor(files.first()) + val events = directory.walkFiles().toList().flatMap { eventStorage.eventsFor(it) } Assert.assertEquals(1, events.size) Assert.assertEquals( ConfidenceValue.String("override"), diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt index 526bd67f..69bcf43f 100644 --- a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt @@ -52,6 +52,11 @@ class ConfidenceFeatureProvider private constructor( } } + /** + * Triggers a best-effort flush of tracked events: delivery is asynchronous + * and not guaranteed before process death. Undelivered batches are retried + * on the next SDK startup. + */ override fun shutdown() { confidence.flush() } diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt index 4ae965ba..a7832e68 100644 --- a/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt @@ -20,9 +20,10 @@ internal fun EvaluationContext?.toTrackContextMap(): Map() val targetingKey = getTargetingKey() - if (targetingKey.isNotEmpty() && !asMap().containsKey("targeting_key")) { + if (targetingKey.isNotEmpty()) { map["targeting_key"] = ConfidenceValue.String(targetingKey) } + // Explicit attributes win over the injected targeting key map.putAll(asMap().mapValues { it.value.toConfidenceValue() }) return map } @@ -38,6 +39,12 @@ internal fun TrackingEventDetails?.toTrackingData(): Map ConfidenceValue.Integer(this) + is Long -> if (this in Int.MIN_VALUE..Int.MAX_VALUE) { + ConfidenceValue.Integer(toInt()) + } else { + ConfidenceValue.Double(toDouble()) + } is Double -> ConfidenceValue.Double(this) + is Float -> ConfidenceValue.Double(toDouble()) else -> ConfidenceValue.Null } diff --git a/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt b/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt index 87dd86c5..69793e85 100644 --- a/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt @@ -36,6 +36,18 @@ class OpenFeatureTrackMapperTest { assertEquals(ConfidenceValue.String("override"), data["value"]) } + @Test + fun trackingDetailsLongAndFloatValuesArePreserved() { + val longDetails = TrackingEventDetails(499L, ImmutableStructure()) + assertEquals(ConfidenceValue.Integer(499), longDetails.toTrackingData()["value"]) + + val bigLongDetails = TrackingEventDetails(10_000_000_000L, ImmutableStructure()) + assertEquals(ConfidenceValue.Double(1.0E10), bigLongDetails.toTrackingData()["value"]) + + val floatDetails = TrackingEventDetails(1.5f, ImmutableStructure()) + assertEquals(ConfidenceValue.Double(1.5), floatDetails.toTrackingData()["value"]) + } + @Test fun trackContextMapIncludesTargetingKey() { val context = ImmutableContext( From 644668a6f425a9acf0aeecf04410ad127f6c1803 Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Fri, 14 Aug 2026 08:06:09 +0200 Subject: [PATCH 06/15] docs: explain why writeReqChannel uses UNLIMITED capacity Clarify the rendezvous-to-buffered channel change for stop() drain reliability and trySend() semantics in response to PR review. Co-authored-by: Cursor --- .../com/spotify/confidence/EventSenderEngine.kt | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 0dc9b3ad..dfe8c4bf 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -41,8 +41,18 @@ internal class EventSenderEngineImpl( private val debugLogger: DebugLogger?, private val flushIntervalMillis: Long? = null ) : EventSenderEngine { - // Buffered so emit()/flush() can enqueue synchronously: every event accepted - // before stop() is drained to disk by the final flush. + // Main used Channel() (rendezvous, capacity 0) with suspending send() inside + // coroutineScope.launch. stop() only cancelled the scope, so in-flight emits + // could be lost. + // + // emit()/flush() now use trySend() on the caller thread so an event is either + // queued or rejected before stop() sets isStopped. stop() then closes this + // channel and joins writeJob to drain every queued event to disk. + // + // trySend on a rendezvous channel fails unless the consumer is already waiting, + // which would silently drop events whenever the writer is busy with disk I/O. + // UNLIMITED buffering guarantees trySend succeeds for all events accepted + // before stop(); see stopDrainsQueuedEventsBeforeUploading. private val writeReqChannel: Channel = Channel(Channel.UNLIMITED) private val sendChannel: Channel = Channel() private val payloadMerger: PayloadMerger = PayloadMergerImpl(debugLogger) From 73f2916a5b600c38c26c6180680c3fbbe0c8cb0f Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Mon, 17 Aug 2026 13:36:00 +0200 Subject: [PATCH 07/15] fix: non-blocking flush signals so stop() drains all events Use a conflated sendChannel with trySend so the writer loop never blocks on flush signaling while upload holds uploadMutex. Adds stopDrainsAllEventsWhenFlushPolicyBlocksWriter repro from review. Addresses vahidlazio feedback on PR #252. --- .../spotify/confidence/EventSenderEngine.kt | 9 +++- .../EventSenderEngineReliabilityTest.kt | 48 +++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index dfe8c4bf..71da72f6 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -54,7 +54,12 @@ internal class EventSenderEngineImpl( // UNLIMITED buffering guarantees trySend succeeds for all events accepted // before stop(); see stopDrainsQueuedEventsBeforeUploading. private val writeReqChannel: Channel = Channel(Channel.UNLIMITED) - private val sendChannel: Channel = Channel() + // Conflated + trySend so the writer never suspends while signaling flush. A + // rendezvous sendChannel.send() blocks the write loop during slow uploads + // (uploadMutex held), preventing writeReqChannel drain before stop() times out. + // Duplicate flush signals are harmless: uploadReadyBatches processes all + // ready files per invocation. + private val sendChannel: Channel = Channel(Channel.CONFLATED) private val payloadMerger: PayloadMerger = PayloadMergerImpl(debugLogger) // Serializes read-upload-delete of ready files across the flush consumer, @@ -93,7 +98,7 @@ internal class EventSenderEngineImpl( message = "Flush policy $policy triggered to flush. Flushing." ) } - sendChannel.send(SEND_SIG) + sendChannel.trySend(SEND_SIG) } } } diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt index 5bde8331..12667e81 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -8,6 +8,7 @@ import kotlinx.coroutines.test.TestDispatcher import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.delay import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue import org.junit.Before @@ -80,6 +81,40 @@ class EventSenderEngineReliabilityTest { assertEquals(50, uploader.uploadedEventNames.size) } + @Test + fun stopDrainsAllEventsWhenFlushPolicyBlocksWriter() { + val slowUploader = SlowEventUploader(uploadDelayMillis = 3_000) + val batchFlush = object : FlushPolicy { + private var count = 0 + override fun reset() { + count = 0 + } + override fun hit(event: EngineEvent) { + count++ + } + override fun shouldFlush(): Boolean = count > 4 + } + val engine = EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = slowUploader, + flushPolicies = mutableListOf(batchFlush), + dispatcher = Dispatchers.IO, + sdkMetadata = SdkMetadata("id", "1.0"), + debugLogger = null + ) + + repeat(12) { engine.emit("event-$it", mapOf(), mapOf()) } + Thread.sleep(100) + engine.stop() + + assertEquals( + "All 12 events should be written to storage before stop() returns", + 12, + storage.storedEventCount() + ) + } + @Test fun emitAfterStopIsIgnored() = runTest(testDispatcher) { val engine = engine() @@ -163,5 +198,18 @@ class EventSenderEngineReliabilityTest { override fun stop() { } + + fun storedEventCount(): Int = synchronized(this) { + currentEvents.size + readyEvents.values.sumOf { it.size } + } + } + + private class SlowEventUploader( + private val uploadDelayMillis: Long + ) : EventSenderUploader { + override suspend fun upload(events: EventBatchRequest): Boolean { + delay(uploadDelayMillis) + return true + } } } From e2fb4e473fdd0687d6d71236eae90fe3a3ef248b Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Mon, 17 Aug 2026 13:54:32 +0200 Subject: [PATCH 08/15] fix: resolve ktlint spacing-between-declarations-with-comments failure --- .../src/main/java/com/spotify/confidence/EventSenderEngine.kt | 1 + 1 file changed, 1 insertion(+) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 71da72f6..367ccd0f 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -54,6 +54,7 @@ internal class EventSenderEngineImpl( // UNLIMITED buffering guarantees trySend succeeds for all events accepted // before stop(); see stopDrainsQueuedEventsBeforeUploading. private val writeReqChannel: Channel = Channel(Channel.UNLIMITED) + // Conflated + trySend so the writer never suspends while signaling flush. A // rendezvous sendChannel.send() blocks the write loop during slow uploads // (uploadMutex held), preventing writeReqChannel drain before stop() times out. From aebae9a076264018ac1c885fec8d9ece8545fbbf Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Mon, 17 Aug 2026 13:58:15 +0200 Subject: [PATCH 09/15] fix: resolve ktlint import ordering in EventSenderEngineReliabilityTest --- .../com/spotify/confidence/EventSenderEngineReliabilityTest.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt index 12667e81..7bf2db91 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -4,11 +4,11 @@ import com.spotify.confidence.client.SdkMetadata import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.delay import kotlinx.coroutines.test.TestDispatcher import kotlinx.coroutines.test.UnconfinedTestDispatcher import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest -import kotlinx.coroutines.delay import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue import org.junit.Before From 7b5498956f69a5627ffcc2013cd7cf074ace1c01 Mon Sep 17 00:00:00 2001 From: Fabrizio Demaria Date: Wed, 19 Aug 2026 14:54:10 +0200 Subject: [PATCH 10/15] fix: make event shutdown durable and non-blocking Drain accepted events asynchronously before bounding network delivery, and snapshot caller-owned payload maps so queued events remain stable. Co-authored-by: Cursor --- .../spotify/confidence/EventSenderEngine.kt | 55 ++++++++-------- .../com/spotify/confidence/PayloadMerger.kt | 2 +- .../EventSenderEngineReliabilityTest.kt | 63 +++++++++++++++++-- .../spotify/confidence/PayloadMergerTest.kt | 5 +- 4 files changed, 90 insertions(+), 35 deletions(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt index 367ccd0f..6c8c8aff 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -16,7 +16,6 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withTimeoutOrNull @@ -41,25 +40,14 @@ internal class EventSenderEngineImpl( private val debugLogger: DebugLogger?, private val flushIntervalMillis: Long? = null ) : EventSenderEngine { - // Main used Channel() (rendezvous, capacity 0) with suspending send() inside - // coroutineScope.launch. stop() only cancelled the scope, so in-flight emits - // could be lost. - // - // emit()/flush() now use trySend() on the caller thread so an event is either - // queued or rejected before stop() sets isStopped. stop() then closes this - // channel and joins writeJob to drain every queued event to disk. - // - // trySend on a rendezvous channel fails unless the consumer is already waiting, - // which would silently drop events whenever the writer is busy with disk I/O. - // UNLIMITED buffering guarantees trySend succeeds for all events accepted - // before stop(); see stopDrainsQueuedEventsBeforeUploading. + // Buffering lets emit()/flush() enqueue without waiting for the disk writer. + // stop() closes the channel and the shutdown coroutine drains every accepted + // event before closing storage. private val writeReqChannel: Channel = Channel(Channel.UNLIMITED) - // Conflated + trySend so the writer never suspends while signaling flush. A - // rendezvous sendChannel.send() blocks the write loop during slow uploads - // (uploadMutex held), preventing writeReqChannel drain before stop() times out. - // Duplicate flush signals are harmless: uploadReadyBatches processes all - // ready files per invocation. + // Conflation preserves a flush signal while an upload is in progress without + // making the disk writer wait. Duplicate signals are unnecessary because each + // upload pass processes every ready batch. private val sendChannel: Channel = Channel(Channel.CONFLATED) private val payloadMerger: PayloadMerger = PayloadMergerImpl(debugLogger) @@ -157,22 +145,31 @@ internal class EventSenderEngineImpl( } } + @Synchronized override fun stop() { + if (isStopped) { + return + } isStopped = true flushIntervalJob?.cancel() - // Best effort: drain queued events to disk and attempt one final upload, - // bounded so stop() can never block the caller indefinitely. Batches that - // don't make it are sealed on disk and retried at next startup. - runBlocking(dispatcher) { - withTimeoutOrNull(STOP_TIMEOUT_MILLIS) { - writeReqChannel.close() + writeReqChannel.close() + // Shutdown stays on the engine dispatcher so callers, including Android's + // main thread, are not blocked by disk or network I/O. + coroutineScope.launch(exceptionHandler) { + try { + // Disk persistence is not timed out: every event accepted before + // stop() is sealed for delivery in this or a later session. writeJob.join() - uploadReadyBatches(sealCurrentBatch = true) + eventStorage.rollover() + withTimeoutOrNull(STOP_UPLOAD_TIMEOUT_MILLIS) { + uploadReadyBatches(sealCurrentBatch = false) + } + } finally { + coroutineScope.cancel() + eventStorage.stop() + debugLogger?.logMessage(message = "EventSenderEngine closed ") } } - coroutineScope.cancel() - eventStorage.stop() - debugLogger?.logMessage(message = "EventSenderEngine closed ") } private suspend fun uploadReadyBatches(sealCurrentBatch: Boolean) = uploadMutex.withLock { @@ -218,7 +215,7 @@ internal class EventSenderEngineImpl( companion object { private const val SEND_SIG = "FLUSH" - private const val STOP_TIMEOUT_MILLIS = 2_000L + private const val STOP_UPLOAD_TIMEOUT_MILLIS = 2_000L private var Instance: EventSenderEngine? = null fun instance( context: Context, diff --git a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt index 18004d5c..6f1c9ced 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -12,7 +12,7 @@ internal class PayloadMergerImpl( message = "Event data contains a 'context' field: it replaces the evaluation context for this event", isWarning = true ) - message + message.toMap() } else { message + mapOf("context" to ConfidenceValue.Struct(context)) } diff --git a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt index 7bf2db91..60801ad2 100644 --- a/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -78,6 +78,7 @@ class EventSenderEngineReliabilityTest { repeat(50) { engine.emit("event-$it", mapOf(), mapOf()) } engine.stop() + awaitCondition { uploader.uploadedEventNames.size == 50 } assertEquals(50, uploader.uploadedEventNames.size) } @@ -106,15 +107,42 @@ class EventSenderEngineReliabilityTest { repeat(12) { engine.emit("event-$it", mapOf(), mapOf()) } Thread.sleep(100) + val stopStartedAt = System.nanoTime() engine.stop() + val stopDurationMillis = (System.nanoTime() - stopStartedAt) / 1_000_000 + assertTrue("stop() blocked for ${stopDurationMillis}ms", stopDurationMillis < 500) + awaitCondition { storage.storedEventCount() == 12 } assertEquals( - "All 12 events should be written to storage before stop() returns", + "All 12 events should be written to storage after stop() starts shutdown", 12, storage.storedEventCount() ) } + @Test + fun stopDoesNotTimeOutDiskDrain() { + storage = RecordingEventStorage(writeDelayMillis = 2_100) + val engine = EventSenderEngineImpl( + eventStorage = storage, + clientSecret = "secret", + uploader = RetainingEventUploader(), + flushPolicies = mutableListOf(), + dispatcher = Dispatchers.IO, + sdkMetadata = SdkMetadata("id", "1.0"), + debugLogger = null + ) + + engine.emit("slow-write", mapOf(), mapOf()) + val stopStartedAt = System.nanoTime() + engine.stop() + + val stopDurationMillis = (System.nanoTime() - stopStartedAt) / 1_000_000 + assertTrue("stop() blocked for ${stopDurationMillis}ms", stopDurationMillis < 500) + awaitCondition { storage.isStopped } + assertEquals(1, storage.storedEventCount()) + } + @Test fun emitAfterStopIsIgnored() = runTest(testDispatcher) { val engine = engine() @@ -166,18 +194,29 @@ class EventSenderEngineReliabilityTest { // Mimics EventStorageImpl: rollover always seals the current batch (even when // empty) and uploaded batches disappear when their file is deleted. - private class RecordingEventStorage : EventStorage { + private class RecordingEventStorage( + private val writeDelayMillis: Long = 0 + ) : EventStorage { val currentEvents = mutableListOf() val readyEvents = mutableMapOf>() private var batchCounter = 0 + @Volatile + var isStopped = false + private set + override suspend fun rollover(): Unit = synchronized(this) { readyEvents["batch-${batchCounter++}"] = currentEvents.toList() currentEvents.clear() } - override suspend fun writeEvent(event: EngineEvent): Unit = synchronized(this) { - currentEvents.add(event) + override suspend fun writeEvent(event: EngineEvent) { + if (writeDelayMillis > 0) { + delay(writeDelayMillis) + } + synchronized(this) { + currentEvents.add(event) + } } override suspend fun batchReadyFiles(): List = synchronized(this) { @@ -197,6 +236,7 @@ class EventSenderEngineReliabilityTest { override fun onLowMemoryChannel() = kotlinx.coroutines.channels.Channel>() override fun stop() { + isStopped = true } fun storedEventCount(): Int = synchronized(this) { @@ -212,4 +252,19 @@ class EventSenderEngineReliabilityTest { return true } } + + private class RetainingEventUploader : EventSenderUploader { + override suspend fun upload(events: EventBatchRequest): Boolean = false + } + + private fun awaitCondition( + timeoutMillis: Long = 5_000, + condition: () -> Boolean + ) { + val deadline = System.nanoTime() + timeoutMillis * 1_000_000 + while (!condition() && System.nanoTime() < deadline) { + Thread.sleep(10) + } + assertTrue("Condition was not met within ${timeoutMillis}ms", condition()) + } } diff --git a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt index e8db8ddc..80f49041 100644 --- a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt @@ -27,11 +27,14 @@ class PayloadMergerTest { fun `context in data overrides evaluation context`() { val payloadMerger = PayloadMergerImpl() val context = mapOf("a" to ConfidenceValue.Integer(1), "b" to ConfidenceValue.Integer(2)) - val message = mapOf( + val message = mutableMapOf( "b" to ConfidenceValue.Integer(3), "context" to ConfidenceValue.String("override") ) val result = payloadMerger(context, message) + message["b"] = ConfidenceValue.Integer(4) + message["new"] = ConfidenceValue.String("late mutation") + assert( result == mapOf( "b" to ConfidenceValue.Integer(3), From 94f39780689c120bd3ef293a92b0f8ec7702d23c Mon Sep 17 00:00:00 2001 From: Nicklas Lundin Date: Mon, 24 Aug 2026 14:01:40 +0200 Subject: [PATCH 11/15] fix: preserve ConfidenceFactory create overloads Co-Authored-By: Codex --- Confidence/api/Confidence.api | 4 ++ .../java/com/spotify/confidence/Confidence.kt | 67 ++++++++++++++++++- 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/Confidence/api/Confidence.api b/Confidence/api/Confidence.api index c1af1d2c..949a9f25 100644 --- a/Confidence/api/Confidence.api +++ b/Confidence/api/Confidence.api @@ -111,9 +111,13 @@ public final class com/spotify/confidence/ConfidenceError$ParseError : java/lang public final class com/spotify/confidence/ConfidenceFactory { public static final field INSTANCE Lcom/spotify/confidence/ConfidenceFactory; + public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;)Lcom/spotify/confidence/Confidence; public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;)Lcom/spotify/confidence/Confidence; + public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;)Lcom/spotify/confidence/Confidence; public final fun create (Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;)Lcom/spotify/confidence/Confidence; + public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; + public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; public static synthetic fun create$default (Lcom/spotify/confidence/ConfidenceFactory;Landroid/content/Context;Ljava/lang/String;Ljava/util/Map;Lcom/spotify/confidence/ConfidenceRegion;Lkotlinx/coroutines/CoroutineDispatcher;Lcom/spotify/confidence/LoggingLevel;JLjava/lang/String;Ljava/lang/Long;ILjava/lang/Object;)Lcom/spotify/confidence/Confidence; } diff --git a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt index 7a36130f..978b6690 100644 --- a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt +++ b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt @@ -385,7 +385,40 @@ object ConfidenceFactory { * @param loggingLevel allows to print warnings or debugging information to the local console. * @param timeoutMillis sets a timeout for completing an HTTP call. Defaults to 10 seconds * @param visitorIdContextKey key to use for the visitor id in the context. Defaults to "visitor_id". - * @param eventFlushIntervalMillis optional periodic flush interval in milliseconds. Disabled by default. + */ + fun create( + context: Context, + clientSecret: String, + initialContext: Map = mapOf(), + region: ConfidenceRegion = ConfidenceRegion.GLOBAL, + dispatcher: CoroutineDispatcher = Dispatchers.IO, + loggingLevel: LoggingLevel = LoggingLevel.WARN, + timeoutMillis: Long = 10000, + visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY + ): Confidence = create( + context = context, + clientSecret = clientSecret, + initialContext = initialContext, + region = region, + dispatcher = dispatcher, + loggingLevel = loggingLevel, + timeoutMillis = timeoutMillis, + visitorIdContextKey = visitorIdContextKey, + resolveBaseUrl = null, + eventFlushIntervalMillis = null + ) + + /** + * Create a Factory Confidence instance. + * @param context application context. + * @param clientSecret confidence clientSecret, which is found in Confidence console. + * @param initialContext can be set initially, e.g. targeting_key:value. + * @param region region of operation. + * @param dispatcher coroutine dispatcher. + * @param loggingLevel allows to print warnings or debugging information to the local console. + * @param timeoutMillis sets a timeout for completing an HTTP call. Defaults to 10 seconds + * @param visitorIdContextKey key to use for the visitor id in the context. Defaults to "visitor_id". + * @param eventFlushIntervalMillis periodic flush interval in milliseconds, or null to disable. */ fun create( context: Context, @@ -396,7 +429,7 @@ object ConfidenceFactory { loggingLevel: LoggingLevel = LoggingLevel.WARN, timeoutMillis: Long = 10000, visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY, - eventFlushIntervalMillis: Long? = null + eventFlushIntervalMillis: Long? ): Confidence = create( context = context, clientSecret = clientSecret, @@ -410,6 +443,34 @@ object ConfidenceFactory { eventFlushIntervalMillis = eventFlushIntervalMillis ) + /** + * Create a Factory Confidence instance using a custom base URL for resolve and apply requests. + * The SDK appends `/v1/flags:resolve` and `/v1/flags:apply` to [resolveBaseUrl]. + * Event tracking continues to use the Confidence events endpoint. + */ + fun create( + context: Context, + clientSecret: String, + resolveBaseUrl: String, + initialContext: Map = mapOf(), + region: ConfidenceRegion = ConfidenceRegion.GLOBAL, + dispatcher: CoroutineDispatcher = Dispatchers.IO, + loggingLevel: LoggingLevel = LoggingLevel.WARN, + timeoutMillis: Long = 10000, + visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY + ): Confidence = create( + context = context, + clientSecret = clientSecret, + initialContext = initialContext, + region = region, + dispatcher = dispatcher, + loggingLevel = loggingLevel, + timeoutMillis = timeoutMillis, + visitorIdContextKey = visitorIdContextKey, + resolveBaseUrl = getResolveBaseUrl(region, resolveBaseUrl), + eventFlushIntervalMillis = null + ) + /** * Create a Factory Confidence instance using a custom base URL for resolve and apply requests. * The SDK appends `/v1/flags:resolve` and `/v1/flags:apply` to [resolveBaseUrl]. @@ -425,7 +486,7 @@ object ConfidenceFactory { loggingLevel: LoggingLevel = LoggingLevel.WARN, timeoutMillis: Long = 10000, visitorIdContextKey: String = VISITOR_ID_CONTEXT_KEY, - eventFlushIntervalMillis: Long? = null + eventFlushIntervalMillis: Long? ): Confidence = create( context = context, clientSecret = clientSecret, From 4845afb7e00156bbdccdc2f6e4040207a431cfa2 Mon Sep 17 00:00:00 2001 From: Nicklas Lundin Date: Mon, 24 Aug 2026 14:05:20 +0200 Subject: [PATCH 12/15] fix: keep explicit track off EventSender Co-Authored-By: Codex --- Confidence/api/Confidence.api | 3 +-- .../src/main/java/com/spotify/confidence/Confidence.kt | 2 +- .../main/java/com/spotify/confidence/EventSender.kt | 10 ---------- 3 files changed, 2 insertions(+), 13 deletions(-) diff --git a/Confidence/api/Confidence.api b/Confidence/api/Confidence.api index 949a9f25..9b4f73a6 100644 --- a/Confidence/api/Confidence.api +++ b/Confidence/api/Confidence.api @@ -19,7 +19,7 @@ public final class com/spotify/confidence/Confidence : com/spotify/confidence/Co public fun stop ()V public fun track (Lcom/spotify/confidence/Producer;)V public fun track (Ljava/lang/String;Ljava/util/Map;)V - public fun track (Ljava/lang/String;Ljava/util/Map;Ljava/util/Map;)V + public final fun track (Ljava/lang/String;Ljava/util/Map;Ljava/util/Map;)V public synthetic fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/Contextual; public fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/EventSender; } @@ -412,7 +412,6 @@ public abstract interface class com/spotify/confidence/EventSender : com/spotify public abstract fun stop ()V public abstract fun track (Lcom/spotify/confidence/Producer;)V public abstract fun track (Ljava/lang/String;Ljava/util/Map;)V - public abstract fun track (Ljava/lang/String;Ljava/util/Map;Ljava/util/Map;)V public abstract fun withContext (Ljava/util/Map;)Lcom/spotify/confidence/EventSender; } diff --git a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt index 978b6690..232adcc8 100644 --- a/Confidence/src/main/java/com/spotify/confidence/Confidence.kt +++ b/Confidence/src/main/java/com/spotify/confidence/Confidence.kt @@ -257,7 +257,7 @@ class Confidence internal constructor( track(eventName, data, getContext()) } - override fun track( + fun track( eventName: String, data: ConfidenceFieldsType, eventContext: Map diff --git a/Confidence/src/main/java/com/spotify/confidence/EventSender.kt b/Confidence/src/main/java/com/spotify/confidence/EventSender.kt index b6de5393..0e8ac036 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSender.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSender.kt @@ -11,16 +11,6 @@ interface EventSender : Contextual { data: ConfidenceFieldsType = mapOf() ) - /** - * Store a custom event to be tracked with an explicit evaluation context. - * @param eventContext evaluation context for this event only; does not mutate session context. - */ - fun track( - eventName: String, - data: ConfidenceFieldsType, - eventContext: Map - ) - /** * Track Android-specific events like activities or Track Context updates. * Please note that this method is collecting data in a coroutine scope and will be From 66fe3fa6836c77227a811294daedbc8a3db7f388 Mon Sep 17 00:00:00 2001 From: Nicklas Lundin Date: Mon, 24 Aug 2026 14:18:00 +0200 Subject: [PATCH 13/15] fix: stop Confidence on provider shutdown Co-Authored-By: Codex --- .../confidence/openfeature/ConfidenceFeatureProvider.kt | 7 +------ .../openfeature/ConfidenceFeatureProviderTrackTest.kt | 9 +++++++++ 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt index 69bcf43f..5f4095c5 100644 --- a/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/ConfidenceFeatureProvider.kt @@ -52,13 +52,8 @@ class ConfidenceFeatureProvider private constructor( } } - /** - * Triggers a best-effort flush of tracked events: delivery is asynchronous - * and not guaranteed before process death. Undelivered batches are retried - * on the next SDK startup. - */ override fun shutdown() { - confidence.flush() + confidence.stop() } override suspend fun onContextSet( diff --git a/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt index 889b1c34..1bb671b3 100644 --- a/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt @@ -15,6 +15,15 @@ import org.junit.Assert.assertTrue import org.junit.Test class ConfidenceFeatureProviderTrackTest { + @Test + fun shutdownStopsConfidence() { + val confidence = mockk(relaxed = true) + + ConfidenceFeatureProvider.create(confidence).shutdown() + + verify(exactly = 1) { confidence.stop() } + } + @Test fun trackForwardsMergedContextAndMappedData() { val confidence = mockk(relaxed = true) From efb33600a66bd0238ec4becc71c82806d24479e0 Mon Sep 17 00:00:00 2001 From: Nicklas Lundin Date: Mon, 24 Aug 2026 14:49:13 +0200 Subject: [PATCH 14/15] fix: snapshot event context payloads Co-Authored-By: Codex --- .../com/spotify/confidence/PayloadMerger.kt | 2 +- .../spotify/confidence/PayloadMergerTest.kt | 21 +++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt index 6f1c9ced..48884cc4 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -14,7 +14,7 @@ internal class PayloadMergerImpl( ) message.toMap() } else { - message + mapOf("context" to ConfidenceValue.Struct(context)) + message.toMap() + mapOf("context" to ConfidenceValue.Struct(context.toMap())) } } } diff --git a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt index 80f49041..73be5d5f 100644 --- a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt @@ -42,4 +42,25 @@ class PayloadMergerTest { ) ) } + + @Test + fun `merged payload snapshots message and context`() { + val payloadMerger = PayloadMergerImpl() + val context: MutableMap = mutableMapOf("a" to ConfidenceValue.Integer(1)) + val message: MutableMap = mutableMapOf("b" to ConfidenceValue.Integer(2)) + val result = payloadMerger(context, message) + context["a"] = ConfidenceValue.Integer(3) + context["new"] = ConfidenceValue.String("late context") + message["b"] = ConfidenceValue.Integer(4) + message["new"] = ConfidenceValue.String("late message") + + assert( + result == mapOf( + "b" to ConfidenceValue.Integer(2), + "context" to ConfidenceValue.Struct( + mapOf("a" to ConfidenceValue.Integer(1)) + ) + ) + ) + } } From 9f0a3db445236d7d99100ec290c1bd824e8c18b4 Mon Sep 17 00:00:00 2001 From: Nicklas Lundin Date: Mon, 24 Aug 2026 16:20:06 +0200 Subject: [PATCH 15/15] fix: snapshot nested event values Co-Authored-By: Codex --- .../com/spotify/confidence/PayloadMerger.kt | 14 ++++++-- .../spotify/confidence/PayloadMergerTest.kt | 35 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt index 48884cc4..7cb9c71c 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -12,9 +12,19 @@ internal class PayloadMergerImpl( message = "Event data contains a 'context' field: it replaces the evaluation context for this event", isWarning = true ) - message.toMap() + message.snapshot() } else { - message.toMap() + mapOf("context" to ConfidenceValue.Struct(context.toMap())) + message.snapshot() + mapOf("context" to ConfidenceValue.Struct(context.snapshot())) } } } + +private fun ConfidenceStruct.snapshot(): ConfidenceStruct = mapValues { (_, value) -> value.snapshot() } + +private fun ConfidenceValue.snapshot(): ConfidenceValue = when (this) { + is ConfidenceValue.Struct -> ConfidenceValue.Struct(map.snapshot()) + is ConfidenceValue.List -> ConfidenceValue.List(list.map { it.snapshot() }) + is ConfidenceValue.Date -> ConfidenceValue.Date(java.util.Date(date.time)) + is ConfidenceValue.Timestamp -> ConfidenceValue.Timestamp(java.util.Date(dateTime.time)) + else -> this +} diff --git a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt index 73be5d5f..ab79c941 100644 --- a/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt +++ b/Confidence/src/test/java/com/spotify/confidence/PayloadMergerTest.kt @@ -1,6 +1,7 @@ package com.spotify.confidence import org.junit.Test +import java.util.Date class PayloadMergerTest { @Test @@ -63,4 +64,38 @@ class PayloadMergerTest { ) ) } + + @Test + fun `merged payload snapshots nested mutable values`() { + val nestedContext = mutableMapOf( + "plan" to ConfidenceValue.String("free") + ) + val nestedMessage = mutableListOf(ConfidenceValue.String("original")) + val eventDate = Date(1_000) + val result = PayloadMergerImpl()( + context = mapOf( + "user" to ConfidenceValue.Struct(nestedContext), + "date" to ConfidenceValue.Date(eventDate) + ), + message = mapOf("items" to ConfidenceValue.List(nestedMessage)) + ) + + nestedContext["plan"] = ConfidenceValue.String("premium") + nestedMessage[0] = ConfidenceValue.String("changed") + eventDate.time = 2_000 + + assert( + result == mapOf( + "items" to ConfidenceValue.List(listOf(ConfidenceValue.String("original"))), + "context" to ConfidenceValue.Struct( + mapOf( + "user" to ConfidenceValue.Struct( + mapOf("plan" to ConfidenceValue.String("free")) + ), + "date" to ConfidenceValue.Date(Date(1_000)) + ) + ) + ) + ) + } }