v10 slice 2: Android queue store, executor, and events

README.md

@@ -191,6 +191,21 @@ notification. That notification is also the worker's foreground-service
 notification, so a silent request runs as an ordinary background worker and
 the OS may defer or restart it. Reserve it for small payloads.
 
+### Android platform notes
+
+**Headless time limit.** On API 31 and later, a WorkManager run that starts
+from the background usually cannot start its foreground service. The run
+then has JobScheduler's limit of about 10 minutes. A single body (`data`,
+`form`, `file`) that does not finish in that time starts again from byte 0
+at the next run, after a growing backoff. A body that needs more than
+10 minutes headless cannot finish that way. Use `parts` for large bodies:
+accepted parts are kept across runs. iOS has no equal limit.
+
+**Backups.** The queue store (`files/rnbgupload-chunked/`) and the journal
+(`files/rnbgupload-settled/`) hold request headers, including auth tokens,
+and staged bodies. Set `android:allowBackup="false"` in the host app, or
+exclude those two directories in its backup rules.
+
 # Reliable delivery
 
 1. **Write-ahead.** Entry, descriptor, and staged body persist before any

android/consumer-rules.pro

@@ -1,32 +1,43 @@
 # These rules go to consumers through consumerProguardFiles. Thus a minified
 # release build of the host app keeps these guarantees.
 #
-# Gson persists Upload, NotificationConfig, and EventJournal.Entry. Upload goes
-# into WorkManager input data. NotificationConfig goes into SharedPreferences.
-# Entry goes into the on-disk event journal. The library reads them back later,
-# across app restarts AND across app updates. Gson finds fields by name through
-# reflection. Gson also needs the generic Signature attribute to rebuild typed
-# collections. Thus, if R8 renames a field or removes Signature, it corrupts the
-# persisted state silently:
+# Gson persists the queue entry (entry.json), the queue settings
+# (settings.json), the settled-outcome journal, NotificationConfig
+# (SharedPreferences), and reads the v9 journal and v9 chunked manifests at
+# the first v10 launch. The library reads them back later, across app
+# restarts AND across app updates. Gson finds fields by name through
+# reflection, and it needs the generic Signature attribute to rebuild typed
+# collections. Thus, if R8 renames a field or removes Signature, it corrupts
+# the persisted state silently:
 #
-#   * Upload.accept is a List<AcceptRule>. Without Signature, Gson decodes the
-#     elements as bare maps. Then no rule ever matches, and a configured accept
-#     status (for example 409) is reported as an http error, not as a completed
-#     upload. ChunkedManifest.parts has the same shape and the same failure.
-#   * A journal Entry from an older build fails to parse if field names changed.
-#     The library then drops the Entry as malformed. This loses the terminal
-#     outcomes that the journal exists to keep. A ChunkedManifest is the resume
-#     record for a chunked upload, and it fails in the same way.
+#   * Descriptor.accept is a List<AcceptRule> and Descriptor.parts a
+#     List<Part>. Without Signature, Gson decodes the elements as bare maps.
+#     Then no accept rule matches, and every chunked part reads as unsent.
+#   * A record from an older build fails to parse if field names changed. The
+#     library drops it as malformed. That loses the outcomes the journal
+#     exists to keep, and the entries the queue exists to run.
+#   * EntryState is an enum persisted by its @SerializedName wire string.
 #
-# Debug builds are not minified and round-trip correctly. Thus neither failure
+# Debug builds are not minified and round-trip correctly, so neither failure
 # is reproducible without R8. Keep these rules.
 -keepattributes Signature
 -keepattributes *Annotation*
 
--keep class ai.openspace.backgroundupload.Upload { *; }
--keep class ai.openspace.backgroundupload.Upload$* { *; }
--keep class ai.openspace.backgroundupload.NotificationConfig { *; }
--keep class ai.openspace.backgroundupload.EventJournal$Entry { *; }
--keep class ai.openspace.backgroundupload.ChunkedManifest { *; }
--keep class ai.openspace.backgroundupload.ChunkedManifest$* { *; }
+# v10 queue
+-keep class ai.openspace.backgroundupload.QueueEntry { *; }
+-keep class ai.openspace.backgroundupload.EntryState { *; }
+-keep class ai.openspace.backgroundupload.Descriptor { *; }
+-keep class ai.openspace.backgroundupload.FormPart { *; }
+-keep class ai.openspace.backgroundupload.RetryOverride { *; }
+-keep class ai.openspace.backgroundupload.StagedBody { *; }
+-keep class ai.openspace.backgroundupload.Part { *; }
+-keep class ai.openspace.backgroundupload.QueueSettings { *; }
+-keep class ai.openspace.backgroundupload.RetryDefaults { *; }
+-keep class ai.openspace.backgroundupload.EventJournal$SettledRecord { *; }
+-keep class ai.openspace.backgroundupload.EventJournal$Response { *; }
 -keep class ai.openspace.backgroundupload.UploadOutcome$AcceptRule { *; }
+-keep class ai.openspace.backgroundupload.NotificationConfig { *; }
+
+# v9 files read once at the first v10 launch
+-keep class ai.openspace.backgroundupload.LegacyImport$V9Entry { *; }
+-keep class ai.openspace.backgroundupload.LegacyManifest { *; }

android/src/main/java/ai/openspace/backgroundupload/AtomicFiles.kt

@@ -0,0 +1,53 @@
+package ai.openspace.backgroundupload
+
+import java.io.File
+import java.io.FileOutputStream
+import java.io.IOException
+import java.nio.channels.FileChannel
+import java.nio.file.StandardOpenOption
+
+/**
+ * Write-ahead file writes. Every durable file in the library goes through
+ * [writeAtomically]: write a tmp sibling, fsync it, rename it over the
+ * target. A crash at any point leaves either the old target or the new one,
+ * never a partial file.
+ */
+internal object AtomicFiles {
+  const val TMP_SUFFIX = ".tmp"
+
+  fun tmpFor(target: File) = File(target.parentFile, target.name + TMP_SUFFIX)
+
+  /** Throws IOException when the target could not be replaced. The old target is then intact. */
+  fun writeAtomically(target: File, write: (FileOutputStream) -> Unit) {
+    val parent = target.parentFile ?: throw IOException("no parent directory for ${target.path}")
+    if (!parent.isDirectory && !parent.mkdirs()) {
+      throw IOException("could not create ${parent.path}")
+    }
+    val tmp = tmpFor(target)
+    try {
+      FileOutputStream(tmp).use { out ->
+        write(out)
+        out.flush()
+        out.fd.sync()
+      }
+      if (!tmp.renameTo(target)) throw IOException("could not rename ${tmp.path} to ${target.name}")
+    } catch (error: Throwable) {
+      tmp.delete()
+      throw error
+    }
+    syncDirectory(parent)
+  }
+
+  fun writeText(target: File, text: String) =
+    writeAtomically(target) { it.write(text.toByteArray(Charsets.UTF_8)) }
+
+  /**
+   * Makes a rename durable. This works on Linux (Android). Some file systems
+   * do not allow it, so a failure is ignored: the rename itself is still atomic.
+   */
+  fun syncDirectory(dir: File) {
+    runCatching {
+      FileChannel.open(dir.toPath(), StandardOpenOption.READ).use { it.force(true) }
+    }
+  }
+}

android/src/main/java/ai/openspace/backgroundupload/AttemptEvent.kt

@@ -0,0 +1,88 @@
+package ai.openspace.backgroundupload
+
+import com.facebook.react.bridge.WritableMap
+
+/**
+ * One HTTP attempt, before the library interprets it. Live only: never
+ * journaled. [outcome] is `completed` when the response is accepted and
+ * `error` otherwise, so a 401 is `error` with httpCode 401 even though the
+ * entry parks. A transport failure is `error` with its own errorKind. Pause,
+ * cancel, supersede, and a system stop emit no attempt event.
+ */
+data class AttemptEvent(
+  val id: String,
+  val key: String,
+  val requestId: String,
+  val attempt: Int,
+  val url: String,
+  val method: String,
+  val partIndex: Int?,
+  val outcome: String,
+  val httpCode: Int?,
+  val responseBody: String?,
+  val responseBodyTruncated: Boolean?,
+  val responseHeaders: Map<String, String>?,
+  val errorKind: String?,
+  val errorMessage: String?,
+  val at: Long,
+) {
+  companion object {
+    fun ofResponse(
+      entry: QueueEntry,
+      requestId: String,
+      url: String,
+      partIndex: Int?,
+      response: UploadResponse,
+      accepted: Boolean,
+      at: Long,
+    ): AttemptEvent {
+      val (body, cut) = BodyCap.cap(response.body, BodyCap.ATTEMPT_MAX_BYTES)
+      return AttemptEvent(
+        id = entry.id, key = entry.key, requestId = requestId, attempt = entry.attempts,
+        url = url, method = entry.descriptor?.method ?: "POST", partIndex = partIndex,
+        outcome = if (accepted) "completed" else "error",
+        httpCode = response.code, responseBody = body, responseBodyTruncated = cut || response.truncated,
+        responseHeaders = response.headers,
+        errorKind = if (accepted) null else "http",
+        errorMessage = if (accepted) null else "HTTP ${response.code}",
+        at = at,
+      )
+    }
+
+    fun ofFailure(
+      entry: QueueEntry,
+      requestId: String,
+      url: String,
+      partIndex: Int?,
+      errorKind: String,
+      message: String,
+      at: Long,
+    ) = AttemptEvent(
+      id = entry.id, key = entry.key, requestId = requestId, attempt = entry.attempts,
+      url = url, method = entry.descriptor?.method ?: "POST", partIndex = partIndex,
+      outcome = "error", httpCode = null, responseBody = null, responseBodyTruncated = null,
+      responseHeaders = null, errorKind = errorKind, errorMessage = message,
+      at = at,
+    )
+  }
+
+  fun toMap(): Map<String, Any?> = LinkedHashMap<String, Any?>().apply {
+    put("id", id)
+    put("key", key)
+    put("requestId", requestId)
+    put("attempt", attempt.toDouble())
+    put("url", url)
+    put("method", method)
+    partIndex?.let { put("partIndex", it.toDouble()) }
+    put("outcome", outcome)
+    httpCode?.let { put("httpCode", it.toDouble()) }
+    responseBody?.let { put("responseBody", it) }
+    responseBodyTruncated?.let { put("responseBodyTruncated", it) }
+    responseHeaders?.let { put("responseHeaders", it) }
+    errorKind?.let { put("errorKind", it) }
+    errorMessage?.let { put("errorMessage", it) }
+    put("at", at.toDouble())
+  }
+
+  fun toWritableMap(): WritableMap = JsonBridge.toWritableMap(toMap())
+}

android/src/main/java/ai/openspace/backgroundupload/BodyCap.kt

@@ -0,0 +1,67 @@
+package ai.openspace.backgroundupload
+
+import okio.Buffer
+import okio.BufferedSource
+import java.nio.charset.Charset
+
+/**
+ * Response body caps, in UTF-8 bytes. A cut never splits a character: it
+ * backs off to the last whole one. [read] applies the cap while the body
+ * streams in, so a huge error page never sits in memory whole.
+ */
+object BodyCap {
+  /** 1 MB, the RawResponse cap of a settled outcome. */
+  const val SETTLED_MAX_BYTES = 1_048_576
+
+  /** 4 KB, the body cap of a live attempt event. */
+  const val ATTEMPT_MAX_BYTES = 4 * 1024
+
+  /** A body as text and whether the cap cut it. */
+  data class Capped(val text: String, val truncated: Boolean)
+
+  /**
+   * Reads at most [maxBytes] of [source]. Bytes past the cap are not read.
+   * A UTF-8 body is cut on a character boundary; another charset is cut at
+   * the byte cap and decoded as it is.
+   */
+  fun read(source: BufferedSource, maxBytes: Int, charset: Charset = Charsets.UTF_8): Capped {
+    val buffer = Buffer()
+    val limit = maxBytes.toLong() + 1
+    while (buffer.size < limit) {
+      if (source.read(buffer, limit - buffer.size) == -1L) break
+    }
+    val truncated = buffer.size > maxBytes
+    val bytes = buffer.readByteArray()
+    val keep = if (!truncated) bytes.size
+    else if (charset == Charsets.UTF_8) utf8Boundary(bytes, maxBytes)
+    else maxBytes
+    return Capped(String(bytes, 0, keep, charset), truncated)
+  }
+
+  /** [text] cut to at most [maxBytes] of UTF-8. Null stays null. */
+  fun cap(text: String?, maxBytes: Int): Pair<String?, Boolean> {
+    if (text == null) return null to false
+    val bytes = text.toByteArray(Charsets.UTF_8)
+    if (bytes.size <= maxBytes) return text to false
+    return String(bytes, 0, utf8Boundary(bytes, maxBytes), Charsets.UTF_8) to true
+  }
+
+  /**
+   * The longest prefix length of [bytes], at most [max], that does not end
+   * inside a UTF-8 sequence. A continuation byte is 10xxxxxx.
+   */
+  internal fun utf8Boundary(bytes: ByteArray, max: Int): Int {
+    if (bytes.size <= max) return bytes.size
+    var end = max
+    // bytes[end] is the first byte cut off. While it continues a sequence,
+    // the sequence started before the cut, so drop its start too. A UTF-8
+    // character is at most 4 bytes; past 3 steps the bytes are not UTF-8,
+    // and the cut stays at max.
+    var steps = 0
+    while (end > 0 && steps < 3 && (bytes[end].toInt() and 0xC0) == 0x80) {
+      end--
+      steps++
+    }
+    return if ((bytes[end].toInt() and 0xC0) == 0x80) max else end
+  }
+}