diff --git a/CHANGELOG.md b/CHANGELOG.md index 12b8e6fd..45ef976a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 968bd12f..2709ebf5 100644 --- a/README.md +++ b/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. diff --git a/android/src/main/java/ai/openspace/backgroundupload/EnqueueRules.kt b/android/src/main/java/ai/openspace/backgroundupload/EnqueueRules.kt index 5fd1ac15..dad38422 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EnqueueRules.kt +++ b/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, diff --git a/android/src/main/java/ai/openspace/backgroundupload/EntryParsing.kt b/android/src/main/java/ai/openspace/backgroundupload/EntryParsing.kt index 5992de45..b7080d71 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EntryParsing.kt +++ b/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", ) } diff --git a/android/src/main/java/ai/openspace/backgroundupload/EntryRun.kt b/android/src/main/java/ai/openspace/backgroundupload/EntryRun.kt index aa130380..79e1428f 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EntryRun.kt +++ b/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) { diff --git a/android/src/main/java/ai/openspace/backgroundupload/EntryWorker.kt b/android/src/main/java/ai/openspace/backgroundupload/EntryWorker.kt index 66e297c5..13b98832 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EntryWorker.kt +++ b/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) diff --git a/android/src/main/java/ai/openspace/backgroundupload/QueueEntry.kt b/android/src/main/java/ai/openspace/backgroundupload/QueueEntry.kt index 09f1b364..f183e5ce 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/QueueEntry.kt +++ b/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 { diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt index cc8093ea..879a2208 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt +++ b/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 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() diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt index 154cac08..e17c37ec 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt @@ -40,6 +40,8 @@ data class TransferRequest( val method: String, val headers: Map, 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. */ diff --git a/android/src/test/java/ai/openspace/backgroundupload/EntryParsingTest.kt b/android/src/test/java/ai/openspace/backgroundupload/EntryParsingTest.kt index 900a94db..39b3a0f9 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/EntryParsingTest.kt +++ b/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())) diff --git a/android/src/test/java/ai/openspace/backgroundupload/EntryRunTest.kt b/android/src/test/java/ai/openspace/backgroundupload/EntryRunTest.kt index 4471e889..7cbadac3 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/EntryRunTest.kt +++ b/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()) diff --git a/android/src/test/java/ai/openspace/backgroundupload/QueueControllerTest.kt b/android/src/test/java/ai/openspace/backgroundupload/QueueControllerTest.kt index e5e4c918..1f70cfa2 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/QueueControllerTest.kt +++ b/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) diff --git a/android/src/test/java/ai/openspace/backgroundupload/SmallPartsTest.kt b/android/src/test/java/ai/openspace/backgroundupload/SmallPartsTest.kt index a5bfda74..b5ebc7d4 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/SmallPartsTest.kt +++ b/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() + val started = Channel(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() + val started = Channel(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() } } diff --git a/android/src/test/java/ai/openspace/backgroundupload/TestSupport.kt b/android/src/test/java/ai/openspace/backgroundupload/TestSupport.kt index bccb309e..bfc98d08 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/TestSupport.kt +++ b/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", diff --git a/src/__tests__/registry.test.ts b/src/__tests__/registry.test.ts index eddd3e43..a5b41aa5 100644 --- a/src/__tests__/registry.test.ts +++ b/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, diff --git a/src/registry.ts b/src/registry.ts index ceb5471d..b35ea1a8 100644 --- a/src/registry.ts +++ b/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) diff --git a/src/types.ts b/src/types.ts index 96f65dba..a8bd7525 100644 --- a/src/types.ts +++ b/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 }; };