NAT-549: Handle standardized-payload resume safely

library/src/main/java/com/mux/video/upload/api/MuxUpload.kt

@@ -122,7 +122,7 @@ class MuxUpload private constructor(
    *
    * To suspend the execution of the upload, use [pause]. To cancel it completely, use [cancel]
    *
-   * @param forceRestart Start the upload from the beginning even if the file is partially uploaded
+   * @param forceRestart Replace the current attempt while retaining its selected payload and acknowledged bytes
    *
    * @see pause
    * @see cancel
@@ -152,14 +152,13 @@ class MuxUpload private constructor(
     } else {
       // If we're not managing the worker, the job is purely internal to this object
       if (forceRestart) {
-        val blocked = uploadInfo.generatedResumeBlocked || uploadInfo.attempt?.preparation?.generatedRequestStarted == true ||
-          readUploadResumeState(uploadInfo).generatedResumeBlocked
+        val blocked = uploadInfo.standardizedResumeBlocked || readUploadResumeState(uploadInfo).standardizedResumeBlocked
         val previous = uploadInfo
-        if (previous.attempt == null) forgetUploadState(previous)
-        else previous.attempt.cancel(forRestart = true) { forgetUploadState(previous) }
+        previous.attempt?.cancel(forRestart = true) { forgetUploadState(previous) }
         previous.uploadJob?.cancel()
         uploadInfo = uploadInfo.update(attempt = null, statusFlow = null,
-          restoredFromOriginal = false, generatedResumeBlocked = blocked)
+          restoredFromOriginal = previous.attempt?.preparation?.originalSelected == true || previous.restoredFromOriginal,
+          standardizedResumeBlocked = blocked)
       }
       /*uploadInfo =*/ MuxUploadSdk.uploadJobFactory().createUploadJob(uploadInfo, coroutineScope)
     }
@@ -252,7 +251,7 @@ class MuxUpload private constructor(
   /**
    * Cancels this upload and ends this handle. Later [start] calls have no effect and
    * [awaitSuccess] returns a failure. Create a new handle to start another upload.
-   * A destination that received generated bytes requires a new Direct Upload URL.
+   * A destination that received standardized bytes requires a new Direct Upload URL.
    */
   @Suppress("MemberVisibilityCanBePrivate")
   fun cancel() {

library/src/main/java/com/mux/video/upload/api/MuxUploadManager.kt

@@ -63,9 +63,7 @@ object MuxUploadManager {
   @JvmSynthetic
   @MainThread
   internal fun resumeCachedJobs(includePaused: Boolean): List<MuxUpload> {
-    val restored = readCachedUploadSnapshots().filter { cached ->
-      cached.upload.inputFile.exists().also { if (!it) forgetUploadState(cached.upload) }
-    }.map { (upload, saved) ->
+    val restored = readCachedUploadSnapshots().map { (upload, saved) ->
       if (!includePaused && saved.paused) {
         uploadsByFilename.getOrPut(upload.inputFile.absolutePath) {
           val attempt = createUploadAttempt(upload, saved)
@@ -169,10 +167,8 @@ object MuxUploadManager {
     upload.uploadJob?.cancel()
   }
 
-  private fun cancelAttempt(upload: UploadInfo, forgetUnstarted: Boolean = false, forRestart: Boolean = false) {
-    val attempt = upload.attempt
-    if (attempt != null) attempt.cancel(forRestart = forRestart) { forgetUploadState(upload) }
-    else if (forgetUnstarted) forgetUploadState(upload)
+  private fun cancelAttempt(upload: UploadInfo, forRestart: Boolean = false) {
+    upload.attempt?.cancel(forRestart = forRestart) { forgetUploadState(upload) }
     upload.uploadJob?.cancel()
   }
 
@@ -184,14 +180,15 @@ object MuxUploadManager {
       return previous
     var source = previous ?: upload
     if (restart || newDestination) {
-      val generatedMayExistRemotely = !newDestination && (source.generatedResumeBlocked ||
-        source.attempt?.preparation?.generatedRequestStarted == true || readUploadResumeState(upload).generatedResumeBlocked)
+      val standardizedMayExistRemotely = !newDestination && (source.standardizedResumeBlocked || readUploadResumeState(upload).standardizedResumeBlocked)
       if (newDestination) {
         source.attempt?.replace { forgetUploadState(source) }
         source.uploadJob?.cancel()
-      } else cancelAttempt(source, forgetUnstarted = true, forRestart = true)
+      } else cancelAttempt(source, forRestart = true)
       source = upload.update(attempt = null, uploadJob = source.uploadJob,
-        statusFlow = null, restoredFromOriginal = false, generatedResumeBlocked = generatedMayExistRemotely).also {
+        statusFlow = null, restoredFromOriginal = !newDestination &&
+          (source.attempt?.preparation?.originalSelected == true || source.restoredFromOriginal),
+        standardizedResumeBlocked = standardizedMayExistRemotely).also {
         it.session = if (newDestination) upload.session else source.session
       }
     }

library/src/main/java/com/mux/video/upload/api/UploadEventListener.kt

@@ -5,7 +5,7 @@ package com.mux.video.upload.api
  */
 fun interface UploadEventListener<EventType> {
   /**
-   * Called when an event is generated
+   * Called when an event is standardized
    */
   fun onEvent(event: EventType)
 }

library/src/main/java/com/mux/video/upload/internal/ChunkWorker.kt

@@ -1,17 +1,12 @@
 package com.mux.video.upload.internal
 
-import com.mux.video.upload.MuxUploadSdk
 import com.mux.video.upload.api.MuxUpload
 import com.mux.video.upload.internal.network.asCountingRequestBody
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.MutableSharedFlow
 import okhttp3.MediaType.Companion.toMediaTypeOrNull
 import okhttp3.Request
-import okhttp3.Call
-import okhttp3.Callback
-import kotlin.coroutines.resumeWithException
 import okhttp3.Response
-import java.io.IOException
 
 /**
  * Uploads one single chunk, reporting progress as it goes, and returning the final state when the
@@ -25,10 +20,7 @@ internal class ChunkWorker private constructor(
   private val progressFlow: MutableSharedFlow<MuxUpload.Progress>,
 ) {
   companion object {
-    // Progress updates are only sent once in this time frame. The latest event is always sent
     const val EVENT_DEBOUNCE_DELAY_MS: Long = 200
-    val ACCEPTABLE_STATUS_CODES = listOf(200, 201, 202, 204, 308)
-    val RETRYABLE_STATUS_CODES = listOf(408, 502, 503, 504)
 
     @JvmSynthetic
     internal fun create(
@@ -39,38 +31,18 @@ internal class ChunkWorker private constructor(
     ): ChunkWorker = ChunkWorker(chunk, uploadInfo, videoMimeType, progressFlow)
   }
 
-  private val logger get() = MuxUploadSdk.logger
-
-  private var mostRecentUploadState: RecentState? = null
+  private var mostRecentUploadBytes = 0L
   private var updateCallersJob: Job? = null
 
   @Throws
   suspend fun upload(): MuxUpload.Progress {
-    val moreRetries = { triesSoFar: Int -> triesSoFar < uploadInfo.retriesPerChunk }
-    suspend fun tryUpload(triesSoFar: Int): Result<MuxUpload.Progress> {
-      try {
-        currentCoroutineContext().ensureActive()
-        val (finalState, httpResponse) = doUpload()
-        httpResponse.use {
-          if (ACCEPTABLE_STATUS_CODES.contains(it.code)) return Result.success(finalState)
-          val failure = IOException("Upload request failed: ${it.code}/${it.message}")
-          if (it.code !in RETRYABLE_STATUS_CODES || !moreRetries(triesSoFar)) return Result.failure(failure)
-        }
-        return tryUpload(triesSoFar + 1)
-      } catch (e: CancellationException) {
-        throw e
-      } catch (e: Exception) {
-        return if (moreRetries(triesSoFar)) {
-          // Still have more retries so try again
-          tryUpload(triesSoFar + 1)
-        } else {
-          Result.failure(e)
-        }
-      }
-    }
-
     currentCoroutineContext().ensureActive()
-    return tryUpload(0).getOrThrow()
+    val (finalState, response) = doUpload()
+    return response.use {
+      val acknowledged = finalState.copy(bytesUploaded = UploadProtocol.acknowledge(it, chunk))
+      progressFlow.emit(acknowledged)
+      acknowledged
+    }
   }
 
   @Throws
@@ -80,33 +52,23 @@ internal class ChunkWorker private constructor(
     return supervisorScope {
       val stream = chunk.sliceData
       val chunkSize = chunk.endByte - chunk.startByte + 1
-      val httpClient = MuxUploadSdk.httpClient()
+      val httpClient = UploadProtocol.httpClient()
 
       val putBody =
-        stream.asCountingRequestBody(videoMimeType.toMediaTypeOrNull(), chunkSize) { bytes ->
-          val elapsedRealtime = System.currentTimeMillis()
-          // This process happens really fast, so we debounce the callbacks using a coroutine.
-          // If there's no job to update callers, create one. That job delays for a set duration
-          // then sends a message out on the progress flow with the most-recent known progress
-          synchronized(this) { // Synchronize checking/creating update jobs
-            mostRecentUploadState = RecentState(elapsedRealtime, bytes)
-            if (updateCallersJob == null) {
-              updateCallersJob = async {
-                // Update callers at most once every EVENT_DEBOUNCE_DELAY
-                delay(EVENT_DEBOUNCE_DELAY_MS)
-                val currentState = MuxUpload.Progress(
-                  bytesUploaded = mostRecentUploadState?.uploadBytes ?: 0,
-                  totalBytes = chunkSize,
-                  startTime = startTime,
-                  updatedTime = elapsedRealtime,
-                )
-                progressFlow.emit(currentState)
-                // synchronize again since we're on a different worker inside async { }
-                synchronized(this) { updateCallersJob = null }
-              } // ..async { ...
-            } // if (updateCallersJob == null)
-          } // synchronized(this)
-        } // stream.asCountingRequestBody
+        stream.asCountingRequestBody(videoMimeType.toMediaTypeOrNull(), chunkSize, oneShot = true) { bytes ->
+          synchronized(this@ChunkWorker) {
+            mostRecentUploadBytes = bytes
+            if (updateCallersJob == null) updateCallersJob = launch {
+              delay(EVENT_DEBOUNCE_DELAY_MS)
+              val value = synchronized(this@ChunkWorker) {
+                updateCallersJob = null
+                mostRecentUploadBytes
+              }
+              progressFlow.emit(MuxUpload.Progress(bytesUploaded = value, totalBytes = chunkSize,
+                startTime = startTime, updatedTime = System.currentTimeMillis()))
+            }
+          }
+        }
 
       val request = Request.Builder()
         .url(uploadInfo.remoteUri.toString())
@@ -118,38 +80,23 @@ internal class ChunkWorker private constructor(
         )
         .build()
 
-      logger.v("MuxUpload", "Uploading with request $request")
-      val call = httpClient.newCall(request)
-      val httpResponse = suspendCancellableCoroutine<Response> { continuation ->
-        continuation.invokeOnCancellation { call.cancel() }
-        call.enqueue(object : Callback {
-          override fun onFailure(call: Call, e: IOException) { continuation.resumeWithException(e) }
-          override fun onResponse(call: Call, response: Response) {
-            continuation.resume(response) { _, value, _ -> value.close() }
-          }
-        })
-      }
+      val httpResponse = try { executeUploadRequest(request, httpClient) }
+        finally { updateCallersJob?.cancel() }
       try {
-        logger.v("MuxUpload", "Chunk Response: $httpResponse")
         val finalState = MuxUpload.Progress(
           bytesUploaded = chunkSize,
           totalBytes = chunkSize,
           startTime = startTime,
           updatedTime = System.currentTimeMillis()
         )
-        // Cancel progress updates and make sure no one is stuck listening for more
-        updateCallersJob?.cancel()
-        progressFlow.emit(finalState)
         Pair(finalState, httpResponse)
-        } catch (e: Exception) {
+      } catch (e: Exception) {
         httpResponse.close()
         throw e
       }
     } // supervisorScope
   } // suspend fun doUpload
 
-  private data class RecentState(val updatedTime: Long, val uploadBytes: Long)
-
   internal data class Chunk(
     val startByte: Long,
     val endByte: Long,

library/src/main/java/com/mux/video/upload/internal/PayloadIdentity.kt

@@ -0,0 +1,37 @@
+package com.mux.video.upload.internal
+
+import org.json.JSONObject
+import java.io.File
+import java.io.RandomAccessFile
+import java.security.MessageDigest
+
+/** Bounded first/last 64 KiB fingerprint, with path and stat checks around the read. */
+internal data class PayloadIdentity(val path: String, val size: Long, val modified: Long, val edgeHash: String) {
+  fun matches(file: File): Boolean = this == capture(file)
+  fun toJson() = JSONObject().put("path", path).put("size", size).put("mtime", modified).put("edge_hash", edgeHash)
+
+  companion object {
+    fun fromJson(json: JSONObject) = PayloadIdentity(json.getString("path"), json.getLong("size"),
+      json.getLong("mtime"), json.getString("edge_hash")).also {
+      require(it.size > 0 && it.modified >= 0 && it.edgeHash.matches(Regex("[0-9a-f]{64}")))
+    }
+    fun capture(file: File): PayloadIdentity? = try {
+      val path = file.canonicalPath
+      val size = file.length()
+      val modified = file.lastModified()
+      if (!file.isFile || !file.canRead() || size <= 0) null else {
+        val digest = MessageDigest.getInstance("SHA-256")
+        RandomAccessFile(file, "r").use { input ->
+          val first = ByteArray(minOf(size, 64 * 1024L).toInt())
+          input.readFully(first); digest.update(first)
+          if (size > first.size) {
+            val last = ByteArray(minOf(size - first.size, 64 * 1024L).toInt())
+            input.seek(size - last.size); input.readFully(last); digest.update(last)
+          }
+        }
+        if (file.canonicalPath != path || file.length() != size || file.lastModified() != modified) null
+        else PayloadIdentity(path, size, modified, digest.digest().joinToString("") { "%02x".format(it) })
+      }
+    } catch (_: Exception) { null }
+  }
+}

library/src/main/java/com/mux/video/upload/internal/StandardizedResumeState.kt

@@ -0,0 +1,31 @@
+package com.mux.video.upload.internal
+
+import org.json.JSONObject
+
+internal enum class PreparationPhase { Preparing, Validated, Uploading }
+
+/** Tracks the selected standardized file and whether transport has fixed its bytes. */
+internal data class StandardizedResumeState(
+  val source: PayloadIdentity,
+  val phase: PreparationPhase = PreparationPhase.Preparing,
+  val ownedPath: String? = null,
+  val payload: PayloadIdentity? = null,
+  val networkStarted: Boolean = false,
+  val abandoned: Boolean = false,
+) {
+  fun toJson() = JSONObject().put("source", source.toJson()).put("phase", phase.name)
+    .put("owned_path", ownedPath).put("payload", payload?.toJson()).put("network_started", networkStarted).put("abandoned", abandoned)
+  companion object {
+    fun fromJson(json: JSONObject) = StandardizedResumeState(
+      source = PayloadIdentity.fromJson(json.getJSONObject("source")),
+      phase = PreparationPhase.valueOf(json.getString("phase")),
+      ownedPath = json.optString("owned_path").takeIf { it.isNotEmpty() },
+      payload = json.optJSONObject("payload")?.let(PayloadIdentity::fromJson),
+      networkStarted = json.getBoolean("network_started"),
+      abandoned = json.optBoolean("abandoned", false),
+    ).also {
+      require(it.payload == null || it.payload.path == it.ownedPath)
+      require(it.phase == PreparationPhase.Preparing || it.payload != null)
+    }
+  }
+}