Skip to content

Commit 1311377

Browse files
didiergarciadidiergarcia
authored
feat: honor Retry-After header on any retryable HTTP response (#313)
feat: honor Retry-After header on any retryable HTTP response
1 parent 30d02fa commit 1311377

3 files changed

Lines changed: 204 additions & 6 deletions

File tree

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

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -267,8 +267,21 @@ open class EventPipeline(
267267
batchFile = url,
268268
currentTime = timeProvider.currentTimeMillis()
269269
)
270+
val previousState = retryState
270271
retryState = retryStateMachine.handleResponse(retryState, responseInfo)
271-
272+
273+
// Log when Retry-After triggers a new pipeline pause
274+
if (responseInfo.retryAfterSeconds != null &&
275+
previousState.pipelineState != PipelineState.RATE_LIMITED &&
276+
retryState.pipelineState == PipelineState.RATE_LIMITED) {
277+
val waitSeconds = responseInfo.retryAfterSeconds
278+
val waitUntilMs = retryState.waitUntilTime ?: 0L
279+
Analytics.segmentLog(
280+
message = "Retry-After (${waitSeconds}s) received on HTTP ${responseInfo.statusCode} — pausing pipeline until ${java.util.Date(waitUntilMs)}",
281+
kind = LogKind.WARNING
282+
)
283+
}
284+
272285
// Persist updated retry state
273286
withContext(fileIODispatcher) {
274287
storage.saveRetryState(retryState)

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

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -45,11 +45,19 @@ class RetryStateMachine(
4545
}
4646
}
4747

48-
behavior == RetryBehavior.RETRY && config.backoffConfig.enabled -> {
49-
handleRetryableError(state, response, currentTime)
48+
behavior == RetryBehavior.RETRY -> {
49+
when {
50+
response.retryAfterSeconds != null && config.rateLimitConfig.enabled -> {
51+
handleRateLimitResponse(state, response, currentTime)
52+
}
53+
config.backoffConfig.enabled -> {
54+
handleRetryableError(state, response, currentTime)
55+
}
56+
else -> state.removeBatch(response.batchFile)
57+
}
5058
}
5159

52-
// Drop non-retryable errors, or retryable errors when backoff is disabled
60+
// Drop non-retryable errors
5361
else -> {
5462
state.removeBatch(response.batchFile)
5563
}
@@ -189,8 +197,11 @@ class RetryStateMachine(
189197
// 429 with rate limit handling disabled: delete
190198
if (statusCode == 429) return !config.rateLimitConfig.enabled
191199
val behavior = resolveStatusCodeBehavior(statusCode)
192-
// Retryable error with backoff disabled: delete
193-
if (behavior == RetryBehavior.RETRY && !config.backoffConfig.enabled) return true
200+
if (behavior == RetryBehavior.RETRY) {
201+
if (config.rateLimitConfig.enabled) return false // rate-limit path handles it
202+
if (!config.backoffConfig.enabled) return true // no retry mechanism: delete
203+
return false
204+
}
194205
return behavior == RetryBehavior.DROP
195206
}
196207

‎core/src/test/kotlin/com/segment/analytics/kotlin/core/retry/RetryAfterGenericTest.kt‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
package com.segment.analytics.kotlin.core.retry
2+
3+
import com.segment.analytics.kotlin.core.retry.*
4+
import org.junit.jupiter.api.Assertions.*
5+
import org.junit.jupiter.api.Test
6+
7+
class RetryAfterGenericTest {
8+
9+
private fun makeConfig(maxRetryCount: Int = 100): RetryConfig = RetryConfig(
10+
rateLimitConfig = RateLimitConfig(
11+
enabled = true,
12+
maxRetryCount = maxRetryCount,
13+
maxRetryInterval = 300
14+
),
15+
backoffConfig = BackoffConfig(
16+
enabled = true,
17+
baseBackoffInterval = 0.5,
18+
maxBackoffInterval = 300,
19+
maxTotalBackoffDuration = 43200,
20+
jitterPercent = 0
21+
)
22+
)
23+
24+
private fun makeResponse(
25+
statusCode: Int,
26+
batchFile: String = "batch1",
27+
retryAfterSeconds: Int? = null,
28+
currentTime: Long = 1_000_000L
29+
) = ResponseInfo(
30+
statusCode = statusCode,
31+
batchFile = batchFile,
32+
retryAfterSeconds = retryAfterSeconds,
33+
currentTime = currentTime
34+
)
35+
36+
@Test
37+
fun `529 with Retry-After triggers pipeline-level pause`() {
38+
val machine = RetryStateMachine(makeConfig())
39+
val newState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = 30, currentTime = 1_000_000L))
40+
41+
assertEquals(PipelineState.RATE_LIMITED, newState.pipelineState)
42+
assertEquals(1_030_000L, newState.waitUntilTime)
43+
assertEquals(1, newState.globalRetryCount)
44+
assertNull(newState.batchMetadata["batch1"])
45+
}
46+
47+
@Test
48+
fun `529 with Retry-After does not increment failureCount`() {
49+
val machine = RetryStateMachine(makeConfig())
50+
val newState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = 30))
51+
assertNull(newState.batchMetadata["batch1"])
52+
}
53+
54+
@Test
55+
fun `503 with Retry-After triggers pipeline-level pause`() {
56+
val machine = RetryStateMachine(makeConfig())
57+
val newState = machine.handleResponse(RetryState(), makeResponse(503, retryAfterSeconds = 60, currentTime = 1_000_000L))
58+
59+
assertEquals(PipelineState.RATE_LIMITED, newState.pipelineState)
60+
assertEquals(1_060_000L, newState.waitUntilTime)
61+
assertEquals(1, newState.globalRetryCount)
62+
}
63+
64+
@Test
65+
fun `408 with Retry-After triggers pipeline-level pause`() {
66+
val machine = RetryStateMachine(makeConfig())
67+
val newState = machine.handleResponse(RetryState(), makeResponse(408, retryAfterSeconds = 10, currentTime = 1_000_000L))
68+
69+
assertEquals(PipelineState.RATE_LIMITED, newState.pipelineState)
70+
assertEquals(1_010_000L, newState.waitUntilTime)
71+
}
72+
73+
@Test
74+
fun `529 without Retry-After uses exponential backoff`() {
75+
val machine = RetryStateMachine(makeConfig())
76+
val newState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = null))
77+
78+
assertEquals(PipelineState.READY, newState.pipelineState)
79+
assertNull(newState.waitUntilTime)
80+
assertEquals(0, newState.globalRetryCount)
81+
assertNotNull(newState.batchMetadata["batch1"])
82+
assertEquals(1, newState.batchMetadata["batch1"]?.failureCount)
83+
}
84+
85+
@Test
86+
fun `503 without Retry-After uses exponential backoff`() {
87+
val machine = RetryStateMachine(makeConfig())
88+
val newState = machine.handleResponse(RetryState(), makeResponse(503, retryAfterSeconds = null))
89+
90+
assertEquals(PipelineState.READY, newState.pipelineState)
91+
assertNotNull(newState.batchMetadata["batch1"])
92+
}
93+
94+
@Test
95+
fun `Retry-After clamped at maxRetryInterval`() {
96+
val machine = RetryStateMachine(makeConfig())
97+
val newState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = 99999, currentTime = 1_000_000L))
98+
assertEquals(1_300_000L, newState.waitUntilTime)
99+
}
100+
101+
@Test
102+
fun `Retry-After zero sets waitUntilTime to now`() {
103+
val machine = RetryStateMachine(makeConfig())
104+
val newState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = 0, currentTime = 1_000_000L))
105+
106+
assertEquals(PipelineState.RATE_LIMITED, newState.pipelineState)
107+
assertEquals(1_000_000L, newState.waitUntilTime)
108+
}
109+
110+
@Test
111+
fun `globalRetryCount cap drops batch after maxRetryCount`() {
112+
val config = makeConfig(maxRetryCount = 3)
113+
val machine = RetryStateMachine(config)
114+
115+
var state = RetryState()
116+
for (i in 1..3) {
117+
state = machine.handleResponse(state, makeResponse(529, retryAfterSeconds = 1, currentTime = i * 100_000L))
118+
}
119+
assertEquals(3, state.globalRetryCount)
120+
121+
val timeProvider = FakeTimeProvider(currentTime = 10_000_000L)
122+
val dropMachine = RetryStateMachine(config, timeProvider)
123+
val (decision, _) = dropMachine.shouldUploadBatch(state, "batch1")
124+
assertEquals(UploadDecision.DropBatch(DropReason.MAX_RETRIES_EXCEEDED), decision)
125+
}
126+
127+
@Test
128+
fun `400 with Retry-After drops immediately`() {
129+
val machine = RetryStateMachine(makeConfig())
130+
val newState = machine.handleResponse(RetryState(), makeResponse(400, retryAfterSeconds = 60))
131+
132+
assertEquals(PipelineState.READY, newState.pipelineState)
133+
assertNull(newState.batchMetadata["batch1"])
134+
assertNull(newState.waitUntilTime)
135+
assertEquals(0, newState.globalRetryCount)
136+
}
137+
138+
@Test
139+
fun `429 behavior unchanged`() {
140+
val machine = RetryStateMachine(makeConfig())
141+
val newState = machine.handleResponse(RetryState(), makeResponse(429, retryAfterSeconds = 45, currentTime = 1_000_000L))
142+
143+
assertEquals(PipelineState.RATE_LIMITED, newState.pipelineState)
144+
assertEquals(1_045_000L, newState.waitUntilTime)
145+
assertEquals(1, newState.globalRetryCount)
146+
assertNull(newState.batchMetadata["batch1"])
147+
}
148+
149+
@Test
150+
fun `shouldDeleteBatch returns false for retryable code when rateLimitConfig enabled`() {
151+
val config = RetryConfig(
152+
rateLimitConfig = RateLimitConfig(enabled = true),
153+
backoffConfig = BackoffConfig(enabled = false)
154+
)
155+
val machine = RetryStateMachine(config)
156+
assertFalse(machine.shouldDeleteBatch(503))
157+
assertFalse(machine.shouldDeleteBatch(529))
158+
}
159+
160+
@Test
161+
fun `pipeline pause holds all batches`() {
162+
val config = makeConfig()
163+
val timeProvider = FakeTimeProvider(currentTime = 1_000_000L)
164+
val machine = RetryStateMachine(config, timeProvider)
165+
166+
val pausedState = machine.handleResponse(RetryState(), makeResponse(529, retryAfterSeconds = 30, currentTime = 1_000_000L))
167+
168+
val (decision1, _) = machine.shouldUploadBatch(pausedState, "batch1")
169+
val (decision2, _) = machine.shouldUploadBatch(pausedState, "batch2")
170+
171+
assertEquals(UploadDecision.SkipAllBatches, decision1)
172+
assertEquals(UploadDecision.SkipAllBatches, decision2)
173+
}
174+
}

0 commit comments

Comments
 (0)