diff --git a/Confidence/api/Confidence.api b/Confidence/api/Confidence.api index 8535a5ed..9b4f73a6 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 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; } @@ -111,9 +112,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; } 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 3fece006..232adcc8 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()) + } + + fun track( + eventName: String, + data: ConfidenceFieldsType, + eventContext: Map + ) { + eventSenderEngine.emit(eventName, data, eventContext) } override fun flush() { @@ -396,7 +404,43 @@ object ConfidenceFactory { loggingLevel = loggingLevel, timeoutMillis = timeoutMillis, visitorIdContextKey = visitorIdContextKey, - resolveBaseUrl = null + 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, + 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, + eventFlushIntervalMillis: Long? + ): Confidence = create( + context = context, + clientSecret = clientSecret, + initialContext = initialContext, + region = region, + dispatcher = dispatcher, + loggingLevel = loggingLevel, + timeoutMillis = timeoutMillis, + visitorIdContextKey = visitorIdContextKey, + resolveBaseUrl = null, + eventFlushIntervalMillis = eventFlushIntervalMillis ) /** @@ -423,7 +467,37 @@ object ConfidenceFactory { loggingLevel = loggingLevel, timeoutMillis = timeoutMillis, visitorIdContextKey = visitorIdContextKey, - resolveBaseUrl = getResolveBaseUrl(region, resolveBaseUrl) + 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]. + * 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, + eventFlushIntervalMillis: Long? + ): Confidence = create( + context = context, + clientSecret = clientSecret, + initialContext = initialContext, + region = region, + dispatcher = dispatcher, + loggingLevel = loggingLevel, + timeoutMillis = timeoutMillis, + visitorIdContextKey = visitorIdContextKey, + resolveBaseUrl = getResolveBaseUrl(region, resolveBaseUrl), + eventFlushIntervalMillis = eventFlushIntervalMillis ) private fun create( @@ -435,8 +509,12 @@ object ConfidenceFactory { loggingLevel: LoggingLevel, timeoutMillis: Long, visitorIdContextKey: String, - resolveBaseUrl: HttpUrl? + 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 { @@ -450,7 +528,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/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 877729ec..6c8c8aff 100644 --- a/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt +++ b/Confidence/src/main/java/com/spotify/confidence/EventSenderEngine.kt @@ -4,14 +4,21 @@ 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 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.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withTimeoutOrNull import okhttp3.OkHttpClient import java.io.File @@ -30,11 +37,23 @@ 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() - private val payloadMerger: PayloadMerger = PayloadMergerImpl() + // 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) + + // 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) + + // 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) } @@ -43,13 +62,17 @@ internal class EventSenderEngineImpl( debugLogger?.logMessage(message = "EventSenderEngine error: $e", isWarning = true) } } + 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) { - // skip storing manual flush event eventStorage.writeEvent(event) debugLogger?.logEvent(action = "DiskWrite ", event = event) } @@ -64,41 +87,29 @@ internal class EventSenderEngineImpl( message = "Flush policy $policy triggered to flush. Flushing." ) } - sendChannel.send(SEND_SIG) + sendChannel.trySend(SEND_SIG) } } } - // 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,33 +122,100 @@ internal class EventSenderEngineImpl( data: ConfidenceFieldsType, context: Map ) { - coroutineScope.launch { - val payload = payloadMerger(context, data) - val event = EngineEvent( - eventDefinition = eventName, - eventTime = clock.currentTime(), - payload = payload - ) - writeReqChannel.send(event) + if (isStopped) { + return + } + 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) } } override fun flush() { - coroutineScope.launch { - writeReqChannel.send(manualFlushEvent) + if (isStopped) { + return + } + if (writeReqChannel.trySend(manualFlushEvent).isSuccess) { debugLogger?.logEvent(action = "Flush ", event = manualFlushEvent) } } + @Synchronized override fun stop() { - coroutineScope.cancel() - eventStorage.stop() - debugLogger?.logMessage(message = "EventSenderEngine closed ") + if (isStopped) { + return + } + isStopped = true + flushIntervalJob?.cancel() + 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() + eventStorage.rollover() + withTimeoutOrNull(STOP_UPLOAD_TIMEOUT_MILLIS) { + uploadReadyBatches(sealCurrentBatch = false) + } + } finally { + coroutineScope.cancel() + eventStorage.stop() + debugLogger?.logMessage(message = "EventSenderEngine closed ") + } + } + } + + private suspend fun uploadReadyBatches(sealCurrentBatch: Boolean) = uploadMutex.withLock { + 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 + ) + } + if (events.isEmpty()) { + readyFile.delete() + continue + } + val batch = EventBatchRequest( + clientSecret = clientSecret, + events = events, + sendTime = clock.currentTime(), + sdk = Sdk(sdkMetadata.sdkId, sdkMetadata.sdkVersion) + ) + 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_UPLOAD_TIMEOUT_MILLIS = 2_000L private var Instance: EventSenderEngine? = null fun instance( context: Context, @@ -145,7 +223,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,7 +234,8 @@ internal class EventSenderEngineImpl( flushPolicies = flushPolicies.toMutableList(), dispatcher = dispatcher, sdkMetadata = sdkMetadata, - debugLogger = debugLogger + debugLogger = debugLogger, + flushIntervalMillis = flushIntervalMillis ) } } 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 52449d68..7cb9c71c 100644 --- a/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt +++ b/Confidence/src/main/java/com/spotify/confidence/PayloadMerger.kt @@ -2,11 +2,29 @@ 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 { - 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. + debugLogger?.logMessage( + message = "Event data contains a 'context' field: it replaces the evaluation context for this event", + isWarning = true + ) + message.snapshot() + } else { + message.snapshot() + mapOf("context" to ConfidenceValue.Struct(context.snapshot())) } - return message + (mapOf("context" to ConfidenceValue.Struct(context))) } } + +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/EventSenderEngineReliabilityTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt new file mode 100644 index 00000000..60801ad2 --- /dev/null +++ b/Confidence/src/test/java/com/spotify/confidence/EventSenderEngineReliabilityTest.kt @@ -0,0 +1,270 @@ +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.delay +import kotlinx.coroutines.test.TestDispatcher +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: TestDispatcher + private lateinit var uploader: RecordingEventUploader + private lateinit var storage: RecordingEventStorage + + @Before + fun setUp() { + testDispatcher = UnconfinedTestDispatcher() + uploader = RecordingEventUploader() + 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( + EngineEvent("pending", Date(), mapOf()) + ) + storage.currentEvents.add( + EngineEvent("current", Date(), mapOf()) + ) + + 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 = engine() + + engine.emit("session-end", mapOf(), mapOf()) + advanceUntilIdle() + engine.stop() + + assertEquals(listOf("session-end"), uploader.uploadedEventNames) + } + + @Test + fun stopDrainsQueuedEventsBeforeUploading() { + val engine = engine(dispatcher = Dispatchers.IO) + + repeat(50) { engine.emit("event-$it", mapOf(), mapOf()) } + engine.stop() + + awaitCondition { uploader.uploadedEventNames.size == 50 } + 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) + 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 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() + + 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) + 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.isEmpty()) + } + + private class RecordingEventUploader : EventSenderUploader { + val uploadedEventNames = mutableListOf() + + override suspend fun upload(events: EventBatchRequest): Boolean { + 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( + 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) { + if (writeDelayMillis > 0) { + delay(writeDelayMillis) + } + synchronized(this) { + currentEvents.add(event) + } + } + + 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 = synchronized(this) { + readyEvents[file.name].orEmpty() + } + + override fun onLowMemoryChannel() = kotlinx.coroutines.channels.Channel>() + + override fun stop() { + isStopped = true + } + + 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 + } + } + + 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/EventSenderIntegrationTest.kt b/Confidence/src/test/java/com/spotify/confidence/EventSenderIntegrationTest.kt index 12bb21a7..0317fe43 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 @@ -38,20 +37,35 @@ 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() } } - @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 events = directory.walkFiles().toList().flatMap { eventStorage.eventsFor(it) } + 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..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 @@ -22,4 +23,79 @@ 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 = 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), + "context" to ConfidenceValue.String("override") + ) + ) + } + + @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)) + ) + ) + ) + } + + @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)) + ) + ) + ) + ) + } } 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..5f4095c5 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.stop() } override suspend fun onContextSet( @@ -113,7 +114,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 +166,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..a7832e68 --- /dev/null +++ b/Provider/src/main/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapper.kt @@ -0,0 +1,50 @@ +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()) { + map["targeting_key"] = ConfidenceValue.String(targetingKey) + } + // Explicit attributes win over the injected targeting key + 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 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/ConfidenceFeatureProviderTrackTest.kt b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt new file mode 100644 index 00000000..1bb671b3 --- /dev/null +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/ConfidenceFeatureProviderTrackTest.kt @@ -0,0 +1,127 @@ +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 shutdownStopsConfidence() { + val confidence = mockk(relaxed = true) + + ConfidenceFeatureProvider.create(confidence).shutdown() + + verify(exactly = 1) { confidence.stop() } + } + + @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..69793e85 --- /dev/null +++ b/Provider/src/test/java/com/spotify/confidence/openfeature/OpenFeatureTrackMapperTest.kt @@ -0,0 +1,61 @@ +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 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( + 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"]) + } +}