Skip to content

Commit 7561305

Browse files
polbinspolbins
authored
fix: 4xx-handling-when-on-proxy (#262)
* move parse events to a function * remove single event special case for bad request (400) * move shouldRetryUploadOnFailure to ResponseHandler * treat 400 errors with no invalid fields as retryable * use handleBadRequestResponse from interface * don't modify event file when there's nothing to drop - this prevents changing the order of events when this scenario happens
1 parent 789e261 commit 7561305

File tree

8 files changed

+307
-108
lines changed

8 files changed

+307
-108
lines changed

‎android/src/test/java/com/amplitude/android/ResponseHandlerTest.kt renamed to ‎android/src/test/java/com/amplitude/android/ResponseHandlerIntegrationTest.kt

+3-3
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ private const val FLUSH_MAX_RETRIES = 3
3636

3737
@ExperimentalCoroutinesApi
3838
@RunWith(RobolectricTestRunner::class)
39-
class ResponseHandlerTest {
39+
class ResponseHandlerIntegrationTest {
4040
private lateinit var server: MockWebServer
4141
private lateinit var amplitude: Amplitude
4242

@@ -203,7 +203,7 @@ class ResponseHandlerTest {
203203
}
204204

205205
@Test
206-
fun `test handle bad request response`() = runTest {
206+
fun `test handle bad request response - missing field`() = runTest {
207207
setAmplitudeDispatchers(amplitude, testScheduler)
208208
val badRequestResponseBody = """
209209
{
@@ -216,7 +216,7 @@ class ResponseHandlerTest {
216216
},
217217
"events_with_missing_fields": {
218218
"event_type": [
219-
2
219+
0
220220
]
221221
}
222222
}

‎core/src/main/java/com/amplitude/core/platform/EventPipeline.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -143,11 +143,11 @@ class EventPipeline(
143143

144144
val diagnostics = amplitude.diagnostics.extractDiagnostics()
145145
val response = httpClient.upload(eventsString, diagnostics)
146-
responseHandler.handle(response, eventFile, eventsString)
146+
val shouldRetryUploadOnFailure = responseHandler.handle(response, eventFile, eventsString)
147147

148148
// if we encounter a retryable error, we retry with delay and
149149
// restart the loop to get the newest event files
150-
if (response.status.shouldRetryUploadOnFailure == true) {
150+
if (shouldRetryUploadOnFailure == true) {
151151
retryUploadHandler.attemptRetry { canRetry ->
152152
val retrySignal = if (canRetry) UPLOAD_SIG else MAX_RETRY_ATTEMPT_SIG
153153
uploadChannel.trySend(retrySignal)

‎core/src/main/java/com/amplitude/core/utilities/FileResponseHandler.kt

+42-33
Original file line numberDiff line numberDiff line change
@@ -34,16 +34,7 @@ class FileResponseHandler(
3434
) {
3535
val eventFilePath = events as String
3636
logger?.debug("Handle response, status: ${successResponse.status}")
37-
val eventsList: List<BaseEvent>
38-
try {
39-
eventsList = JSONArray(eventsString).toEvents()
40-
} catch (e: JSONException) {
41-
scope.launch(storageDispatcher) {
42-
storage.removeFile(eventFilePath)
43-
}
44-
removeCallbackByInsertId(eventsString)
45-
throw e
46-
}
37+
val eventsList = parseEvents(eventsString, eventFilePath).toEvents()
4738
triggerEventsCallback(eventsList, HttpStatus.SUCCESS.code, "Event sent success.")
4839
scope.launch(storageDispatcher) {
4940
storage.removeFile(eventFilePath)
@@ -54,27 +45,18 @@ class FileResponseHandler(
5445
badRequestResponse: BadRequestResponse,
5546
events: Any,
5647
eventsString: String,
57-
) {
48+
): Boolean {
5849
logger?.debug(
5950
"Handle response, status: ${badRequestResponse.status}, error: ${badRequestResponse.error}"
6051
)
6152
val eventFilePath = events as String
62-
val eventsList: List<BaseEvent>
63-
try {
64-
eventsList = JSONArray(eventsString).toEvents()
65-
} catch (e: JSONException) {
66-
scope.launch(storageDispatcher) {
67-
storage.removeFile(eventFilePath)
68-
}
69-
removeCallbackByInsertId(eventsString)
70-
throw e
71-
}
72-
if (eventsList.size == 1 || badRequestResponse.isInvalidApiKeyResponse()) {
53+
val eventsList = parseEvents(eventsString, eventFilePath).toEvents()
54+
if (badRequestResponse.isInvalidApiKeyResponse()) {
7355
triggerEventsCallback(eventsList, HttpStatus.BAD_REQUEST.code, badRequestResponse.error)
7456
scope.launch(storageDispatcher) {
7557
storage.removeFile(eventFilePath)
7658
}
77-
return
59+
return false
7860
}
7961
val droppedIndices = badRequestResponse.getEventIndicesToDrop()
8062
val eventsToDrop = mutableListOf<BaseEvent>()
@@ -86,13 +68,27 @@ class FileResponseHandler(
8668
eventsToRetry.add(event)
8769
}
8870
}
71+
// shouldRetryUploadOnFailure is true if there are NO events to drop, this happens
72+
// when connected to a proxy and it returns 400 with w/o the eventsToDrop fields
73+
if (eventsToDrop.isEmpty()) {
74+
scope.launch(storageDispatcher) {
75+
storage.releaseFile(events)
76+
}
77+
return true
78+
}
79+
8980
triggerEventsCallback(eventsToDrop, HttpStatus.BAD_REQUEST.code, badRequestResponse.error)
9081
eventsToRetry.forEach {
9182
eventPipeline.put(it)
9283
}
9384
scope.launch(storageDispatcher) {
85+
logger?.debug(
86+
"--> remove file: ${eventFilePath.split("-").takeLast(2)}, dropped events: ${eventsToDrop.size}, " +
87+
"retry events: ${eventsToRetry.size}"
88+
)
9489
storage.removeFile(eventFilePath)
9590
}
91+
return false
9692
}
9793

9894
override fun handlePayloadTooLargeResponse(
@@ -104,16 +100,7 @@ class FileResponseHandler(
104100
"Handle response, status: ${payloadTooLargeResponse.status}, error: ${payloadTooLargeResponse.error}"
105101
)
106102
val eventFilePath = events as String
107-
val rawEvents: JSONArray
108-
try {
109-
rawEvents = JSONArray(eventsString)
110-
} catch (e: JSONException) {
111-
scope.launch(storageDispatcher) {
112-
storage.removeFile(eventFilePath)
113-
}
114-
removeCallbackByInsertId(eventsString)
115-
throw e
116-
}
103+
val rawEvents = parseEvents(eventsString, eventFilePath)
117104
if (rawEvents.length() == 1) {
118105
val eventsList = rawEvents.toEvents()
119106
triggerEventsCallback(
@@ -168,6 +155,28 @@ class FileResponseHandler(
168155
}
169156
}
170157

158+
/**
159+
* Parse events from the [eventsString] at the given [eventFilePath].
160+
* If parsing fails, this removes the file at [eventFilePath], and
161+
* remove the callback by insert ID, and throws a [JSONException].
162+
*/
163+
private fun parseEvents(
164+
eventsString: String,
165+
eventFilePath: String,
166+
): JSONArray {
167+
val rawEvents: JSONArray
168+
try {
169+
rawEvents = JSONArray(eventsString)
170+
} catch (e: JSONException) {
171+
scope.launch(storageDispatcher) {
172+
storage.removeFile(eventFilePath)
173+
}
174+
removeCallbackByInsertId(eventsString)
175+
throw e
176+
}
177+
return rawEvents
178+
}
179+
171180
private fun triggerEventsCallback(
172181
events: List<BaseEvent>,
173182
status: Int,

‎core/src/main/java/com/amplitude/core/utilities/InMemoryResponseHandler.kt

+52-17
Original file line numberDiff line numberDiff line change
@@ -17,25 +17,35 @@ import kotlinx.coroutines.delay
1717
import kotlinx.coroutines.launch
1818

1919
internal class InMemoryResponseHandler(
20-
val eventPipeline: EventPipeline,
21-
val configuration: Configuration,
22-
val scope: CoroutineScope,
23-
val dispatcher: CoroutineDispatcher
20+
private val eventPipeline: EventPipeline,
21+
private val configuration: Configuration,
22+
private val scope: CoroutineScope,
23+
private val storageDispatcher: CoroutineDispatcher,
2424
) : ResponseHandler {
2525

2626
companion object {
2727
const val BACK_OFF: Long = 30000
2828
}
2929

30-
override fun handleSuccessResponse(successResponse: SuccessResponse, events: Any, eventsString: String) {
31-
triggerEventsCallback(events as List<BaseEvent>, HttpStatus.SUCCESS.code, "Event sent success.")
30+
override fun handleSuccessResponse(
31+
successResponse: SuccessResponse,
32+
events: Any,
33+
eventsString: String,
34+
) {
35+
triggerEventsCallback(
36+
events as List<BaseEvent>, HttpStatus.SUCCESS.code, "Event sent success."
37+
)
3238
}
3339

34-
override fun handleBadRequestResponse(badRequestResponse: BadRequestResponse, events: Any, eventsString: String) {
40+
override fun handleBadRequestResponse(
41+
badRequestResponse: BadRequestResponse,
42+
events: Any,
43+
eventsString: String,
44+
): Boolean {
3545
val eventsList = events as List<BaseEvent>
3646
if (eventsList.size == 1 || badRequestResponse.isInvalidApiKeyResponse()) {
3747
triggerEventsCallback(eventsList, HttpStatus.BAD_REQUEST.code, badRequestResponse.error)
38-
return
48+
return false
3949
}
4050
val droppedIndices = badRequestResponse.getEventIndicesToDrop()
4151
val eventsToDrop = mutableListOf<BaseEvent>()
@@ -51,12 +61,19 @@ internal class InMemoryResponseHandler(
5161
eventsToRetry.forEach {
5262
eventPipeline.put(it)
5363
}
64+
return eventsToDrop.isEmpty()
5465
}
5566

56-
override fun handlePayloadTooLargeResponse(payloadTooLargeResponse: PayloadTooLargeResponse, events: Any, eventsString: String) {
67+
override fun handlePayloadTooLargeResponse(
68+
payloadTooLargeResponse: PayloadTooLargeResponse,
69+
events: Any,
70+
eventsString: String,
71+
) {
5772
val eventsList = events as List<BaseEvent>
5873
if (eventsList.size == 1) {
59-
triggerEventsCallback(eventsList, HttpStatus.PAYLOAD_TOO_LARGE.code, payloadTooLargeResponse.error)
74+
triggerEventsCallback(
75+
eventsList, HttpStatus.PAYLOAD_TOO_LARGE.code, payloadTooLargeResponse.error
76+
)
6077
return
6178
}
6279
eventPipeline.flushSizeDivider.incrementAndGet()
@@ -65,7 +82,11 @@ internal class InMemoryResponseHandler(
6582
}
6683
}
6784

68-
override fun handleTooManyRequestsResponse(tooManyRequestsResponse: TooManyRequestsResponse, events: Any, eventsString: String) {
85+
override fun handleTooManyRequestsResponse(
86+
tooManyRequestsResponse: TooManyRequestsResponse,
87+
events: Any,
88+
eventsString: String,
89+
) {
6990
val eventsList = events as List<BaseEvent>
7091
val eventsToDrop = mutableListOf<BaseEvent>()
7192
val eventsToRetryNow = mutableListOf<BaseEvent>()
@@ -79,29 +100,39 @@ internal class InMemoryResponseHandler(
79100
eventsToRetryNow.add(event)
80101
}
81102
}
82-
triggerEventsCallback(eventsToDrop, HttpStatus.TOO_MANY_REQUESTS.code, tooManyRequestsResponse.error)
103+
triggerEventsCallback(
104+
eventsToDrop, HttpStatus.TOO_MANY_REQUESTS.code, tooManyRequestsResponse.error
105+
)
83106
eventsToRetryNow.forEach {
84107
eventPipeline.put(it)
85108
}
86-
scope.launch(dispatcher) {
109+
scope.launch(storageDispatcher) {
87110
delay(BACK_OFF)
88111
eventsToRetryLater.forEach {
89112
eventPipeline.put(it)
90113
}
91114
}
92115
}
93116

94-
override fun handleTimeoutResponse(timeoutResponse: TimeoutResponse, events: Any, eventsString: String) {
117+
override fun handleTimeoutResponse(
118+
timeoutResponse: TimeoutResponse,
119+
events: Any,
120+
eventsString: String,
121+
) {
95122
val eventsList = events as List<BaseEvent>
96-
scope.launch(dispatcher) {
123+
scope.launch(storageDispatcher) {
97124
delay(BACK_OFF)
98125
eventsList.forEach {
99126
eventPipeline.put(it)
100127
}
101128
}
102129
}
103130

104-
override fun handleFailedResponse(failedResponse: FailedResponse, events: Any, eventsString: String) {
131+
override fun handleFailedResponse(
132+
failedResponse: FailedResponse,
133+
events: Any,
134+
eventsString: String,
135+
) {
105136
val eventsList = events as List<BaseEvent>
106137
val eventsToDrop = mutableListOf<BaseEvent>()
107138
val eventsToRetry = mutableListOf<BaseEvent>()
@@ -118,7 +149,11 @@ internal class InMemoryResponseHandler(
118149
}
119150
}
120151

121-
private fun triggerEventsCallback(events: List<BaseEvent>, status: Int, message: String) {
152+
private fun triggerEventsCallback(
153+
events: List<BaseEvent>,
154+
status: Int,
155+
message: String,
156+
) {
122157
events.forEach { event ->
123158
configuration.callback?.let {
124159
it(event, status, message)

0 commit comments

Comments
 (0)