RAD-17289 Add descriptor priority: keep one Android transmit slot for high priority

CHANGELOG.md

@@ -5,6 +5,11 @@ warnings and errors, including ones from background runs with no JS, and a
 snapshot of the queue and its WorkManager runs each time the module starts.
 Before, these went to logcat only. iOS returns an empty list.
 
+Added `priority` to the request descriptor. On Android, normal requests now
+use at most 3 of the 4 transmit slots, and the fourth is kept for requests
+with `priority: 'high'`. A small request then no longer waits behind a
+backlog of file parts. iOS ignores the field for now.
+
 ## 11.0.2
 
 Android no longer lets stale wakes stop the queue. A parked entry got a wake

README.md

@@ -129,6 +129,7 @@ TypeScript does not flag a misspelled key on an inferred arrow return.
 | `expiresAt` | Epoch ms. Default now + `lifetimeMs` (14 days). Past it: `error` with `errorKind: 'expired'`. |
 | `retry` | Per-request override of the `configure()` retry defaults. |
 | `wifiOnly` | `true` waits for Wi-Fi before each attempt; `false` never waits. Overrides `setWifiOnly()` for this entry. Omit it to follow `setWifiOnly()`, including later toggles. |
+| `priority` | `'high'` or `'normal'` (default). On Android at most 4 requests transmit at once, and one of those slots is kept for `'high'`. Normal requests can use only the other 3, so a high-priority request, such as a small JSON POST that creates a session, gets a slot without waiting behind a backlog of file parts. It can still wait when other high-priority requests hold every slot. iOS ignores it for now. |
 | `android` | `{ noNotification?: boolean }`. See Silent uploads. |
 
 `vars` and `data` cross to native as JSON strings, and native parses them.

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

@@ -121,11 +121,11 @@ object EnqueueRules {
   )
 
   /**
-   * Same body. New headers, expiresAt, vars, accept, retry, wifiOnly, and
-   * notification flag replace the stored ones; the body and accepted parts
-   * stay. A settled entry reopens with a fresh generation and attempts 0
-   * (attempts count the current generation). A running one stays running
-   * (the worker reads the new headers before its next attempt).
+   * Same body. New headers, expiresAt, vars, accept, retry, wifiOnly,
+   * priority, and notification flag replace the stored ones; the body and
+   * accepted parts stay. A settled entry reopens with a fresh generation and
+   * attempts 0 (attempts count the current generation). A running one stays
+   * running (the worker reads the new headers before its next attempt).
    */
   fun resumed(
     existing: QueueEntry,
@@ -149,6 +149,7 @@ object EnqueueRules {
         retry = p.descriptor.retry,
         noNotification = p.descriptor.noNotification,
         wifiOnly = p.descriptor.wifiOnly,
+        highPriority = p.descriptor.highPriority,
       ),
       state = if (running) EntryState.RUNNING else initialState(paused),
       attempts = if (reopen) 0 else existing.attempts,

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

@@ -66,6 +66,8 @@ object EntryParsing {
     val headers = parseHeaderMap(d.map("headers"))
     requireValidHeaders(headers, "headers")
     if (d.isSet("wifiOnly") && d.bool("wifiOnly") == null) invalid("wifiOnly must be a boolean")
+    val priority = if (d.isSet("priority")) d.string("priority") else "normal"
+    if (priority != "high" && priority != "normal") invalid("priority must be 'high' or 'normal'")
 
     return Descriptor(
       url = url,
@@ -79,6 +81,7 @@ object EntryParsing {
       retry = d.map("retry")?.let { parseRetry(it) },
       noNotification = d.map("android")?.bool("noNotification") ?: false,
       wifiOnly = d.bool("wifiOnly"),
+      highPriority = priority == "high",
     )
   }
 

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

@@ -261,7 +261,7 @@ internal class EntryRun(
     val policy = policy(entry)
 
     val response = try {
-      host.send(TransferRequest(url, d.method, headersFor(d, part, requestId), body(d, part)), onProgress)
+      host.send(TransferRequest(url, d.method, headersFor(d, part, requestId), body(d, part), d.highPriority), onProgress)
     } catch (error: CancellationException) {
       throw error
     } catch (error: Throwable) {

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

@@ -9,7 +9,6 @@ import androidx.work.WorkerParameters
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.sync.withPermit
 import kotlinx.coroutines.withContext
 
 /**
@@ -79,7 +78,7 @@ open class EntryWorker(protected val context: Context, params: WorkerParameters)
     override suspend fun sleep(ms: Long) = delay(ms)
 
     override suspend fun send(request: TransferRequest, onProgress: (Long) -> Unit): UploadResponse =
-      transferSemaphore.withPermit { okhttpSend(uploadHttpClient, request, onProgress) }
+      transferGate.withPermit(request.highPriority) { okhttpSend(uploadHttpClient, request, onProgress) }
 
     override fun connectivity(wifiOnly: Boolean): Connectivity {
       connectivity = validateConnectivity(context, wifiOnly)

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

@@ -51,6 +51,8 @@ data class Descriptor(
   val noNotification: Boolean,
   /** The request's own Wi-Fi rule. Null follows the queue's setWifiOnly(). */
   val wifiOnly: Boolean? = null,
+  /** descriptor.priority 'high'. An entry saved before the field reads false. */
+  val highPriority: Boolean = false,
 ) {
   val bodyKind: String
     get() = when {

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

@@ -1,6 +1,7 @@
 package ai.openspace.backgroundupload
 
 import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withPermit
 import okhttp3.OkHttpClient
 import java.util.concurrent.TimeUnit
 
@@ -21,7 +22,34 @@ private val REQUEST_TIMEOUT_UNIT = TimeUnit.HOURS
 // Android the cap is hard. A semaphore controls this, not OkHttp's connection
 // limits, because those limits add a delay between requests.
 internal const val MAX_TRANSFER_CONCURRENCY = 4
-internal val transferSemaphore = Semaphore(MAX_TRANSFER_CONCURRENCY)
+
+// The slots of that cap that only a high-priority request may use. A small
+// request, such as a session's JSON, then never waits behind a backlog of
+// file parts.
+internal const val HIGH_PRIORITY_RESERVED = 1
+
+/**
+ * The library-wide cap of [total] requests transmitting at one time, with
+ * [reserved] of them kept for high-priority requests. A normal request takes
+ * a lane permit, then a slot, so normal requests hold at most total -
+ * reserved slots. A high-priority request takes any free slot.
+ */
+internal class TransferGate(total: Int, reserved: Int) {
+  private val slots = Semaphore(total)
+  private val normalLane = Semaphore(total - reserved)
+
+  val availableSlots: Int
+    get() = slots.availablePermits
+
+  suspend fun <T> withPermit(highPriority: Boolean, block: suspend () -> T): T =
+    if (highPriority) {
+      slots.withPermit { block() }
+    } else {
+      normalLane.withPermit { slots.withPermit { block() } }
+    }
+}
+
+internal val transferGate = TransferGate(MAX_TRANSFER_CONCURRENCY, HIGH_PRIORITY_RESERVED)
 
 // Use Okhttp as it provides the most standard behaviors even though it's not coroutine friendly
 internal val uploadHttpClient = OkHttpClient.Builder()

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

@@ -40,6 +40,8 @@ data class TransferRequest(
   val method: String,
   val headers: Map<String, String>,
   val body: RequestBody?,
+  /** Whether the request may use the slots kept for high priority. See [TransferGate]. */
+  val highPriority: Boolean = false,
 )
 
 /** Sends one request and reports bytes written. The headers are sent as they are. */

android/src/test/java/ai/openspace/backgroundupload/EntryParsingTest.kt

@@ -6,6 +6,7 @@ import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
 import org.junit.Assert.assertThrows
+import org.junit.Assert.assertTrue
 import org.junit.Test
 
 class EntryParsingTest {
@@ -126,6 +127,19 @@ class EntryParsingTest {
     }
   }
 
+  @Test
+  fun `priority parses high, normal, or absent`() {
+    assertTrue(EntryParsing.parse(entryMap(base("priority", "high"))).descriptor.highPriority)
+    assertFalse(EntryParsing.parse(entryMap(base("priority", "normal"))).descriptor.highPriority)
+    assertFalse(EntryParsing.parse(entryMap(base())).descriptor.highPriority)
+    assertThrows(EntryParsing.InvalidEntryException::class.java) {
+      EntryParsing.parse(entryMap(base("priority", "urgent")))
+    }
+    assertThrows(EntryParsing.InvalidEntryException::class.java) {
+      EntryParsing.parse(entryMap(base("priority", true)))
+    }
+  }
+
   @Test
   fun `a pause scope with no keys is the whole queue, and keys must be non-empty strings`() {
     assertNull(EntryParsing.scopeKeys(JavaOnlyMap()))

android/src/test/java/ai/openspace/backgroundupload/EntryRunTest.kt

@@ -127,6 +127,14 @@ class EntryRunTest {
     assertEquals(listOf("start:7:0", "7", "end:true"), host.progress)
   }
 
+  @Test
+  fun `a high-priority entry sends its request as high priority`() {
+    controller.enqueue(parsed(descriptor = desc(dataJson = """{"a":1}""", highPriority = true)))
+    respond(200)
+    run()
+    assertTrue(host.requests.single().highPriority)
+  }
+
   @Test
   fun `a transient response waits a short backoff in place, with nextAttemptAt on the running row`() {
     controller.enqueue(parsed())

android/src/test/java/ai/openspace/backgroundupload/QueueControllerTest.kt

@@ -654,6 +654,13 @@ class QueueControllerTest {
     assertNull(reread.load("f")!!.descriptor!!.wifiOnly)
   }
 
+  @Test
+  fun `same body with a new priority resumes with that priority`() {
+    controller.enqueue(parsed(id = "s", descriptor = desc(dataJson = """{"a":1}""")))
+    controller.enqueue(parsed(id = "s", descriptor = desc(dataJson = """{"a":1}""", highPriority = true)))
+    assertTrue(store.load("s")!!.descriptor!!.highPriority)
+  }
+
   @Test
   fun `setWifiOnly persists`() {
     controller.setWifiOnly(true)

android/src/test/java/ai/openspace/backgroundupload/SmallPartsTest.kt

@@ -6,7 +6,12 @@ import androidx.work.WorkInfo.State.ENQUEUED
 import androidx.work.WorkInfo.State.FAILED
 import androidx.work.WorkInfo.State.RUNNING
 import androidx.work.WorkInfo.State.SUCCEEDED
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.joinAll
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
@@ -159,18 +164,43 @@ class SchedulerTest {
   }
 }
 
-class TransferSemaphoreTest {
+class TransferGateTest {
   @Test
-  fun `the global cap is 4 and a fifth request waits`() = runBlocking {
+  fun `the global cap is 4, with 1 slot kept for high priority`() {
     assertEquals(4, MAX_TRANSFER_CONCURRENCY)
-    repeat(4) { transferSemaphore.acquire() }
-    try {
-      assertFalse(transferSemaphore.tryAcquire()) // no fifth permit
-      transferSemaphore.release()
-      assertTrue(transferSemaphore.tryAcquire()) // one freed, one taken
-    } finally {
-      repeat(4) { transferSemaphore.release() }
+    assertEquals(1, HIGH_PRIORITY_RESERVED)
+    assertEquals(4, transferGate.availableSlots)
+  }
+
+  @Test
+  fun `normal requests take at most 3 slots, and a high-priority request takes the 4th`() = runBlocking {
+    val gate = TransferGate(total = 4, reserved = 1)
+    val release = CompletableDeferred<Unit>()
+    val started = Channel<String>(Channel.UNLIMITED)
+    val normals = (1..4).map { n ->
+      launch { gate.withPermit(highPriority = false) { started.send("normal$n"); release.await() } }
     }
-    assertEquals(4, transferSemaphore.availablePermits)
+    repeat(3) { started.receive() }
+    yield()
+    assertNull(started.tryReceive().getOrNull()) // the 4th normal waits
+    assertEquals(1, gate.availableSlots)
+    val high = launch { gate.withPermit(highPriority = true) { started.send("high"); release.await() } }
+    assertEquals("high", started.receive())
+    assertEquals(0, gate.availableSlots)
+    release.complete(Unit)
+    (normals + high).joinAll()
+    assertEquals(4, gate.availableSlots)
+  }
+
+  @Test
+  fun `high-priority requests may use every slot`() = runBlocking {
+    val gate = TransferGate(total = 4, reserved = 1)
+    val release = CompletableDeferred<Unit>()
+    val started = Channel<Unit>(Channel.UNLIMITED)
+    val highs = (1..4).map { launch { gate.withPermit(highPriority = true) { started.send(Unit); release.await() } } }
+    repeat(4) { started.receive() }
+    assertEquals(0, gate.availableSlots)
+    release.complete(Unit)
+    highs.joinAll()
   }
 }

android/src/test/java/ai/openspace/backgroundupload/TestSupport.kt

@@ -16,7 +16,8 @@ internal fun desc(
   retry: RetryOverride? = null,
   noNotification: Boolean = false,
   wifiOnly: Boolean? = null,
-) = Descriptor(url, method, headers, dataJson, form, file, parts, accept, retry, noNotification, wifiOnly)
+  highPriority: Boolean = false,
+) = Descriptor(url, method, headers, dataJson, form, file, parts, accept, retry, noNotification, wifiOnly, highPriority)
 
 internal fun part(start: Long, end: Long, accepted: Boolean = false, url: String? = null) = Part(
   url = url ?: "https://example.com/part?start=$start",

src/__tests__/registry.test.ts

@@ -581,6 +581,23 @@ describe('mutate', () => {
       );
     });
 
+    it('forwards priority to native when set', async () => {
+      const high = mutateWith({ url: 'https://x', priority: 'high' });
+      await high.promise;
+      expect(high.enqueue.mock.calls[0][0].descriptor.priority).toBe('high');
+      const plain = mutateWith({ url: 'https://x' });
+      await plain.promise;
+      expect(plain.enqueue.mock.calls[0][0].descriptor).not.toHaveProperty(
+        'priority',
+      );
+    });
+
+    it('rejects a priority other than high or normal', async () => {
+      await expect(
+        mutateWith({ url: 'https://x', priority: 'urgent' }).promise,
+      ).rejects.toThrow(/priority must be 'high' or 'normal'/);
+    });
+
     it('rejects a wifiOnly that is not a boolean, and a misspelling', async () => {
       await expect(
         mutateWith({ url: 'https://x', wifiOnly: 'yes' }).promise,

src/registry.ts

@@ -40,6 +40,7 @@ const DESCRIPTOR_KEYS = [
   'expiresAt',
   'retry',
   'wifiOnly',
+  'priority',
   'android',
 ];
 const PART_KEYS = ['url', 'headers', 'range'];
@@ -483,6 +484,13 @@ export const validateDescriptor = (descriptor: unknown): NativeDescriptor => {
   if (d.wifiOnly !== undefined && typeof d.wifiOnly !== 'boolean') {
     throw new Error('mutate: wifiOnly must be a boolean when present');
   }
+  if (
+    d.priority !== undefined &&
+    d.priority !== 'high' &&
+    d.priority !== 'normal'
+  ) {
+    throw new Error("mutate: priority must be 'high' or 'normal' when present");
+  }
   if (
     d.expiresAt !== undefined &&
     (!Number.isFinite(d.expiresAt) || d.expiresAt <= 0)

src/types.ts

@@ -106,6 +106,13 @@ export type RequestDescriptor = {
    * entry.
    */
   wifiOnly?: boolean;
+  /**
+   * 'high' lets the request use the transmit slot kept for high priority,
+   * so it does not wait behind a backlog of file parts. Use it for small
+   * requests that other work depends on, such as creating a session.
+   * Default 'normal'. Android only for now; iOS ignores it.
+   */
+  priority?: 'high' | 'normal';
   android?: { noNotification?: boolean };
 };