Block on payload disk write at intake when doing resurrection

Author: bidetofevil

embrace-android-core/src/main/kotlin/io/embrace/android/embracesdk/internal/resurrection/PayloadResurrectionServiceImpl.kt

@@ -34,6 +34,8 @@ import io.embrace.android.embracesdk.internal.session.getSessionSpan
 import io.embrace.android.embracesdk.internal.utils.Provider
 import io.opentelemetry.kotlin.semconv.SessionAttributes
 import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
 import java.util.zip.GZIPInputStream
 import kotlin.math.max
 
@@ -222,11 +224,17 @@ internal class PayloadResurrectionServiceImpl(
             else -> null
         }
 
+        // Synchronously provide the payload to the IntakeService, blocking on the returned Future
         if (resurrectedPayload != null) {
-            intakeService.take(
+            val task = intakeService.take(
                 intake = resurrectedPayload,
                 metadata = copy(complete = true)
             )
+            try {
+                task.get(5, TimeUnit.SECONDS)
+            } catch (e: TimeoutException) {
+                logger.trackInternalError(InternalErrorType.INTAKE_FAIL, e)
+            }
         }
     }
 

embrace-android-delivery-fakes/src/main/kotlin/io/embrace/android/embracesdk/fakes/FakeIntakeService.kt

@@ -3,6 +3,8 @@ package io.embrace.android.embracesdk.fakes
 import io.embrace.android.embracesdk.internal.delivery.StoredTelemetryMetadata
 import io.embrace.android.embracesdk.internal.delivery.intake.IntakeService
 import io.embrace.android.embracesdk.internal.payload.Envelope
+import java.util.concurrent.Future
+import java.util.concurrent.TimeUnit
 
 class FakeIntakeService : IntakeService {
 
@@ -23,11 +25,24 @@ class FakeIntakeService : IntakeService {
         shutdownCount++
     }
 
-    override fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata) {
+    override fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata): Future<*> {
         val dst = when (metadata.complete) {
             true -> intakeList
             false -> cacheList
         }
         dst.add(FakePayloadIntake(intake, metadata))
+        return fakeFuture
+    }
+
+    val fakeFuture = object : Future<Boolean> {
+        override fun cancel(p0: Boolean): Boolean = false
+
+        override fun isCancelled(): Boolean = false
+
+        override fun isDone(): Boolean = true
+
+        override fun get() = true
+
+        override fun get(p0: Long, p1: TimeUnit) = true
     }
 }

embrace-android-delivery/src/main/kotlin/io/embrace/android/embracesdk/internal/delivery/intake/IntakeService.kt

@@ -3,6 +3,7 @@ package io.embrace.android.embracesdk.internal.delivery.intake
 import io.embrace.android.embracesdk.internal.delivery.Shutdownable
 import io.embrace.android.embracesdk.internal.delivery.StoredTelemetryMetadata
 import io.embrace.android.embracesdk.internal.payload.Envelope
+import java.util.concurrent.Future
 
 /**
  * IntakeService is responsible for storing telemetry payloads to disk and notifying the
@@ -19,5 +20,5 @@ interface IntakeService : Shutdownable {
     /**
      * Stores a payload on disk as its JSON representation.
      */
-    fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata)
+    fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata): Future<*>
 }

embrace-android-delivery/src/main/kotlin/io/embrace/android/embracesdk/internal/delivery/intake/IntakeServiceImpl.kt

@@ -32,7 +32,7 @@ class IntakeServiceImpl(
         worker.shutdownAndWait(shutdownTimeoutMs)
     }
 
-    override fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata) {
+    override fun take(intake: Envelope<*>, metadata: StoredTelemetryMetadata): Future<*> {
         deliveryTracer?.onTake(metadata)
         val future = worker.submit(metadata) {
             processIntake(intake, metadata)
@@ -44,6 +44,8 @@ class IntakeServiceImpl(
             cachingTasks[metadata.envelopeType] = future
             prev?.cancel(false)
         }
+
+        return future
     }
 
     @Suppress("UNCHECKED_CAST")