feat: Phase 4 - EventPipeline integration with smart retry system (#300)

Smart retry system with rate-limiting and exponential backoff.

Co-authored-by: Michael Grosse Huelsewiesche <mihuelsewiesche@twilio.com>

Author: didiergarcia

core/build.gradle

@@ -19,6 +19,19 @@ compileKotlin {
 
 test {
     useJUnitPlatform()
+
+    if (System.getenv("CI") != null) {
+        // Prevent indefinite hangs in CI and surface the exact timed-out test.
+        systemProperty "junit.jupiter.execution.timeout.default", "2 m"
+    }
+
+    testLogging {
+        events "failed", "skipped"
+        exceptionFormat "full"
+        showExceptions true
+        showCauses true
+        showStackTraces true
+    }
 }
 
 dependencies {

core/src/main/java/com/segment/analytics/kotlin/core/Configuration.kt

@@ -3,6 +3,7 @@ package com.segment.analytics.kotlin.core
 import com.segment.analytics.kotlin.core.Constants.DEFAULT_API_HOST
 import com.segment.analytics.kotlin.core.Constants.DEFAULT_CDN_HOST
 import com.segment.analytics.kotlin.core.platform.policies.FlushPolicy
+import com.segment.analytics.kotlin.core.retry.HttpConfig
 import com.segment.analytics.kotlin.core.utilities.ConcreteStorageProvider
 import kotlinx.coroutines.*
 import sovran.kotlin.Store
@@ -21,6 +22,7 @@ import sovran.kotlin.Store
  * @property defaultSettings Settings object that will be used as fallback in case of network failure, defaults to empty
  * @property autoAddSegmentDestination automatically add SegmentDestination plugin, defaults to `true`
  * @property apiHost set a default apiHost to which Segment sends events, defaults to `api.segment.io/v1`
+ * @property httpConfig HTTP retry configuration for rate limiting and exponential backoff, defaults to `null` (legacy mode)
  */
 data class Configuration(
     val writeKey: String,
@@ -38,7 +40,8 @@ data class Configuration(
     var apiHost: String = DEFAULT_API_HOST,
     var cdnHost: String = DEFAULT_CDN_HOST,
     var requestFactory: RequestFactory = RequestFactory(),
-    var errorHandler: ErrorHandler? = null
+    var errorHandler: ErrorHandler? = null,
+    var httpConfig: HttpConfig? = null
 ) {
     fun isValid(): Boolean {
         return writeKey.isNotBlank() && application != null

core/src/main/java/com/segment/analytics/kotlin/core/Telemetry.kt

@@ -276,7 +276,7 @@ object Telemetry: Subscriber {
             errorHandler?.invoke(e)
             if (e.responseCode == 429) {
                 val headers = e.responseHeaders
-                val rateLimit = headers["Retry-After"]?.firstOrNull()?.toLongOrNull()
+                val rateLimit = headers["retry-after"]?.firstOrNull()?.toLongOrNull()
                 if (rateLimit != null) {
                     rateLimitEndTime = rateLimit + (System.currentTimeMillis() / 1000)
                 }

core/src/main/java/com/segment/analytics/kotlin/core/platform/EventPipeline.kt

@@ -5,6 +5,7 @@ import com.segment.analytics.kotlin.core.platform.plugins.logger.LogKind
 import com.segment.analytics.kotlin.core.platform.plugins.logger.log
 import com.segment.analytics.kotlin.core.platform.plugins.logger.segmentLog
 import com.segment.analytics.kotlin.core.platform.policies.FlushPolicy
+import com.segment.analytics.kotlin.core.retry.*
 import com.segment.analytics.kotlin.core.utilities.EncodeDefaultsJson
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
@@ -22,7 +23,9 @@ open class EventPipeline(
     private val logTag: String,
     apiKey: String,
     private val flushPolicies: List<FlushPolicy>,
-    var apiHost: String = Constants.DEFAULT_API_HOST
+    var apiHost: String = Constants.DEFAULT_API_HOST,
+    private val httpConfig: HttpConfig? = null,
+    private val timeProvider: TimeProvider = SystemTimeProvider()
 ) {
 
     private var writeChannel: Channel<BaseEvent>
@@ -39,6 +42,12 @@ open class EventPipeline(
 
     protected open val networkIODispatcher get() = analytics.networkIODispatcher
 
+    // Retry state machine for smart retry logic
+    private var retryStateMachine: RetryStateMachine
+    private var retryState: RetryState
+
+
+
     var running: Boolean
         private set
 
@@ -53,6 +62,23 @@ open class EventPipeline(
 
         writeChannel = Channel(UNLIMITED)
         uploadChannel = Channel(UNLIMITED)
+
+        // Initialize retry state machine with config (or defaults if null)
+        // Convert HttpConfig to RetryConfig (they have the same structure)
+        val retryConfig = httpConfig?.let {
+            RetryConfig(
+                rateLimitConfig = it.rateLimitConfig,
+                backoffConfig = it.backoffConfig
+            )
+        } ?: RetryConfig()
+        
+        retryStateMachine = RetryStateMachine(
+            retryConfig,
+            timeProvider
+        )
+
+        // Load persisted retry state (or start with defaults)
+        retryState = storage.loadRetryState()
     }
 
     fun put(event: BaseEvent) {
@@ -87,6 +113,18 @@ open class EventPipeline(
         unschedule()
     }
 
+    /**
+     * Update the retry configuration from CDN settings.
+     * Recreates the RetryStateMachine with the new config while preserving retry state.
+     */
+    fun updateHttpConfig(newConfig: HttpConfig) {
+        val retryConfig = RetryConfig(
+            rateLimitConfig = newConfig.rateLimitConfig,
+            backoffConfig = newConfig.backoffConfig
+        )
+        retryStateMachine = RetryStateMachine(retryConfig, timeProvider)
+    }
+
     open fun stringifyBaseEvent(payload: BaseEvent): String {
         val finalPayload = EncodeDefaultsJson.encodeToJsonElement(payload)
             .jsonObject.filterNot { (k, v) ->
@@ -129,13 +167,74 @@ open class EventPipeline(
                 storage.rollover()
             }
 
+            // Upload Gate - Check if pipeline is rate-limited
+            val currentTime = timeProvider.currentTimeMillis()
+            if (retryState.isRateLimited(currentTime)) {
+                analytics.log("$logTag skipping uploads: pipeline is rate-limited until ${retryState.waitUntilTime}")
+                return@consumeEach // Skip all uploads for this flush
+            }
+            
+            // Clear RATE_LIMITED state if wait time has passed
+            val waitTime = retryState.waitUntilTime
+            if (retryState.pipelineState == PipelineState.RATE_LIMITED &&
+                waitTime != null &&
+                currentTime >= waitTime) {
+                retryState = retryState.copy(
+                    pipelineState = PipelineState.READY,
+                    waitUntilTime = null
+                )
+            }
             val fileUrlList = parseFilePaths(storage.read(Storage.Constants.Events))
             for (url in fileUrlList) {
+                // Load batch metadata and check if we should upload
+                val (decision, updatedState) = retryStateMachine.shouldUploadBatch(
+                    retryState,
+                    url
+                )
+                retryState = updatedState
+
+                // Check if we should skip this batch
+                when (decision) {
+                    UploadDecision.SkipAllBatches -> {
+                        // This shouldn't happen here (caught by upload gate), but handle it
+                        analytics.log("$logTag skipping remaining uploads")
+                        break // Stop processing remaining files
+                    }
+                    UploadDecision.SkipThisBatch -> {
+                        analytics.log("$logTag skipping batch $url: not ready for retry")
+                        continue // Skip this file, continue with next
+                    }
+                    is UploadDecision.DropBatch -> {
+                        // Batch exceeded retry limits - delete it
+                        val reason = decision.reason
+                        analytics.log("$logTag dropping batch $url: $reason")
+                        analytics.reportInternalError(
+                            Exception("Batch dropped: $reason")
+                        )
+                        storage.removeFile(url)
+                        continue
+                    }
+                    UploadDecision.Proceed -> {
+                        // Continue with upload
+                    }
+                }
+                // Get retry count for X-Retry-Count header
+                val retryCount = retryStateMachine.getRetryCount(retryState, url)
+                
                 // upload event file
                 var shouldCleanup = true
+                var statusCode = 0
+                var retryAfterSeconds: Int? = null
+
                 storage.readAsStream(url)?.use { data ->
                     try {
                         val connection = httpClient.upload(apiHost)
+
+                        // Add X-Retry-Count header (only on retries, not first attempt)
+                        if (retryCount > 0) {
+                            connection.connection.setRequestProperty("X-Retry-Count", retryCount.toString())
+                        }
+                        
                         connection.outputStream?.let {
                             // Write the payloads into the OutputStream
                             data.copyTo(connection.outputStream)
@@ -144,21 +243,43 @@ open class EventPipeline(
                             // Upload the payloads.
                             connection.close()
                         }
-                        // Cleanup uploaded payloads
+
+                        // Success!
+                        statusCode = 200
                         analytics.log("$logTag uploaded $url")
                     } catch (e: Exception) {
                         analytics.reportInternalError(e)
+
+                        // Extract status code and retry-after from exception
+                        if (e is HTTPException) {
+                            statusCode = e.responseCode
+                            retryAfterSeconds = (e.responseHeaders["retry-after"])?.firstOrNull()?.toIntOrNull()
+                        }
+
                         shouldCleanup = handleUploadException(e, url)
                     }
                 }
 
+                // Update retry state based on response
+                val responseInfo = ResponseInfo(
+                    statusCode = if (statusCode > 0) statusCode else 500, // Default to 500 for unknown errors
+                    retryAfterSeconds = retryAfterSeconds,
+                    batchFile = url,
+                    currentTime = timeProvider.currentTimeMillis()
+                )
+                retryState = retryStateMachine.handleResponse(retryState, responseInfo)
+                
+                // Persist updated retry state
+                withContext(fileIODispatcher) {
+                    storage.saveRetryState(retryState)
+                }
+
                 if (shouldCleanup) {
                     storage.removeFile(url)
                 }
             }
         }
     }
-
     private fun schedule() {
         flushPolicies.forEach { it.schedule(analytics) }
     }
@@ -173,12 +294,15 @@ open class EventPipeline(
         var shouldCleanup = false
         if (e is HTTPException) {
             analytics.log("$logTag exception while uploading, ${e.message}")
-            if (e.is4xx() && e.responseCode != 429) {
+            if (retryStateMachine.shouldDeleteBatch(e.responseCode)) {
                 // Simply log and proceed to remove the rejected payloads from the queue.
                 Analytics.segmentLog(
                     message = "Payloads were rejected by server. Marked for removal.",
                     kind = LogKind.ERROR
                 )
+                analytics.reportInternalError(
+                    Exception("Batch dropped due to non-retryable HTTP ${e.responseCode}")
+                )
                 shouldCleanup = true
             } else {
                 Analytics.segmentLog(

core/src/main/java/com/segment/analytics/kotlin/core/platform/plugins/SegmentDestination.kt

@@ -11,6 +11,9 @@ import com.segment.analytics.kotlin.core.platform.policies.FrequencyFlushPolicy
 import com.segment.analytics.kotlin.core.retry.HttpConfig
 import kotlinx.coroutines.launch
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.jsonObject
+import kotlinx.serialization.json.jsonPrimitive
+import kotlinx.serialization.json.booleanOrNull
 import sovran.kotlin.Subscriber
 
 @Serializable
@@ -85,7 +88,8 @@ class SegmentDestination: DestinationPlugin(), VersionedPlugin, Subscriber {
                 key,
                 configuration.writeKey,
                 flushPolicies,
-                configuration.apiHost
+                configuration.apiHost,
+                configuration.httpConfig
             )
 
             analyticsScope.launch(analyticsDispatcher) {
@@ -102,10 +106,33 @@ class SegmentDestination: DestinationPlugin(), VersionedPlugin, Subscriber {
     override fun update(settings: Settings, type: Plugin.UpdateType) {
         super.update(settings, type)
         if (settings.hasIntegrationSettings(this)) {
-            // only populate the apiHost value if it exists
-            settings.destinationSettings<SegmentSettings>(key)?.apiHost?.let {
+            val segmentSettings = settings.destinationSettings<SegmentSettings>(key)
+
+            // Update apiHost if it exists
+            segmentSettings?.apiHost?.let {
                 pipeline?.apiHost = it
             }
+
+            // Read httpConfig from CDN settings and apply to pipeline
+            segmentSettings?.httpConfig?.let { cdnConfig ->
+                // CDN-sourced config defaults enabled to true (presence implies active).
+                // Only honor explicit enabled: false from CDN.
+                val rawJson = settings.integrations[key]?.jsonObject
+                val httpConfigJson = rawJson?.get("httpConfig")?.jsonObject
+
+                val rlEnabled = httpConfigJson?.get("rateLimitConfig")?.jsonObject
+                    ?.get("enabled")?.jsonPrimitive?.booleanOrNull
+                val boEnabled = httpConfigJson?.get("backoffConfig")?.jsonObject
+                    ?.get("enabled")?.jsonPrimitive?.booleanOrNull
+
+                val adjustedConfig = HttpConfig(
+                    rateLimitConfig = cdnConfig.rateLimitConfig.copy(enabled = rlEnabled ?: true),
+                    backoffConfig = cdnConfig.backoffConfig.copy(enabled = boEnabled ?: true)
+                )
+
+                analytics.configuration.httpConfig = adjustedConfig
+                pipeline?.updateHttpConfig(adjustedConfig)
+            }
         }
     }
 

core/src/main/java/com/segment/analytics/kotlin/core/platform/plugins/StartupQueue.kt

@@ -32,6 +32,7 @@ class StartupQueue : Plugin, Subscriber {
                     subscriber = this@StartupQueue,
                     stateClazz = System::class,
                     initialState = true,
+                    queue = analyticsDispatcher,
                     handler = this@StartupQueue::runningUpdate
                 )
             }

core/src/main/java/com/segment/analytics/kotlin/core/retry/RetryConfig.kt

@@ -6,32 +6,42 @@ import kotlinx.serialization.encoding.Decoder
 import kotlinx.serialization.encoding.Encoder
 import kotlinx.serialization.json.*
 
+/**
+ * Retry configuration for smart retry logic.
+ * Both rate limiting and exponential backoff default to disabled (legacy mode).
+ */
 @Serializable
 data class RetryConfig(
     val rateLimitConfig: RateLimitConfig = RateLimitConfig(),
+/**
+ * Configuration for rate limiting (429 responses).
+ * Default: disabled for backward compatibility.
+ */
     val backoffConfig: BackoffConfig = BackoffConfig()
 )
 
 @Serializable
 data class RateLimitConfig(
-    val enabled: Boolean = true,
+    val enabled: Boolean = false,
     val maxRetryCount: Int = 100,
-    val maxRetryInterval: Int = 300,
-    val maxRateLimitDuration: Long = 43200
+    val maxRetryInterval: Int = 300
 ) {
     /**
      * Validate and clamp all numeric values to safe ranges.
      */
     fun validated(): RateLimitConfig = copy(
         maxRetryCount = maxRetryCount.coerceIn(0, 1000),
-        maxRetryInterval = maxRetryInterval.coerceIn(1, 3600),
-        maxRateLimitDuration = maxRateLimitDuration.coerceIn(0, 604800)
+        maxRetryInterval = maxRetryInterval.coerceIn(1, 3600)
     )
+/**
+ * Configuration for exponential backoff on retryable errors.
+ * Default: disabled for backward compatibility.
+ */
 }
 
 @Serializable
 data class BackoffConfig(
-    val enabled: Boolean = true,
+    val enabled: Boolean = false,
     val maxRetryCount: Int = 100,
     val baseBackoffInterval: Double = 0.5,
     val maxBackoffInterval: Int = 300,