diff --git a/CHANGELOG.md b/CHANGELOG.md index 2aaf6abc..a22f35b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,64 @@ +## 9.0.0 + +Chunked uploads move into the library: one file, many part requests, one upload +id and event stream. The consumer authors the parts (URL, headers, byte range) +once; the library owns transport, the bytes, and resume. See the README's +"Chunked uploads" section. + +Breaking: +- **Listeners are global-only.** `addListener(event, uploadId, callback)` is + gone; use `addListener(event, callback)` and discriminate on `data.id`. +- **`acceptStatus: number[]` is replaced by `accept` rules** on all uploads: + `accept: [{ status, bodyIncludes? }]`. `bodyIncludes` narrows by + response-body substring, for statuses that carry several meanings. +- **`customUploadId` is renamed to `id`** on all upload types. +- **`android.maxRetries` is removed.** Retry policy belongs to the library; + chunked uploads are bounded by `expiresAt` instead. +- **`ios.getUploadStatus` is removed.** Its only use was pre-dispatch dedupe, + which idempotent `startUpload` (below) makes unnecessary. +- **Per-upload notification text is removed.** Set it once with `configure()`; + it is persisted natively so a worker relaunched by WorkManager with no JS + running shows the same text. Per-upload `android` options reduce to + `noNotification`. + +Added: +- **Chunked uploads**: `startUpload({ type: 'chunked', id, path, + parts, accept?, expiresAt, wifiOnly? })`. The library takes ownership of the + file at `startUpload` and deletes it only after a `completed` event is + acknowledged; every other terminal outcome keeps the manifest and bytes for + resume or recreate. +- **`startUpload` is idempotent for every upload, always.** Re-calling with a + running id is never an error; for chunked uploads it reconciles — accepted + parts are skipped, the rest continue with the new call's headers (how a fresh + auth token reaches parts stalled on 401). On iOS the guarantee is race-free: + concurrent same-id calls are serialized natively, so they can never enqueue a + duplicate task. +- **Recreate under the same id**: calling `startUpload` with an existing id and + a *different* parts array replaces the parts over the owned bytes — every + part resets to unsent, and the new ranges must tile the same total size. + Accepted only while the upload is not running (stalled on a terminal error, + expired, or cancelled); rejected while it runs. This is how a consumer + re-uploads under a fresh server uploadId after the old one dies. +- **`expiresAt` / `errorKind: 'expired'`**: a chunked upload past its required + deadline journals a terminal `error` and stops, keeping the bytes. Within the + deadline, transient failures retry on backoff with no attempt cap. +- **`removeUpload(uploadId)`** releases a kept manifest and bytes. +- **`configure(options)`** one-time setup (Android notification text/identity). +- **`chunkPlan(sizeBytes, { min?, max? })`**: the deterministic range splitter, + exposed so the part count told to the server and the parts array derive from + one result. +- Concurrency: on Android a hard cap of 4 concurrent requests across all + uploads — every request passes the shared transfer semaphore (previously + fully serial), with a chunked upload's parts windowed inside it. On iOS a + per-session connection-level backstop: `httpMaximumConnectionsPerHost = 4` + (previously 1), with chunked parts bounded by the per-upload window of 3. + +Fixed: +- Android jobs enqueued by a v8 build replay safely after upgrading. WorkManager + can hand a v9 worker a job serialized by v8 (`acceptStatus`, no `accept`); + the worker now normalizes the legacy shape instead of crashing after the file + has fully transmitted and re-sending it on every retry. + ## 8.1.0 Added `android.noNotification`, which uploads a file without posting a progress diff --git a/README.md b/README.md index 24d87446..2947d31e 100644 --- a/README.md +++ b/README.md @@ -54,28 +54,102 @@ generated header is Objective-C++ only — so the handler lives on `RNBackground ```js import Upload from 'react-native-background-upload'; -const options = { +// Optional. Call one time at app startup to set the Android notification text. +// The library keeps the text in native storage. Thus a worker relaunched with +// no JS shows the same text. If you do not call configure(), the library uses +// default text and makes its own channel. The call does nothing on iOS. +Upload.configure({ android: { notificationTitle: 'Uploading…' } }); + +// Listeners are global. Every event carries the upload's id. +Upload.addListener('progress', ({ id, progress }) => {}); +// responseCode/responseBody are set for simple uploads only. A chunked +// 'completed' carries neither, because no single response represents N parts. +Upload.addListener('completed', ({ id, responseCode, responseBody }) => {}); +Upload.addListener('error', ({ id, error, errorKind, responseCode }) => {}); +Upload.addListener('cancelled', ({ id, cancelReason }) => {}); + +const uploadId = await Upload.startUpload({ + type: 'raw', url: 'https://myservice.com/path/to/post', path: 'file://path/to/file/on/device', method: 'POST', - type: 'raw', headers: { 'content-type': 'application/octet-stream' }, - // Optional. Treat these non-2xx statuses as success (e.g. an idempotent - // create that conflicts). Any other non-2xx is an 'error' with errorKind 'http'. - acceptStatus: [409], - // Optional on Android — the library supplies notification defaults and creates - // its own channel. Override any of these to customize. - android: { notificationTitle: 'Uploading…' }, -}; - -const uploadId = await Upload.startUpload(options); - -Upload.addListener('progress', uploadId, ({ progress }) => {}); -Upload.addListener('completed', uploadId, ({ responseCode, responseBody }) => {}); -Upload.addListener('error', uploadId, ({ error, errorKind, responseCode }) => {}); -Upload.addListener('cancelled', uploadId, ({ cancelReason }) => {}); + // Optional. Non-2xx responses to treat as success (for example, an + // idempotent create that conflicts). Each other non-2xx response is an + // 'error' with errorKind 'http'. + accept: [{ status: 409, bodyIncludes: 'already completed' }], +}); +``` + +## Chunked uploads + +A `type: 'chunked'` upload sends one file as many part requests but stays one +logical upload: one id, one event stream, byte-weighted `progress`, and +`completed` only when every part has been accepted. You author the parts — URL, +headers, byte range — once, at creation; the library owns the transport and +never constructs or edits a protocol field. Author the ranges with `chunkPlan` +so the part count you tell your server and the parts the library sends derive +from the same array: + +```js +const size = (await stat(path)).size; +const ranges = Upload.chunkPlan(size, { min: 8 * 2 ** 20, max: 20 * 2 ** 20 }); +// Tell your server ranges.length parts, then: +await Upload.startUpload({ + type: 'chunked', + id: myDurableId, // required + path, // see file ownership below + parts: ranges.map((range, i) => ({ + url: partUrl(i + 1), + headers: { + Authorization: token, + 'Content-Type': 'application/octet-stream', + 'Content-Range': `bytes ${range.start}-${range.end - 1}/${size}`, + }, + range, // bytes, end exclusive + })), + accept: [{ status: 409, bodyIncludes: 'already completed' }], + expiresAt: Date.now() + 14 * 24 * 60 * 60 * 1000, // required, epoch ms +}); ``` +**File ownership.** The library takes the file: an O(1) rename into its own +directory at `startUpload`. Nothing your app does afterward (cache sweeps, +logout cleanup) can destroy the bytes mid-upload. The file is deleted in +exactly one case — a `completed` event has been acknowledged via `ackEvents`. +Copy the file first if you need it afterward. + +**Resume is re-calling `startUpload`.** The parts are persisted in a native +manifest, so crash recovery, resume after `cancelUpload`, resume after expiry, +and refreshing auth headers are all the same call: `startUpload` again with the +same id and the same part ranges/URLs. Parts already accepted are skipped; the +rest continue with the new call's headers and `expiresAt` (this is how a fresh +token reaches parts that stalled on 401). Once a manifest exists, `path` is +ignored — the library's owned bytes are the source of truth. + +**Recreate is the same call with different parts.** When the old server upload +is dead (for example, swept server-side), author fresh part URLs and call +`startUpload` with the same id and the new parts array. The owned bytes are +kept, the parts are replaced, and every part resets to unsent; the new ranges +must tile the same total size. A recreate is accepted only while the upload is +not running — stalled on a terminal error, expired, or cancelled. While it is +running, a differing parts array is rejected: that is a consumer bug, not a +recreate. + +**Lifetime.** Within `expiresAt`, transient failures (network, 5xx) retry on +exponential backoff with no attempt cap. Past it, the library journals an +`error` with `errorKind: 'expired'` and stops — keeping the manifest and bytes, +so you can resume the same server upload with a later `expiresAt`, or recreate +under a new one. When neither is wanted, release them with `removeUpload`. + +Choosing a value: `expiresAt` is when your app *hears about* a stuck upload, +not when data is lost — bytes survive expiry. Pick something well inside your +backend's own cleanup horizon so expiry fires while the server upload is still +resumable, and generous enough for real offline stretches. The OpenSpace +backend prunes incomplete multipart uploads 31 days after creation +(`UploadPartCleanup`); Diana passes 14 days, leaving a 17-day window where an +expired upload can still resume the same server uploadId. + # Reliable delivery Terminal events (`completed` / `error` / `cancelled`) are journaled natively @@ -97,11 +171,13 @@ const live = await Upload.getAllUploads(); // [{ id, state, ... }] ``` Notes: -- **`completed` fires only for 2xx** (or a request's `acceptStatus`). Every other - HTTP response is an `error` with `errorKind: 'http'` and the response attached — - a 400 is an error, not a completion. -- `errorKind` is `'http' | 'network' | 'file' | 'unknown'`. Retry transport - failures; treat client errors as terminal. +- **`completed` fires only for 2xx** (or a response matching the request's + `accept` rules). Every other HTTP response is an `error` with + `errorKind: 'http'` and the response attached — a 400 is an error, not a + completion. +- `errorKind` is `'http' | 'network' | 'file' | 'expired' | 'unknown'`. Retry + transport failures; treat client errors as terminal; `expired` means a chunked + upload's `expiresAt` passed (see Chunked uploads for recovery). - `cancelReason` distinguishes a user cancel (`'user'`) from a system kill (`'system'`). - Duplicate journal entries for one upload id are possible if the process dies at @@ -113,22 +189,61 @@ Notes: All methods are on the default export. +### `configure(options): void` +One-time setup — call at app startup. `options.android` sets the upload +notification's text and identity: +`notificationId/Title/TitleNoWifi/TitleNoInternet/Channel`. The config is +persisted natively, so a worker relaunched by WorkManager with no JS running +shows the same text. Optional: omitted fields keep the library defaults (each +call replaces the whole config). A no-op on iOS, which has no library +notification. + ### `startUpload(options): Promise` -Starts an upload; resolves to its id. Rejects only on a bad option (missing/invalid -`url` or `path`) — transport failures and HTTP error responses arrive later as -`error` events, not a rejection. +Starts an upload; resolves to its id. Discriminated on `options.type`: `'raw'` +sends the whole file as one request body, `'chunked'` sends the authored parts +(see Chunked uploads). Rejects (or, for malformed chunked input, throws +synchronously) only on bad options — transport failures and HTTP error responses +arrive later as `error` events, not a rejection. + +**Idempotent for every upload, always.** Calling `startUpload` again with an id +that is already pending or running is never an error: a raw upload resolves with +the same id instead of starting a duplicate; a chunked upload reconciles — parts +already accepted are skipped, the rest continue with the new call's headers. No +pre-dispatch dedupe is needed on your side. + +Options for `type: 'raw'`: | Option | Type | Notes | | --- | --- | --- | | `url` | string | Required. | | `path` | string | Required. Local file path (`file://…`). URIs are not escaped for you. | -| `type` | `'raw'` | Only `raw` is supported. | | `method` | string | Default `POST`. | | `headers` | object | HTTP headers. | -| `customUploadId` | string | Defaults to a generated UUID. | +| `id` | string | Defaults to a generated UUID. | +| `wifiOnly` | boolean | Wait for wifi before/while uploading. | +| `accept` | AcceptRule[] | Non-2xx responses to treat as success — see Accept rules. | +| `android` | object | Optional. `noNotification` (default false) — see Silent uploads. Notification text is set once via `configure()`, not per upload. | + +Options for `type: 'chunked'`: + +| Option | Type | Notes | +| --- | --- | --- | +| `id` | string | Required — your durable id. | +| `path` | string | Required. The library takes ownership of the file — see Chunked uploads. | +| `parts` | array | Required. `{ url, headers, range: { start, end } }` per part; ranges in bytes, end exclusive. Sent verbatim as PUTs. | +| `expiresAt` | number | Required, epoch ms. Past it: terminal `error` with `errorKind: 'expired'`. | +| `accept` | AcceptRule[] | See Accept rules. | | `wifiOnly` | boolean | Wait for wifi before/while uploading. | -| `acceptStatus` | number[] | Non-2xx statuses to treat as success. | -| `android` | object | Optional. `notificationId/Title/TitleNoWifi/TitleNoInternet/Channel`, `maxRetries` (default 5), `noNotification` (default false). Sensible defaults + auto-created channel if omitted. | +| `android` | object | Same as raw. | + +#### Accept rules + +`accept: Array<{ status: number, bodyIncludes?: string }>` — non-2xx responses +to treat as success, for both upload types. `bodyIncludes` narrows a rule by +response-body substring, for servers where one status carries several meanings +distinguishable only by message. A response matching a rule completes the +request (for chunked, marks the part accepted); any other non-2xx is an `error` +with `errorKind: 'http'`. #### Silent uploads (Android) @@ -141,15 +256,33 @@ to defer it, or to stop it mid-flight and let WorkManager re-run it later. Keep the notification for anything that takes real time to upload; reserve `noNotification` for small payloads a restart would cost nothing. -Uploads sharing a `notificationId` share one notification, and its progress bar -reports every in-flight upload — silent ones included. +All uploads share one notification (identified by the configured +`notificationId`), and its progress bar reports every in-flight upload — silent +ones included. ### `cancelUpload(uploadId): Promise` -Cancels an upload. Fires a `cancelled` event with `cancelReason: 'user'`. - -### `addListener(eventType, uploadId | null, listener): EventSubscription` -Listen for `'progress' | 'error' | 'completed' | 'cancelled'`. Pass `null` for -`uploadId` to receive events for all uploads. Call `.remove()` on the result to +Cancels an upload. Fires a `cancelled` event with `cancelReason: 'user'`. For a +chunked upload this cancels in-flight requests but keeps the manifest and bytes — +the next `startUpload` with the same id resumes it (there is no separate pause +API). + +### `removeUpload(uploadId): Promise` +Releases an upload's native manifest and bytes. Every terminal outcome other +than an acked `completed` (expired, error, cancelled) keeps both so you can +resume or recreate; call this once neither is wanted. + +### `chunkPlan(sizeBytes, { min?, max? }): Array<{ start, end }>` +Splits a byte count into contiguous, end-exclusive ranges: a deterministic +greedy walk of `max`-sized chunks (default 20MB), with a final remainder smaller +than `min` (default 8MB) absorbed into the previous chunk. A file smaller than +`min` is a single chunk. Pure and deterministic on purpose: call it once and +derive both your server's part count and the `parts` array from the same result, +so the two can never disagree. + +### `addListener(eventType, listener): EventSubscription` +`addListener(event: 'progress' | 'error' | 'completed' | 'cancelled', callback)`. +Listeners are global — there is no per-upload subscription; every event carries +the upload's `id`, so discriminate on it. Call `.remove()` on the result to unsubscribe. ### `getUnacknowledgedEvents(): Promise` @@ -161,10 +294,6 @@ Removes journaled events once processed. ### `getAllUploads(): Promise` Uploads the OS still knows about, for boot-time reconciliation. -### `ios.getUploadStatus(uploadId)` -iOS-only live task state (`running | suspended | canceling`, plus byte counts), or -`undefined` if the task isn't active. - ### `android.addNotificationListener(listener)` Fires when the Android progress notification is pressed. No event data. @@ -173,8 +302,8 @@ Fires when the Android progress notification is pressed. No event data. | Event | Data | | --- | --- | | `progress` | `{ id, progress: 0-100 }` | -| `completed` | `{ id, responseCode, responseBody, responseHeaders?, eventId? }` | -| `error` | `{ id, error, errorKind?, responseCode?, responseBody?, responseHeaders? }` | +| `completed` | `{ id, responseCode?, responseBody?, responseHeaders?, eventId? }` — response fields on simple uploads only; a chunked `completed` carries none (no single response represents N parts) | +| `error` | `{ id, error, errorKind?, partIndex?, responseCode?, responseBody?, responseHeaders? }` | | `cancelled` | `{ id, cancelReason?: 'user' | 'system' }` | # Contributing diff --git a/android/consumer-rules.pro b/android/consumer-rules.pro index 3da6e998..754b74ca 100644 --- a/android/consumer-rules.pro +++ b/android/consumer-rules.pro @@ -1,25 +1,32 @@ -# Shipped to consumers via consumerProguardFiles, so a minified release build of -# the host app keeps these guarantees. +# These rules go to consumers through consumerProguardFiles. Thus a minified +# release build of the host app keeps these guarantees. # -# Upload and EventJournal.Entry are persisted with Gson — Upload into WorkManager -# input data, Entry into the on-disk event journal — and read back later, across -# app restarts AND across app updates. Gson resolves fields reflectively by name -# and needs the generic Signature attribute to reconstruct typed collections, so -# R8 renaming either one corrupts persisted state silently: +# 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: # -# * Upload.acceptStatus is a List. Without Signature, Gson deserializes it -# as List, so acceptStatus.contains(code) never matches and a -# configured accept status (e.g. 409) is reported as an http error instead of -# a completed upload. -# * A journal Entry written by an older build fails to parse if field names -# changed, and is then dropped as malformed — losing exactly the terminal -# outcomes the journal exists to preserve. +# * Upload.accept is a List. 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. # -# Debug builds are unminified and round-trip symmetrically, so neither failure is -# reproducible without R8; keep these rules. +# Debug builds are not minified and round-trip correctly. Thus 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$* { *; } +-keep class ai.openspace.backgroundupload.UploadOutcome$AcceptRule { *; } diff --git a/android/src/main/java/ai/openspace/backgroundupload/ChunkedEngine.kt b/android/src/main/java/ai/openspace/backgroundupload/ChunkedEngine.kt new file mode 100644 index 00000000..e398c750 --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/ChunkedEngine.kt @@ -0,0 +1,124 @@ +package ai.openspace.backgroundupload + +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit + +/** + * The pure scheduling half of chunked execution: the window, the retry + * policy, and the backoff. It is kept free of Android and OkHttp types. Thus + * the highest-consequence invariants (at most WINDOW parts in flight, and + * never two requests for one part index) are unit-testable on a plain JVM. + * [ChunkedUploadWorker] supplies the part executor. + */ +object ChunkedEngine { + + // The number of parts of one upload in flight at one time. This is a library + // constant, not an option. If soak data shows that a different value is + // better, this constant changes, not the API. + const val WINDOW = 3 + + // The library retries a non-accepted, non-transient HTTP response this many + // times per part. Then the response becomes a terminal error and stalls the + // upload. The budget is small on purpose. A response that the server repeats + // (401, 400) does not change without a new startUpload. Only transient + // failures retry without a limit. + const val PART_HTTP_RETRIES = 3 + + // The poll interval while the network is unusable (offline, or waiting for + // wifi). The interval is constant, not exponential. We wait for conditions + // here; we do not back off a server. And expiresAt bounds the total wait. + const val CONNECTIVITY_POLL_MS = 10_000L + + private const val BACKOFF_BASE_MS = 1_000L + private const val BACKOFF_CAP_MS = 60_000L + + // A 5xx means that the server failed, not that the request is wrong. Thus it + // retries like a transport failure: without a limit, until expiresAt. + fun isTransientHttp(code: Int) = code in 500..599 + + /** What a starting worker must do for the manifest that it finds (or does not find). */ + enum class StartAction { + /** + * No manifest exists. The upload was completed and acknowledged, or it was + * explicitly removed, while this run sat in the queue. Both are legitimate + * ends, already reported (or deliberately not reported). Exit with success + * and in silence. A journaled terminal here would be a spurious 'file' + * error for an upload that nobody owns any more. + */ + NO_MANIFEST, + + /** + * Every part is already accepted: this is a trailing resume run. Re-report + * the journaled completion (never mint a second terminal event) and stop. + * Start no foreground service and no transfers. + */ + ALREADY_COMPLETE, + + /** Pending parts remain. Run the engine. */ + RUN, + } + + fun startAction(manifest: ChunkedManifest?): StartAction = when { + manifest == null -> StartAction.NO_MANIFEST + manifest.allAccepted -> StartAction.ALREADY_COMPLETE + else -> StartAction.RUN + } + + /** How a run that found (or produced) an all-accepted manifest reports the completion. */ + sealed class CompletionReport { + /** An unacknowledged 'completed' entry exists. Re-emit it. Never mint a second entry. */ + data class ReEmit(val entry: EventJournal.Entry) : CompletionReport() + + /** A fresh completion with no journal entry yet. Journal and emit a new entry. */ + object Mint : CompletionReport() + + /** + * A trailing run with nothing unacknowledged: the completion was journaled + * AND acknowledged. Nobody is owed an event. This occurs when the trailing + * run races ackEvents, which deletes the journal entry just before the + * manifest. An event minted here would be a duplicate 'completed' for an + * upload that the consumer already settled. + */ + object None : CompletionReport() + } + + fun completionReport( + unacked: List, + uploadId: String, + freshCompletion: Boolean, + ): CompletionReport { + val existing = unacked.firstOrNull { it.uploadId == uploadId && it.type == "completed" } + return when { + existing != null -> CompletionReport.ReEmit(existing) + freshCompletion -> CompletionReport.Mint + else -> CompletionReport.None + } + } + + /** Exponential backoff for transient failures: 1s, 2s, 4s, and more, capped at 60s. */ + fun backoffMs(attempt: Int): Long = + (BACKOFF_BASE_MS shl (attempt - 1).coerceIn(0, 6)).coerceAtMost(BACKOFF_CAP_MS) + + /** + * Runs [executePart] exactly one time per index, with at most [window] parts + * at one time. One coroutine per part index is what guarantees that no two + * requests for the same part are in flight (concurrent PUTs of one partNum + * are verified unsafe on the server side). An executor that throws cancels + * the remaining parts, and the error propagates. Terminal classification is + * the caller's job. + */ + suspend fun run( + partIndexes: List, + window: Int = WINDOW, + executePart: suspend (Int) -> Unit, + ) { + val gate = Semaphore(window) + coroutineScope { + for (index in partIndexes) { + launch { gate.withPermit { executePart(index) } } + } + } + } +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/ChunkedManifest.kt b/android/src/main/java/ai/openspace/backgroundupload/ChunkedManifest.kt new file mode 100644 index 00000000..6bbe523b --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/ChunkedManifest.kt @@ -0,0 +1,298 @@ +package ai.openspace.backgroundupload + +import android.content.Context +import com.facebook.react.bridge.ReadableMap +import com.google.gson.Gson +import java.io.File +import java.util.Base64 + +/** + * The durable record of one chunked upload: the moved source file, the parts + * that the consumer authored, and which of them the server has accepted. + * [ChunkedManifestStore] persists it as JSON at startUpload, BEFORE the work + * is enqueued. Thus a worker rescheduled after process death (or a startUpload + * after a crash, a stop, or a reauth) resumes from it without a call into JS. + * This manifest IS the resume mechanism. + * + * The data shape is kept free of Android and React types (Gson round-trips + * it, and JVM tests construct it directly). The ReadableMap parsing lives in + * the companion, like [Upload]'s. + */ +data class ChunkedManifest( + val id: String, + /** The library-owned copy of the bytes (the consumer's file, renamed in). */ + val sourcePath: String, + val parts: List, + val accept: List, + /** Epoch ms. After this time, the upload stops with errorKind 'expired'. */ + val expiresAt: Long, + val wifiOnly: Boolean, + val noNotification: Boolean, + val createdAt: Long, +) { + /** + * One part, exactly as the consumer authored it. The library sends the file + * bytes [start, end) as the body of a PUT to [url], with [headers] + * unchanged. It never derives or edits a protocol field. + */ + data class Part( + val url: String, + val headers: Map, + val start: Long, + val end: Long, // exclusive + val accepted: Boolean = false, + ) { + val size get() = end - start + } + + val showsNotification get() = !noNotification + val totalBytes get() = parts.sumOf { it.size } + val acceptedBytes get() = parts.filter { it.accepted }.sumOf { it.size } + + /** The server's auto-publish condition. It is the only thing that 'completed' may mean. */ + val allAccepted get() = parts.all { it.accepted } + + fun isExpired(now: Long) = now >= expiresAt + + fun pendingIndexes() = parts.indices.filter { !parts[it].accepted } + + fun withPartAccepted(index: Int) = copy( + parts = parts.mapIndexed { i, part -> if (i == index) part.copy(accepted = true) else part }, + ) + + class ReconcileException(message: String) : IllegalArgumentException(message) + + /** + * A startUpload re-call with an existing id is one of two things: + * + * **Resume** — the incoming parts are the SAME array (identical count, + * ranges, and urls). The headers, the accept rules, expiresAt, and the flags + * come from the new call. This is how fresh auth reaches stalled parts, and + * how a salvage extends the deadline. The accepted part statuses, the moved + * source, and createdAt survive from this manifest. A resume is permitted at + * any time, running or not. A running worker re-reads the stored copy before + * every attempt. + * + * **Recreate** — a DIFFERENT parts array. The consumer re-authored the + * upload under a fresh server uploadId after the old one died (it expired + * past the server's 31-day window, or it is otherwise unrecoverable). The + * owned bytes are kept. The parts are replaced as a whole, and every part + * status resets to unsent. The headers, the accept rules, and expiresAt come + * from the new call. The new ranges must tile exactly [0, blobSize). A + * partial or overlapping cover would silently upload wrong bytes. A recreate + * is accepted only while the upload is NOT running (stalled on a terminal + * error, expired, or cancelled). A different parts array while a worker + * executes is a consumer bug, not a recreate, because the in-flight requests + * belong to the old parts. + */ + fun reconcile(incoming: ChunkedManifest, running: Boolean, blobSize: Long): ChunkedManifest { + if (samePartsAs(incoming)) { + // Accepted flags follow the RANGE, not the array index. samePartsAs is + // order-independent, so the same tile can sit at a different index. + val acceptedStarts = parts.filter { it.accepted }.map { it.start }.toSet() + return incoming.copy( + sourcePath = sourcePath, + createdAt = createdAt, + parts = incoming.parts.map { it.copy(accepted = it.start in acceptedStarts) }, + ) + } + if (running) throw ReconcileException( + "chunked upload '$id' is running; a different parts array is only accepted once it stops", + ) + if (!tilesExactly(incoming.parts, blobSize)) throw ReconcileException( + "chunked upload '$id' recreate parts must tile exactly [0, $blobSize)", + ) + return incoming.copy(sourcePath = sourcePath, createdAt = createdAt) + } + + // Order-independent, like tilesExactly. The same tiles, authored in a + // different order, are the SAME upload (a resume), never a recreate. + private fun samePartsAs(incoming: ChunkedManifest): Boolean { + if (incoming.parts.size != parts.size) return false + val stored = parts.sortedBy { it.start } + val fresh = incoming.parts.sortedBy { it.start } + return stored.indices.all { i -> + fresh[i].url == stored[i].url && + fresh[i].start == stored[i].start && + fresh[i].end == stored[i].end + } + } + + companion object { + /** + * Whether [parts] cover [0, size) exactly: no gap, no overlap, and nothing + * past the end. Order-independent, like everything else about parts. + */ + fun tilesExactly(parts: List, size: Long): Boolean { + if (parts.isEmpty()) return false + val sorted = parts.sortedBy { it.start } + var cursor = 0L + for (part in sorted) { + if (part.start != cursor || part.end <= part.start) return false + cursor = part.end + } + return cursor == size + } + + /** + * Validates a first-call (create) manifest against the just-owned bytes. + * Like a recreate, the parts must tile exactly [0, blobSize). A partial or + * overlapping cover would silently upload wrong bytes. It throws BEFORE + * the manifest is saved. Thus the moved blob stays adoptable by a + * corrected retry (see UploaderModule.takeOwnership's orphan branch). + */ + fun validatedForCreate(incoming: ChunkedManifest, blobSize: Long): ChunkedManifest { + if (!tilesExactly(incoming.parts, blobSize)) throw ReconcileException( + "chunked upload '${incoming.id}' parts must tile exactly [0, $blobSize)", + ) + return incoming + } + + /** @param sourcePath the library-owned destination, not the consumer's path. */ + fun fromReadableMap(map: ReadableMap, sourcePath: String, createdAt: Long): ChunkedManifest { + val partsArr = map.getArray("parts") ?: throw Upload.MissingOptionException("parts") + if (partsArr.size() == 0) throw IllegalArgumentException("parts must be a non-empty array") + if (!map.hasKey("expiresAt")) throw Upload.MissingOptionException("expiresAt") + return ChunkedManifest( + id = map.getString("id") ?: throw Upload.MissingOptionException("id"), + sourcePath = sourcePath, + parts = (0 until partsArr.size()).map { i -> + val part = partsArr.getMap(i) ?: throw Upload.MissingOptionException("parts[$i]") + val range = part.getMap("range") ?: throw Upload.MissingOptionException("parts[$i].range") + Part( + url = part.getString("url") ?: throw Upload.MissingOptionException("parts[$i].url"), + headers = parseHeaderMap(part.getMap("headers")), + start = range.getDouble("start").toLong(), + end = range.getDouble("end").toLong(), + ) + }, + accept = parseAcceptRules(map.getArray("accept")), + expiresAt = map.getDouble("expiresAt").toLong(), + wifiOnly = if (map.hasKey("wifiOnly")) map.getBoolean("wifiOnly") else false, + noNotification = if (map.hasKey("noNotification")) map.getBoolean("noNotification") else false, + createdAt = createdAt, + ) + } + } +} + +/** + * A file-backed store: one directory per upload id, which holds + * `manifest.json` and `blob` (the moved source bytes). It has the same + * durability pattern as [EventJournal]: tmp+rename writes, and corrupt files + * read as absent. It is reachable from a bare Context, because the worker can + * run in a process where React never initialized. + */ +class ChunkedManifestStore(private val dir: File) { + + companion object { + private val gson = Gson() + + @Volatile + private var instance: ChunkedManifestStore? = null + + fun get(context: Context): ChunkedManifestStore = + instance ?: synchronized(this) { + instance ?: ChunkedManifestStore(File(context.filesDir, "rnbgupload-chunked")) + .also { instance = it } + } + } + + init { + dir.mkdirs() + } + + // Upload ids come from the consumer, and they can contain path separators or + // other filesystem-hostile characters. Thus the directory name is an encoding + // of the id, never the id itself. The id is read back from the manifest, not + // decoded from the name. + private fun uploadDir(id: String) = + File(dir, Base64.getUrlEncoder().withoutPadding().encodeToString(id.toByteArray())) + + private fun manifestFile(id: String) = File(uploadDir(id), "manifest.json") + + /** Where startUpload moves the source file for this id. */ + fun blobFile(id: String) = File(uploadDir(id), "blob") + + @Synchronized + fun load(id: String): ChunkedManifest? { + val file = manifestFile(id) + if (!file.exists()) return null + val parsed = runCatching { gson.fromJson(file.readText(), ChunkedManifest::class.java) } + .getOrNull() + return validated(parsed) + } + + /** Throws on a write failure. A manifest that did not persist must fail the startUpload call. */ + @Synchronized + fun save(manifest: ChunkedManifest) { + val dir = uploadDir(manifest.id) + dir.mkdirs() + val tmp = File(dir, "manifest.tmp") + tmp.writeText(gson.toJson(manifest)) + if (!tmp.renameTo(manifestFile(manifest.id))) { + throw java.io.IOException("failed to persist chunked manifest for '${manifest.id}'") + } + } + + /** + * An atomic read-modify-write. Thus a worker that marks a part accepted can + * never clobber a concurrent startUpload's fresh headers (or another part's + * flag). Returns null, without a throw, when the manifest is gone or the + * write failed. A caller that can continue from memory does that. + */ + @Synchronized + fun update(id: String, transform: (ChunkedManifest) -> ChunkedManifest): ChunkedManifest? = + runCatching { + val manifest = load(id) ?: return null + val next = transform(manifest) + save(next) + next + }.getOrNull() + + /** + * An atomic create-or-transform. The store lock spans load, [transform], and + * save. Thus nothing — a running worker's markAccepted included — can write + * between them and be erased. startUpload's load, reconcile, and save must + * go through here, not as three separate calls. [transform] receives null + * when no manifest exists. Unlike [update], a transform that throws (a + * reconcile rejection) or a failed write propagates, because startUpload + * must fail loudly, not continue from memory. + */ + @Synchronized + fun compute(id: String, transform: (ChunkedManifest?) -> ChunkedManifest): ChunkedManifest { + val next = transform(load(id)) + save(next) + return next + } + + /** Whether a manifest is stored for this id (without parsing it). */ + @Synchronized + fun contains(id: String): Boolean = manifestFile(id).exists() + + /** Deletes the manifest AND the moved bytes. Does nothing for an unknown id (a simple upload). */ + @Synchronized + fun remove(id: String) { + uploadDir(id).deleteRecursively() + } + + @Synchronized + fun all(): List = + (dir.listFiles { f -> f.isDirectory } ?: emptyArray()) + .mapNotNull { d -> + val file = File(d, "manifest.json") + if (!file.exists()) return@mapNotNull null + validated(runCatching { gson.fromJson(file.readText(), ChunkedManifest::class.java) }.getOrNull()) + } + + // Gson does not use the constructor. Thus a corrupt or field-renamed file can + // make non-null Kotlin fields null. Reject a file that lacks a field that the + // engine relies on. Normalize an absent accept list; do not reject it. + @Suppress("SENSELESS_COMPARISON") + private fun validated(m: ChunkedManifest?): ChunkedManifest? { + if (m == null || m.id == null || m.sourcePath == null || m.parts == null) return null + if (m.parts.isEmpty()) return null + if (m.parts.any { it == null || it.url == null || it.headers == null }) return null + return if (m.accept == null) m.copy(accept = listOf()) else m + } +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/ChunkedUploadWorker.kt b/android/src/main/java/ai/openspace/backgroundupload/ChunkedUploadWorker.kt new file mode 100644 index 00000000..33750073 --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/ChunkedUploadWorker.kt @@ -0,0 +1,410 @@ +package ai.openspace.backgroundupload + +import android.app.NotificationManager +import android.content.Context +import androidx.work.CoroutineWorker +import androidx.work.ForegroundInfo +import androidx.work.ListenableWorker +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 +import java.io.File +import java.io.IOException +import java.util.UUID +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong + +/** + * Executes one chunked upload from its durable [ChunkedManifest]. The input + * data carries only the upload id. The manifest is the record: startUpload + * persists it before this work is enqueued. Thus a worker rescheduled after + * process death resumes from disk, with no JS involved. + * + * One logical upload has one event stream: byte-weighted aggregate progress, + * and one terminal event. 'completed' is journaled only when every part is + * accepted. Every other terminal keeps the manifest and the bytes, so a later + * startUpload can resume. The bytes are deleted only when a 'completed' event + * is ACKED (see UploaderModule.ackEvents). + */ +class ChunkedUploadWorker(private val context: Context, params: WorkerParameters) : + CoroutineWorker(context, params) { + + companion object { + /** + * The key for the upload id in the worker's input data. It is a string + * literal for the same reason as [UploadWorker.PARAMS_KEY]: WorkManager's + * database persists it across builds, and it must survive R8 renames and + * refactors. + */ + const val ID_KEY = "chunkedUploadId" + + /** How often a starting worker re-checks [ChunkedWorkerGate] for its id. */ + private const val GATE_POLL_MS = 100L + } + + private lateinit var uploadId: String + private val store by lazy { ChunkedManifestStore.get(context) } + private val config by lazy { NotificationConfig.load(context) } + private val notificationManager = + context.getSystemService(Context.NOTIFICATION_SERVICE) as NotificationManager + + // The latest known manifest. Part executors re-read the stored copy before + // every attempt (see latest()). Thus a reconciling startUpload's fresh + // headers, and an extended expiresAt, reach a worker that already runs. + @Volatile + private var manifest: ChunkedManifest? = null + + @Volatile + private var connectivity = Connectivity.Ok + + // In-flight bytes per part index, for byte-weighted aggregate progress. + private val partSent = ConcurrentHashMap() + private val acceptedBytes = AtomicLong(0) + + private class ExpiredException : Exception("upload expired") + + private class SourceMissingException(path: String) : + IOException("chunked source file missing: $path") + + private class PartRejectedException(val partIndex: Int, val response: UploadResponse) : + Exception("part $partIndex rejected with HTTP ${response.code}") + + private class PartBeyondEofException(val partIndex: Int, message: String) : Exception(message) + + override suspend fun doWork(): Result = withContext(Dispatchers.IO) { + uploadId = inputData.getString(ID_KEY) ?: throw Throwable("No upload id") + + // Acquire the per-id execution gate BEFORE the first manifest read. A + // cancel-then-start can start this worker while the cancelled one still + // winds down, and two PUTs of one partNum are unsafe. Also, the manifest + // read occurs only after the gate is held. That is what makes the module's + // recreate check race-free (see ChunkedWorkerGate and + // ChunkedManifestStore.compute). + try { + while (!ChunkedWorkerGate.tryAcquire(uploadId, this@ChunkedUploadWorker)) { + delay(GATE_POLL_MS) + } + } catch (error: CancellationException) { + // Cancelled while waiting. A user cancel still owes its terminal event. + checkAndHandleCancellation() + throw error + } + try { + runUpload() + } finally { + ChunkedWorkerGate.release(uploadId, this@ChunkedUploadWorker) + } + } + + private suspend fun runUpload(): Result { + val initial = store.load(uploadId) + when (ChunkedEngine.startAction(initial)) { + // The upload was completed-and-acknowledged, or it was removed, while + // this run sat in the queue. Both are legitimate and already settled. + // Exit in silence. A terminal journaled here would be a spurious error + // for an upload that nobody owns. + ChunkedEngine.StartAction.NO_MANIFEST -> return Result.success() + // A trailing resume of a finished-but-unacknowledged upload. Re-report + // the journaled completion. Skip the foreground service and the engine. + ChunkedEngine.StartAction.ALREADY_COMPLETE -> { + manifest = initial + journalCompleted(freshCompletion = false) + return Result.success() + } + ChunkedEngine.StartAction.RUN -> Unit + } + checkNotNull(initial) // RUN implies a manifest + manifest = initial + acceptedBytes.set(initial.acceptedBytes) + UploadProgress.add(uploadId, initial.totalBytes) + UploadProgress.set(uploadId, initial.acceptedBytes) + + // Initialization. A failure here is terminal: journaled, never retried. + // The EXCEPTION is a refused foreground start, which the transfer + // survives. + try { + if (initial.showsNotification) { + ensureNotificationChannel(notificationManager, config) + setForeground(getForegroundInfo()) + } + } catch (error: Throwable) { + if (!isForegroundStartDenied(error)) { + if (!checkAndHandleCancellation()) { + UploadProgress.remove(uploadId) + handleFailure(error) + } + return terminalErrorResult() + } + // The app is in the background, and API 31+ refused the foreground + // start. This is the usual state for a WorkManager relaunch (a reboot, + // or a quota resume). The upload runs correctly without foreground + // priority. A failure here would brick every headless resume. + } + + return try { + ChunkedEngine.run(initial.pendingIndexes()) { index -> executePart(index) } + // Every executor returned. An executor returns only when its part was + // accepted. That is exactly the server's auto-publish condition. + UploadProgress.complete(uploadId) + journalCompleted(freshCompletion = true) + Result.success() + } catch (error: Throwable) { + if (checkAndHandleCancellation()) throw error + UploadProgress.remove(uploadId) + handleFailure(error) + terminalErrorResult() + } + } + + /** + * Uploads one part until it is accepted, or throws. Terminal conditions + * (expiry, a missing source, or a non-accepted response out of retries) + * propagate and cancel the sibling parts. Everything transient retries here, + * bounded only by expiresAt. + */ + private suspend fun executePart(index: Int) { + var rejections = 0 + var transientAttempts = 0 + while (true) { + val current = latest() + val part = current.parts[index] + if (part.accepted) return + if (current.isExpired(System.currentTimeMillis())) throw ExpiredException() + + // A range past the blob's EOF can never transmit. The read would fail + // on every attempt until expiry. Thus it is a terminal 'file' error + // immediately (iOS classifies it the same way). length() is 0 for a + // missing file. That case falls through to the transfer, which + // classifies it as source-missing. The failed-probe-reads-as-network + // default stays intact. + val blobLength = runCatching { File(current.sourcePath).length() }.getOrDefault(0L) + if (blobLength > 0L && part.end > blobLength) throw PartBeyondEofException( + index, + "part $index range [${part.start}, ${part.end}) exceeds source size $blobLength", + ) + + if (!validateAndReportConnectivity(current.wifiOnly)) { + delay(ChunkedEngine.CONNECTIVITY_POLL_MS) + continue + } + + val response = try { + transferSemaphore.withPermit { + okhttpUploadPart(uploadHttpClient, part, File(current.sourcePath)) { sent -> + onPartProgress(index, sent) + } + } + } catch (error: CancellationException) { + throw error + } catch (error: IOException) { + onPartProgress(index, 0L) + // The default is fileExists=true. Thus a failed probe reads as + // network, not file. + val fileExists = runCatching { File(current.sourcePath).exists() }.getOrDefault(true) + if (!fileExists) throw SourceMissingException(current.sourcePath) + transientAttempts++ + delay(ChunkedEngine.backoffMs(transientAttempts)) + continue + } + + if (UploadOutcome.isAccepted(response.code, response.body, current.accept)) { + markAccepted(index) + return + } + onPartProgress(index, 0L) + if (ChunkedEngine.isTransientHttp(response.code)) { + transientAttempts++ + delay(ChunkedEngine.backoffMs(transientAttempts)) + continue + } + rejections++ + if (rejections > ChunkedEngine.PART_HTTP_RETRIES) throw PartRejectedException(index, response) + delay(ChunkedEngine.backoffMs(rejections)) + } + } + + // The stored copy is the truth: a reconcile can have replaced the headers + // or expiresAt. Fall back to the in-memory copy only when the read fails. + private fun latest(): ChunkedManifest = + store.load(uploadId)?.also { manifest = it } ?: manifest!! + + private fun markAccepted(index: Int) { + // Persist the flag first, atomically against concurrent flips and + // reconciles. This is best-effort. A lost flag only re-sends this part on + // a later resume, and the consumer's accept rules absorb that ('already + // completed'). That is better than a failure of an upload that the server + // accepted. + manifest = store.update(uploadId) { it.withPartAccepted(index) } + ?: manifest?.withPartAccepted(index) + manifest?.parts?.get(index)?.let { acceptedBytes.addAndGet(it.size) } + partSent.remove(index) + reportProgress() + } + + private fun onPartProgress(index: Int, sent: Long) { + if (sent == 0L) partSent.remove(index) else partSent[index] = sent + reportProgress() + } + + private fun reportProgress() { + val total = manifest?.totalBytes ?: return + val sent = (acceptedBytes.get() + partSent.values.sum()).coerceAtMost(total) + UploadProgress.set(uploadId, sent) + EventReporter.progress(uploadId, sent, total) + updateNotification() + } + + // A resume of a finished-but-unacknowledged upload (all parts accepted, + // 'completed' journaled, and the consumer re-called startUpload before the + // ack) must not mint a second terminal event. Re-emit the journaled one. + // Then a live listener still hears it, with the eventId that the consumer + // will acknowledge. And a trailing run whose completion was already ACKED + // reports nothing at all. See ChunkedEngine.CompletionReport. + private fun journalCompleted(freshCompletion: Boolean) { + val report = ChunkedEngine.completionReport( + EventJournal.get(context).unacknowledged(), + uploadId, + freshCompletion, + ) + when (report) { + is ChunkedEngine.CompletionReport.ReEmit -> EventReporter.emit(report.entry) + // No response fields, because no single response represents N accepted + // parts. + ChunkedEngine.CompletionReport.Mint -> journalAndEmit( + EventJournal.Entry( + eventId = UUID.randomUUID().toString(), + uploadId = uploadId, + type = "completed", + timestamp = System.currentTimeMillis(), + ), + ) + ChunkedEngine.CompletionReport.None -> Unit + } + } + + private fun handleFailure(error: Throwable) { + val entry = when (error) { + is ExpiredException -> errorEntry( + error = "upload expired before every part was accepted", + errorKind = "expired", + ) + is PartRejectedException -> { + val (body, truncated) = EventJournal.capBody(error.response.body) + errorEntry( + error = "HTTP ${error.response.code} on part ${error.partIndex}", + errorKind = "http", + partIndex = error.partIndex, + responseCode = error.response.code, + responseBody = body, + responseBodyTruncated = truncated, + responseHeaders = error.response.headers, + ) + } + is SourceMissingException -> errorEntry(error = error.message!!, errorKind = "file") + is PartBeyondEofException -> errorEntry( + error = error.message!!, + errorKind = "file", + partIndex = error.partIndex, + ) + else -> { + val fileExists = manifest?.let { m -> + runCatching { File(m.sourcePath).exists() }.getOrDefault(true) + } ?: true + errorEntry( + error = error.message ?: "Unknown exception", + errorKind = UploadOutcome.errorKind(error, fileExists), + ) + } + } + journalAndEmit(entry) + } + + private fun errorEntry( + error: String, + errorKind: String, + partIndex: Int? = null, + responseCode: Int? = null, + responseBody: String? = null, + responseBodyTruncated: Boolean = false, + responseHeaders: Map? = null, + ) = EventJournal.Entry( + eventId = UUID.randomUUID().toString(), + uploadId = uploadId, + type = "error", + timestamp = System.currentTimeMillis(), + error = error, + errorKind = errorKind, + partIndex = partIndex, + responseCode = responseCode, + responseBody = responseBody, + responseBodyTruncated = responseBodyTruncated, + responseHeaders = responseHeaders, + ) + + // The semantics are the same as UploadWorker's. Only a user cancel is + // terminal (journaled, cancelReason 'user'). A system stop emits nothing, + // because WorkManager will re-run this upload, and the manifest resumes it. + // The manifest and the bytes are kept in both cases. stopUpload's contract + // is that the next startUpload resumes. + private fun checkAndHandleCancellation(): Boolean { + if (!isStopped) return false + + UploadProgress.remove(uploadId) + + if (!UserCancellations.consume(uploadId)) return true + + journalAndEmit( + EventJournal.Entry( + eventId = UUID.randomUUID().toString(), + uploadId = uploadId, + type = "cancelled", + timestamp = System.currentTimeMillis(), + cancelReason = "user", + ), + ) + return true + } + + private fun journalAndEmit(entry: EventJournal.Entry) { + // A terminal event for an id whose manifest is gone would report an + // upload that nobody owns any more. Either removeUpload deleted it mid-run + // (its work cancel races the in-flight PUT's IOException), or a completed + // ack released it. Suppress the event; iOS's removedIds has the same idea. + // A user cancel keeps its manifest, so real 'cancelled' events pass + // through. + if (!store.contains(uploadId)) return + EventReporter.journalAndEmit(context, entry) + } + + private fun validateAndReportConnectivity(wifiOnly: Boolean): Boolean { + connectivity = validateConnectivity(context, wifiOnly) + updateNotification() + return connectivity == Connectivity.Ok + } + + private fun updateNotification() { + if (manifest?.showsNotification != true) return + notificationManager.notify( + config.systemNotificationId, + buildUploadNotification(context, config, connectivity), + ) + } + + override suspend fun getForegroundInfo(): ForegroundInfo = + uploadForegroundInfo(config, buildUploadNotification(context, config, connectivity)) +} + +/** + * The Result that a chunked run returns after it journals a terminal error: + * SUCCESS, deliberately. The journal and the manifest are the upload's outcome + * record, never the WorkManager row state. A row that finishes FAILED destroys + * every appended dependent: WorkManager marks the dependents of a failed + * prerequisite FAILED without a run. Thus a resume enqueued during the failing + * run's teardown window would silently never run (see the APPEND_OR_REPLACE + * note in UploaderModule.enqueueChunkedUpload). getAllUploads derives a + * chunked upload's state from its manifest (allAccepted), not from row states. + */ +internal fun terminalErrorResult(): ListenableWorker.Result = ListenableWorker.Result.success() diff --git a/android/src/main/java/ai/openspace/backgroundupload/ChunkedWorkerGate.kt b/android/src/main/java/ai/openspace/backgroundupload/ChunkedWorkerGate.kt new file mode 100644 index 00000000..424c627b --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/ChunkedWorkerGate.kt @@ -0,0 +1,40 @@ +package ai.openspace.backgroundupload + +import java.util.concurrent.ConcurrentHashMap + +/** + * At most one [ChunkedUploadWorker] EXECUTES per upload id, process-wide. + * + * The unique-work chain almost guarantees this, but not across a cancel. + * cancelUniqueWork marks the row CANCELLED immediately, while the cancelled + * worker's coroutine still winds down. Thus a startUpload that arrives right + * after a cancelUpload can enqueue (and start) a replacement worker while the + * old worker still has a part PUT in flight. Two concurrent PUTs of one + * partNum are verified unsafe on the server side. A starting worker acquires + * its id here, and a successor waits for the release. + * + * This is also the truthful "is this upload running" for the recreate rule. + * A worker registers before its first manifest read, and it releases in a + * finally block. WorkManager's row state stays RUNNING for a moment after + * doWork returns. This gate does not: it never reports a finished run as + * running. + * + * The gate is same-process only, like [UserCancellations]. A worker in a dead + * process holds nothing, and WorkManager runs our workers in the app process. + */ +object ChunkedWorkerGate { + private val holders = ConcurrentHashMap() + + /** True when [token] now holds the id, or already held it. False while another token holds it. */ + fun tryAcquire(id: String, token: Any): Boolean { + val current = holders.putIfAbsent(id, token) + return current == null || current === token + } + + /** Releases only when [token] is the holder. Thus a stale release cannot evict a successor. */ + fun release(id: String, token: Any) { + holders.remove(id, token) + } + + fun isRunning(id: String): Boolean = holders.containsKey(id) +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/EventJournal.kt b/android/src/main/java/ai/openspace/backgroundupload/EventJournal.kt index 5bbc0738..2dd718b3 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EventJournal.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/EventJournal.kt @@ -29,8 +29,11 @@ class EventJournal( val responseBodyTruncated: Boolean = false, val responseHeaders: Map? = null, val error: String? = null, - val errorKind: String? = null, // http | network | file | unknown + val errorKind: String? = null, // http | network | file | expired | unknown val cancelReason: String? = null, // user | system + // Chunked uploads: the index of the failing part, when one part's response + // caused the error. + val partIndex: Int? = null, ) { fun toWritableMap(): com.facebook.react.bridge.WritableMap = com.facebook.react.bridge.Arguments.createMap().apply { @@ -47,6 +50,7 @@ class EventJournal( error?.let { putString("error", it) } errorKind?.let { putString("errorKind", it) } cancelReason?.let { putString("cancelReason", it) } + partIndex?.let { putInt("partIndex", it) } } } diff --git a/android/src/main/java/ai/openspace/backgroundupload/EventReporter.kt b/android/src/main/java/ai/openspace/backgroundupload/EventReporter.kt index a4770c78..46460e1e 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/EventReporter.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/EventReporter.kt @@ -1,5 +1,6 @@ package ai.openspace.backgroundupload +import android.content.Context import com.facebook.react.bridge.Arguments // Sends live events to JS through the module's codegen event emitters. Terminal @@ -8,6 +9,14 @@ import com.facebook.react.bridge.Arguments // it up from getUnacknowledgedEvents instead. object EventReporter { + // Journal first, then emit. The journal is the durable record; it survives + // when JS is dead. The live emit is best-effort. The two carry the identical + // payload. Thus a consumer can acknowledge a live event by its eventId. + fun journalAndEmit(context: Context, entry: EventJournal.Entry) { + EventJournal.get(context).append(entry) + emit(entry) + } + // Emit a terminal event from its journal entry, so the live event carries the // exact same payload (incl. eventId) as the journaled copy — letting a consumer // ackEvents([eventId]) right after handling a live event, and keeping iOS/Android diff --git a/android/src/main/java/ai/openspace/backgroundupload/NotificationConfig.kt b/android/src/main/java/ai/openspace/backgroundupload/NotificationConfig.kt new file mode 100644 index 00000000..ec947bfd --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/NotificationConfig.kt @@ -0,0 +1,83 @@ +package ai.openspace.backgroundupload + +import android.content.Context +import com.facebook.react.bridge.ReadableMap +import com.google.gson.Gson + +/** + * The text and the identity of the upload progress notification. JS sets it + * one time with `configure()`. The class writes it to SharedPreferences. Thus a + * worker that WorkManager relaunches — with no JS, possibly in a process where + * React never initialized — shows the same text. Each field falls back to a + * library default. Thus uploads work when `configure()` was never called. + */ +data class NotificationConfig( + val notificationId: String, + val notificationTitle: String, + val notificationTitleNoInternet: String, + val notificationTitleNoWifi: String, + val notificationChannel: String, +) { + // The id given to NotificationManager. All uploads share the configured id. + // Thus they share one notification, and its progress bar is the total. + val systemNotificationId get() = notificationId.hashCode() + + companion object { + const val DEFAULT_NOTIFICATION_CHANNEL = "background-upload" + + val DEFAULTS = NotificationConfig( + notificationId = DEFAULT_NOTIFICATION_CHANNEL, + notificationTitle = "Uploading…", + notificationTitleNoInternet = "Waiting for connection…", + notificationTitleNoWifi = "Waiting for Wi-Fi…", + notificationChannel = DEFAULT_NOTIFICATION_CHANNEL, + ) + + private const val PREFS_NAME = "rnbgupload-config" + private const val PREFS_KEY = "notificationConfig" + private val gson = Gson() + + fun fromReadableMap(map: ReadableMap) = NotificationConfig( + notificationId = map.getString(NotificationConfig::notificationId.name) + ?: DEFAULTS.notificationId, + notificationTitle = map.getString(NotificationConfig::notificationTitle.name) + ?: DEFAULTS.notificationTitle, + notificationTitleNoInternet = map.getString(NotificationConfig::notificationTitleNoInternet.name) + ?: DEFAULTS.notificationTitleNoInternet, + notificationTitleNoWifi = map.getString(NotificationConfig::notificationTitleNoWifi.name) + ?: DEFAULTS.notificationTitleNoWifi, + notificationChannel = map.getString(NotificationConfig::notificationChannel.name) + ?: DEFAULTS.notificationChannel, + ) + + // Gson does not use the constructor. Thus a blob from a build with + // different fields, or a corrupt blob, can make a non-null field null. + // Each field falls back alone. A bad blob must give default text. It must + // never crash a worker. + @Suppress("SENSELESS_COMPARISON") + fun fromJson(json: String?): NotificationConfig { + val parsed = json?.let { + runCatching { gson.fromJson(it, NotificationConfig::class.java) }.getOrNull() + } ?: return DEFAULTS + return NotificationConfig( + notificationId = parsed.notificationId ?: DEFAULTS.notificationId, + notificationTitle = parsed.notificationTitle ?: DEFAULTS.notificationTitle, + notificationTitleNoInternet = parsed.notificationTitleNoInternet + ?: DEFAULTS.notificationTitleNoInternet, + notificationTitleNoWifi = parsed.notificationTitleNoWifi + ?: DEFAULTS.notificationTitleNoWifi, + notificationChannel = parsed.notificationChannel ?: DEFAULTS.notificationChannel, + ) + } + + fun save(context: Context, config: NotificationConfig) { + prefs(context).edit().putString(PREFS_KEY, gson.toJson(config)).apply() + } + + fun load(context: Context): NotificationConfig = + fromJson(prefs(context).getString(PREFS_KEY, null)) + + private fun prefs(context: Context) = + context.getSharedPreferences(PREFS_NAME, Context.MODE_PRIVATE) + } +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/Upload.kt b/android/src/main/java/ai/openspace/backgroundupload/Upload.kt index d4d753a5..3df469b0 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/Upload.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/Upload.kt @@ -1,5 +1,6 @@ package ai.openspace.backgroundupload +import com.facebook.react.bridge.ReadableArray import com.facebook.react.bridge.ReadableMap import java.util.UUID @@ -11,18 +12,12 @@ data class Upload( val url: String, val path: String, val method: String, - val maxRetries: Int, val wifiOnly: Boolean, - // Non-2xx statuses to treat as a successful completion (e.g. [409] when - // duplicate-create conflicts are expected). Everything else non-2xx is a - // terminal http error. Empty by default. - val acceptStatus: List, + // Non-2xx responses to treat as a successful completion (for example, a 409 + // whose body marks an expected duplicate). Every other non-2xx response is a + // terminal http error. The list is empty by default. + val accept: List, val headers: Map, - val notificationId: Int, - val notificationTitle: String, - val notificationTitleNoInternet: String, - val notificationTitleNoWifi: String, - val notificationChannel: String, /** * Suppresses the progress notification for this upload. * @@ -38,49 +33,75 @@ data class Upload( */ val noNotification: Boolean, ) { + // v8 persisted `acceptStatus: List` where v9 persists `accept`. This is + // not a constructor parameter. It exists only so Gson can surface the legacy + // field to [normalized]. It is null, and thus never serialized, for every + // upload that this build creates. + private val acceptStatus: List? = null + val showsNotification get() = !noNotification + /** + * Gson does not use the constructor. Thus a WorkManager job that an older + * build enqueued can give this worker an object whose non-null fields are + * null. A v8 job carries `acceptStatus` and no `accept`. That NPEs the first + * time the worker touches [accept], after the file has fully transmitted, + * and the re-runs then re-send the whole file. This is the same + * normalize-after-fromJson pattern as ChunkedManifestStore.validated(): map + * the legacy statuses to rules, default what is absent, and give the worker + * an object that is safe to use. + */ + @Suppress("SENSELESS_COMPARISON", "USELESS_ELVIS") + fun normalized(): Upload = Upload( + id = id, + url = url, + path = path, + method = method ?: "POST", + wifiOnly = wifiOnly, + accept = accept + ?: acceptStatus?.map { UploadOutcome.AcceptRule(it) } + ?: emptyList(), + headers = headers ?: emptyMap(), + noNotification = noNotification, + ) + class MissingOptionException(optionName: String) : IllegalArgumentException("Missing '$optionName'") companion object { - const val DEFAULT_NOTIFICATION_CHANNEL = "background-upload" - fun fromReadableMap(map: ReadableMap) = Upload( - id = map.getString("customUploadId") ?: UUID.randomUUID().toString(), + id = map.getString(Upload::id.name) ?: UUID.randomUUID().toString(), url = map.getString(Upload::url.name) ?: throw MissingOptionException(Upload::url.name), path = map.getString(Upload::path.name) ?: throw MissingOptionException(Upload::path.name), method = map.getString(Upload::method.name) ?: "POST", - maxRetries = if (map.hasKey(Upload::maxRetries.name)) map.getInt(Upload::maxRetries.name) else 5, wifiOnly = if (map.hasKey(Upload::wifiOnly.name)) map.getBoolean(Upload::wifiOnly.name) else false, - acceptStatus = map.getArray(Upload::acceptStatus.name)?.let { arr -> - (0 until arr.size()).map { i -> arr.getInt(i) } - } ?: listOf(), - headers = map.getMap(Upload::headers.name).let { headers -> - if (headers == null) return@let mapOf() - val map = mutableMapOf() - for (entry in headers.entryIterator) { - map[entry.key] = entry.value.toString() - } - return@let map - }, - // Notification options are optional: the library supplies sensible defaults - // and creates its own channel, so consumers don't need any notifee plumbing. - notificationId = (map.getString(Upload::notificationId.name) - ?: DEFAULT_NOTIFICATION_CHANNEL).hashCode(), - notificationTitle = map.getString(Upload::notificationTitle.name) - ?: "Uploading…", - notificationTitleNoInternet = map.getString(Upload::notificationTitleNoInternet.name) - ?: "Waiting for connection…", - notificationTitleNoWifi = map.getString(Upload::notificationTitleNoWifi.name) - ?: "Waiting for Wi-Fi…", - notificationChannel = map.getString(Upload::notificationChannel.name) - ?: DEFAULT_NOTIFICATION_CHANNEL, + accept = parseAcceptRules(map.getArray(Upload::accept.name)), + headers = parseHeaderMap(map.getMap(Upload::headers.name)), + // The notification text and identity are not per-upload options. The + // worker reads them from the NotificationConfig that configure() saved. noNotification = if (map.hasKey(Upload::noNotification.name)) map.getBoolean(Upload::noNotification.name) else false, ) } } +// Upload and ChunkedManifest share this: one accept-rules shape, one parser. +internal fun parseAcceptRules(arr: ReadableArray?): List { + if (arr == null) return listOf() + return (0 until arr.size()).mapNotNull { i -> + val rule = arr.getMap(i) ?: return@mapNotNull null + UploadOutcome.AcceptRule( + status = rule.getInt("status"), + bodyIncludes = if (rule.hasKey("bodyIncludes")) rule.getString("bodyIncludes") else null, + ) + } +} - +internal fun parseHeaderMap(headers: ReadableMap?): Map { + if (headers == null) return mapOf() + val map = mutableMapOf() + for (entry in headers.entryIterator) { + map[entry.key] = entry.value.toString() + } + return map +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadNotification.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadNotification.kt new file mode 100644 index 00000000..f1b6b90b --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadNotification.kt @@ -0,0 +1,135 @@ +package ai.openspace.backgroundupload + +import android.app.Notification +import android.app.NotificationChannel +import android.app.NotificationManager +import android.app.PendingIntent +import android.content.Context +import android.content.Intent +import android.content.pm.ServiceInfo +import android.net.ConnectivityManager +import android.net.NetworkCapabilities.NET_CAPABILITY_VALIDATED +import android.net.NetworkCapabilities.TRANSPORT_WIFI +import android.os.Build +import android.widget.RemoteViews +import androidx.core.app.NotificationCompat +import androidx.work.ForegroundInfo + +// This file builds the progress notification and probes connectivity. The +// simple and chunked workers share it. All uploads share one notification, +// identified by the NotificationConfig that configure() saved. Its progress +// bar is the total across the uploads. + +internal enum class Connectivity { NoWifi, NoInternet, Ok } + +// This is synchronized to ensure consistent status across workers +@Synchronized +internal fun validateConnectivity(context: Context, wifiOnly: Boolean): Connectivity { + val manager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager + val network = manager.activeNetwork + val capabilities = manager.getNetworkCapabilities(network) + + val hasInternet = capabilities?.hasCapability(NET_CAPABILITY_VALIDATED) == true + + // not wifiOnly, return early + if (!wifiOnly) return if (hasInternet) Connectivity.Ok else Connectivity.NoInternet + + // handle wifiOnly + return if (hasInternet && capabilities?.hasTransport(TRANSPORT_WIFI) == true) + Connectivity.Ok + else + Connectivity.NoWifi // don't return NoInternet here, more direct to request to join wifi +} + +// Makes sure that the channel for the foreground notification exists. It makes +// the channel only when the channel is absent. Thus a channel that the consumer +// registered, with their own name and importance, always wins. When configure() +// never set a channel, we use a default LOW-importance channel, and no notifee +// setup is necessary. +internal fun ensureNotificationChannel(manager: NotificationManager, config: NotificationConfig) { + // minSdk is 29, so NotificationChannel (API 26) is always available. + if (manager.getNotificationChannel(config.notificationChannel) != null) return + val channel = NotificationChannel( + config.notificationChannel, + "Uploads", + NotificationManager.IMPORTANCE_LOW, + ) + manager.createNotificationChannel(channel) +} + +// builds the notification required to enable Foreground mode +internal fun buildUploadNotification( + context: Context, + config: NotificationConfig, + connectivity: Connectivity, +): Notification { + val channel = config.notificationChannel + val progress = UploadProgress.total() + val progress2Decimals = "%.2f".format(progress) + val title = when (connectivity) { + Connectivity.NoWifi -> config.notificationTitleNoWifi + Connectivity.NoInternet -> config.notificationTitleNoInternet + Connectivity.Ok -> config.notificationTitle + } + + // Custom layout for progress notification. + // The default hides the % text. This one shows it on the right, + // like most examples in various docs. + val content = RemoteViews(context.packageName, R.layout.notification) + content.setTextViewText(R.id.notification_title, title) + content.setTextViewText(R.id.notification_progress, "${progress2Decimals}%") + content.setProgressBar(R.id.notification_progress_bar, 100, progress.toInt(), false) + + return NotificationCompat.Builder(context, channel).run { + // Starting Android 12, the notification shows up with a confusing delay of 10s. + // This fixes that delay. + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) + foregroundServiceBehavior = Notification.FOREGROUND_SERVICE_IMMEDIATE + + // Required by android. Here we use the system's default upload icon + setSmallIcon(android.R.drawable.stat_sys_upload) + // These prevent the notification from being force-dismissed or dismissed when pressed + setOngoing(true) + setAutoCancel(false) + // These help show the same custom content when the notification collapses and expands + setCustomContentView(content) + setCustomBigContentView(content) + // opens the app when the notification is pressed + setContentIntent(openAppIntent(context)) + build() + } +} + +internal fun uploadForegroundInfo(config: NotificationConfig, notification: Notification): ForegroundInfo { + val id = config.systemNotificationId + // Starting Android 14, FOREGROUND_SERVICE_TYPE_DATA_SYNC is mandatory, otherwise app will crash + return if (Build.VERSION.SDK_INT > Build.VERSION_CODES.TIRAMISU) + ForegroundInfo(id, notification, ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC) + else + ForegroundInfo(id, notification) +} + +/** + * Whether [error] means that the system refused to let this worker enter + * foreground mode, because the app is in the background. Android 12 (API 31) + * restricts foreground-service starts from the background, and WorkManager + * relaunches workers exactly there (a reboot, or a quota resume). The Android + * 15 dataSync time-limit denial surfaces as the same exception. The transfer + * itself needs no foreground mode. It only loses process-priority protection. + * Thus callers continue without it. They do not fail an upload that can run. + * The function walks the causes, because setForeground can wrap the platform + * exception. + */ +internal fun isForegroundStartDenied(error: Throwable): Boolean { + if (Build.VERSION.SDK_INT < 31) return false + // ServiceStartNotAllowedException (API 31) covers the two variants: + // Foreground- and BackgroundServiceStartNotAllowedException. + return generateSequence(error) { it.cause } + .any { it is android.app.ServiceStartNotAllowedException } +} + +private fun openAppIntent(context: Context): PendingIntent? { + val intent = Intent(context, NotificationReceiver::class.java) + val flags = PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE + return PendingIntent.getBroadcast(context, "RNFileUpload-notification".hashCode(), intent, flags) +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadOutcome.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadOutcome.kt index d044ac9e..f4fb2763 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploadOutcome.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadOutcome.kt @@ -7,11 +7,26 @@ import java.io.IOException // logic in the uploader (it decides success vs failure), so it's covered directly. object UploadOutcome { - // Whether an HTTP response counts as a successful completion. 2xx always, plus - // any per-request acceptStatus codes (axios validateStatus semantics). Anything - // else — including 4xx/5xx — is a terminal http error, not a completion. - fun isAccepted(code: Int, acceptStatus: List): Boolean = - code in 200..299 || acceptStatus.contains(code) + /** + * A non-2xx response to treat as success. `bodyIncludes` narrows the rule by + * a response-body substring. This is necessary when one status has several + * meanings, and only the message shows the difference (our backend's 409). + * Gson persists it inside [Upload] and [ChunkedManifest]; see + * consumer-rules.pro. + */ + data class AcceptRule( + val status: Int, + val bodyIncludes: String? = null, + ) + + // Whether an HTTP response counts as a successful completion. A 2xx always + // counts, and a matching per-request accept rule counts. Every other response, + // 4xx and 5xx included, is a terminal http error, not a completion. + fun isAccepted(code: Int, body: String?, accept: List): Boolean = + code in 200..299 || accept.any { rule -> + rule.status == code && + (rule.bodyIncludes == null || body?.contains(rule.bodyIncludes) == true) + } // Classify a thrown error into a stable kind for the JS layer. `fileExists` // is passed in (not read here) to keep this pure; callers should default it to diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt new file mode 100644 index 00000000..cc8093ea --- /dev/null +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadTransport.kt @@ -0,0 +1,29 @@ +package ai.openspace.backgroundupload + +import kotlinx.coroutines.sync.Semaphore +import okhttp3.OkHttpClient +import java.util.concurrent.TimeUnit + +// The HTTP client and the library-wide transmission gate. The simple and +// chunked workers share them. Thus "requests transmitting at one time" means +// one thing across all uploads. + +// Max total time for a single request to complete +// This is 24hrs so plenty of time for large uploads +// Worst case is the time maxes out and the upload gets restarted. +// Not using unlimited time to prevent unexpected behaviors. +private const val REQUEST_TIMEOUT = 24L +private val REQUEST_TIMEOUT_UNIT = TimeUnit.HOURS + +// The number of requests transmitting at one time across ALL uploads, chunked +// parts included. This is the design's library-wide cap of 4. Every request, +// a simple upload or a chunked part, must pass this semaphore. Thus on +// 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) + +// Use Okhttp as it provides the most standard behaviors even though it's not coroutine friendly +internal val uploadHttpClient = OkHttpClient.Builder() + .callTimeout(REQUEST_TIMEOUT, REQUEST_TIMEOUT_UNIT) + .build() diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt index 44696541..0e48da98 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadUtils.kt @@ -4,6 +4,7 @@ import kotlinx.coroutines.suspendCancellableCoroutine import okhttp3.Call import okhttp3.Callback import okhttp3.Headers.Companion.toHeaders +import okhttp3.MediaType import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.RequestBody @@ -15,11 +16,14 @@ import okio.ForwardingSink import okio.buffer import java.io.File import java.io.IOException +import java.io.RandomAccessFile import kotlin.coroutines.resumeWithException // Throttling interval of progress reports private const val PROGRESS_INTERVAL = 500 // milliseconds +private const val RANGE_COPY_BUFFER = 64 * 1024 + data class UploadResponse( val code: Int, val body: String, @@ -32,25 +36,36 @@ suspend fun okhttpUpload( upload: Upload, file: File, onProgress: (Long) -> Unit -) = - suspendCancellableCoroutine { continuation -> - val requestBody = file.asRequestBody() - var lastProgressReport = 0L - fun throttled(): Boolean { - val now = System.currentTimeMillis() - if (now - lastProgressReport < PROGRESS_INTERVAL) return true - lastProgressReport = now - return false - } +): UploadResponse { + val request = Request.Builder() + .url(upload.url) + .headers(upload.headers.toHeaders()) + .method(upload.method, withProgressListener(file.asRequestBody(), throttled(onProgress))) + .build() + return awaitResponse(client, request) +} - val request = Request.Builder() - .url(upload.url) - .headers(upload.headers.toHeaders()) - .method(upload.method, withProgressListener(requestBody) { progress -> - if (!throttled()) onProgress(progress) - }) - .build() +/** + * PUTs one byte range of the source file: a chunked part. It streams straight + * from disk, with no temporary chunk file. The headers are the consumer's, + * unchanged. The library adds nothing, per the design's protocol-as-data rule. + */ +suspend fun okhttpUploadPart( + client: OkHttpClient, + part: ChunkedManifest.Part, + file: File, + onProgress: (Long) -> Unit +): UploadResponse { + val request = Request.Builder() + .url(part.url) + .headers(part.headers.toHeaders()) + .put(withProgressListener(rangeRequestBody(file, part.start, part.end), throttled(onProgress))) + .build() + return awaitResponse(client, request) +} +private suspend fun awaitResponse(client: OkHttpClient, request: Request): UploadResponse = + suspendCancellableCoroutine { continuation -> val call = client.newCall(request) continuation.invokeOnCancellation { call.cancel() } call.enqueue(object : Callback { @@ -61,7 +76,11 @@ suspend fun okhttpUpload( val result = response.use { res -> // close the response asap UploadResponse( res.code, - res.body?.string()?.takeIf { str -> str.isNotEmpty() } ?: res.message, + // The body, unchanged: an empty body stays empty. A substituted + // HTTP reason phrase would make accept `bodyIncludes` rules match + // text that the server never sent. iOS also reports the body + // as-is. + res.body?.string().orEmpty(), res.headers.toMultimap().mapValues { it.value.joinToString(", ") } ) } @@ -71,6 +90,47 @@ suspend fun okhttpUpload( }) } +private fun throttled(onProgress: (Long) -> Unit): (Long) -> Unit { + var lastProgressReport = 0L + return { progress -> + val now = System.currentTimeMillis() + if (now - lastProgressReport >= PROGRESS_INTERVAL) { + lastProgressReport = now + onProgress(progress) + } + } +} + +/** + * Streams the file bytes [start, end) as a request body. A RandomAccessFile + * backs it, opened fresh on every writeTo call. OkHttp can replay a body (for + * example, after a connection-level retry), and a one-shot stream would then + * send truncated data silently. + */ +private fun rangeRequestBody(file: File, start: Long, end: Long) = object : RequestBody() { + // Null, so no Content-Type is invented. The consumer's header is already on + // the request, unchanged. + override fun contentType(): MediaType? = null + + override fun contentLength() = end - start + + override fun writeTo(sink: BufferedSink) { + RandomAccessFile(file, "r").use { raf -> + raf.seek(start) + val buffer = ByteArray(RANGE_COPY_BUFFER) + var remaining = end - start + while (remaining > 0L) { + val read = raf.read(buffer, 0, minOf(remaining, buffer.size.toLong()).toInt()) + if (read < 0) throw IOException( + "source file ended before part range [$start, $end): ${file.path}", + ) + sink.write(buffer, 0, read) + remaining -= read + } + } + } +} + // create a request body that allows us to listen to progress. // okhttp has no built-in way of reporting progress private fun withProgressListener( @@ -94,4 +154,4 @@ private fun withProgressListener( body.writeTo(bufferedSink) bufferedSink.flush() } -} \ No newline at end of file +} diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploadWorker.kt b/android/src/main/java/ai/openspace/backgroundupload/UploadWorker.kt index 9ad6380b..1f0c63e8 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploadWorker.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/UploadWorker.kt @@ -1,56 +1,27 @@ package ai.openspace.backgroundupload -import android.app.Notification -import android.app.NotificationChannel import android.app.NotificationManager -import android.app.PendingIntent import android.content.Context -import android.content.Intent -import android.content.pm.ServiceInfo -import android.net.ConnectivityManager -import android.net.NetworkCapabilities.NET_CAPABILITY_VALIDATED -import android.net.NetworkCapabilities.TRANSPORT_WIFI -import android.os.Build -import android.widget.RemoteViews -import androidx.core.app.NotificationCompat import androidx.work.CoroutineWorker import androidx.work.ForegroundInfo import androidx.work.WorkerParameters import com.google.gson.Gson import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay -import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.withContext -import okhttp3.OkHttpClient import java.io.File import java.io.IOException import java.net.UnknownHostException import java.util.UUID import java.util.concurrent.TimeUnit -// All workers will start `doWork` immediately but only 1 request is active at a time. -private const val MAX_CONCURRENCY = 1 - // Retry delay private val RETRY_DELAY = TimeUnit.SECONDS.toMillis(10L) -// Max total time for a single request to complete -// This is 24hrs so plenty of time for large uploads -// Worst case is the time maxes out and the upload gets restarted. -// Not using unlimited time to prevent unexpected behaviors. -private const val REQUEST_TIMEOUT = 24L -private val REQUEST_TIMEOUT_UNIT = TimeUnit.HOURS - -// Control max concurrent requests using semaphore to instead of using -// `maxConnectionsCount` in HttpClient as the latter introduces a delay between requests -private val semaphore = Semaphore(MAX_CONCURRENCY) - -// Use Okhttp as it provides the most standard behaviors even though it's not coroutine friendly -private val client = OkHttpClient.Builder() - .callTimeout(REQUEST_TIMEOUT, REQUEST_TIMEOUT_UNIT) - .build() - -private enum class Connectivity { NoWifi, NoInternet, Ok } +// The retry budget for errors that count (see checkRetry). A connectivity gap +// or flaky-network IO resets the budget. The retry policy is internal to the +// library. It is not an option. +private const val MAX_RETRIES = 5 class UploadWorker(private val context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { @@ -69,6 +40,10 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : } private lateinit var upload: Upload + // configure() saved this. The worker can read it when WorkManager relaunched + // the worker with no JS. It is lazy, so the SharedPreferences read occurs on + // the worker's IO dispatcher, not at construction. + private val config by lazy { NotificationConfig.load(context) } private var retries = 0 private var connectivity = Connectivity.Ok private val notificationManager = @@ -79,7 +54,10 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : // However, the only way it has errors is the implementation is incorrect, // which can be caught in development val paramsJson = inputData.getString(PARAMS_KEY) ?: throw Throwable("No Params") - upload = Gson().fromJson(paramsJson, Upload::class.java) + // normalized(): an older build can have enqueued this job, and its JSON + // shape can make non-null fields null (Gson does not use the constructor). + // See Upload.normalized. + upload = Gson().fromJson(paramsJson, Upload::class.java).normalized() // initialization, errors thrown here won't be retried try { @@ -88,7 +66,7 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : if (upload.showsNotification) { // The foreground notification needs a channel to exist first, or posting // it silently fails and setForeground can crash on newer Android. - ensureNotificationChannel() + ensureNotificationChannel(notificationManager, config) // `setForeground` is recommended for long-running workers. // Foreground mode helps prioritize the worker, reducing the risk // of it being killed during low memory or Doze/App Standby situations. @@ -96,8 +74,13 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : setForeground(getForegroundInfo()) } } catch (error: Throwable) { - if (!checkAndHandleCancellation()) handleError(error) - throw error + if (!isForegroundStartDenied(error)) { + if (!checkAndHandleCancellation()) handleError(error) + throw error + } + // The app is in the background on API 31+ (see isForegroundStartDenied). + // Continue the upload without foreground priority. Do not fail an upload + // that can run. } @@ -113,11 +96,12 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : // which cancels the delay immediately and throws CancellationException. // - Linear backoff instead of exponential. One reason for this is we retry on // invalid connections. Exponential will take too long. - // - We only retry transport failures here (no response). Any HTTP response, - // including 4xx/5xx, is terminal at this layer: handleResponse classifies it - // (2xx/acceptStatus -> completed, else http error) and the worker returns - // without retrying. Response-code-based retry policy is the JS queue's job. - // This is consistent with iOS behavior. + // - We retry only transport failures here (no response). An HTTP + // response, 4xx and 5xx included, is terminal at this layer. + // handleResponse classifies it (a 2xx or an accept rule -> completed, + // else an http error), and the worker returns without a retry. A + // response-code retry policy is the JS queue's job. This matches the + // iOS behavior. if (isRetried) delay(RETRY_DELAY) isRetried = true @@ -148,10 +132,10 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : if (!validateAndReportConnectivity()) return null // wait for its turn to run - semaphore.acquire() + transferSemaphore.acquire() try { - return okhttpUpload(client, upload, file) { progress -> + return okhttpUpload(uploadHttpClient, upload, file) { progress -> handleProgress(progress, size) } } catch (error: Throwable) { @@ -160,7 +144,7 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : // pass the error to upper layer for retry decision throw error } finally { - semaphore.release() + transferSemaphore.release() } } @@ -174,16 +158,20 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : // worker never posted one, and `notify` would create it outside foreground mode. private fun updateNotification() { if (!upload.showsNotification) return - notificationManager.notify(upload.notificationId, buildNotification()) + notificationManager.notify( + config.systemNotificationId, + buildUploadNotification(context, config, connectivity), + ) } - // An HTTP response came back. "completed" only for 2xx or a per-request - // acceptStatus code (axios validateStatus semantics — a 400 is an error, not a - // completion); anything else is a terminal http error carrying the full - // response. Either way the request finished, so the worker does not retry. + // An HTTP response came back. It is "completed" only for a 2xx or a matching + // accept rule (axios validateStatus semantics: a 400 is an error, not a + // completion). Every other response is a terminal http error that carries the + // full response. In both cases the request finished, so the worker does not + // retry. private fun handleResponse(response: UploadResponse) { UploadProgress.complete(upload.id) - val accepted = UploadOutcome.isAccepted(response.code, upload.acceptStatus) + val accepted = UploadOutcome.isAccepted(response.code, response.body, upload.accept) val (body, truncated) = EventJournal.capBody(response.body) journalAndEmit( EventJournal.Entry( @@ -252,13 +240,8 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : return true } - // Journal before emitting: the journal is the durable record (survives JS being - // dead); the live emit is best-effort. Both carry the identical payload, so a - // consumer can ack a live event by its eventId. - private fun journalAndEmit(entry: EventJournal.Entry) { - EventJournal.get(context).append(entry) - EventReporter.emit(entry) - } + private fun journalAndEmit(entry: EventJournal.Entry) = + EventReporter.journalAndEmit(context, entry) /** @return whether to retry */ private fun checkRetry(error: Throwable): Boolean { @@ -287,7 +270,7 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : } retries = if (unlimitedRetry) 0 else retries + 1 - return retries <= upload.maxRetries + return retries <= MAX_RETRIES } // Checks connection and alerts connection issues @@ -298,93 +281,6 @@ class UploadWorker(private val context: Context, params: WorkerParameters) : return this.connectivity == Connectivity.Ok } - // Ensures the channel used by the foreground notification exists. Only creates - // it when absent, so a channel the consumer registered themselves (with their - // own name/importance) always wins; when they pass nothing we fall back to a - // default LOW-importance channel and no notifee setup is required. - private fun ensureNotificationChannel() { - // minSdk is 29, so NotificationChannel (API 26) is always available. - if (notificationManager.getNotificationChannel(upload.notificationChannel) != null) return - val channel = NotificationChannel( - upload.notificationChannel, - "Uploads", - NotificationManager.IMPORTANCE_LOW, - ) - notificationManager.createNotificationChannel(channel) - } - - // builds the notification required to enable Foreground mode - fun buildNotification(): Notification { - val channel = upload.notificationChannel - val progress = UploadProgress.total() - val progress2Decimals = "%.2f".format(progress) - val title = when (connectivity) { - Connectivity.NoWifi -> upload.notificationTitleNoWifi - Connectivity.NoInternet -> upload.notificationTitleNoInternet - Connectivity.Ok -> upload.notificationTitle - } - - // Custom layout for progress notification. - // The default hides the % text. This one shows it on the right, - // like most examples in various docs. - val content = RemoteViews(context.packageName, R.layout.notification) - content.setTextViewText(R.id.notification_title, title) - content.setTextViewText(R.id.notification_progress, "${progress2Decimals}%") - content.setProgressBar(R.id.notification_progress_bar, 100, progress.toInt(), false) - - return NotificationCompat.Builder(context, channel).run { - // Starting Android 12, the notification shows up with a confusing delay of 10s. - // This fixes that delay. - if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) - foregroundServiceBehavior = Notification.FOREGROUND_SERVICE_IMMEDIATE - - // Required by android. Here we use the system's default upload icon - setSmallIcon(android.R.drawable.stat_sys_upload) - // These prevent the notification from being force-dismissed or dismissed when pressed - setOngoing(true) - setAutoCancel(false) - // These help show the same custom content when the notification collapses and expands - setCustomContentView(content) - setCustomBigContentView(content) - // opens the app when the notification is pressed - setContentIntent(openAppIntent(context)) - build() - } - } - - override suspend fun getForegroundInfo(): ForegroundInfo { - val notification = buildNotification() - val id = upload.notificationId - // Starting Android 14, FOREGROUND_SERVICE_TYPE_DATA_SYNC is mandatory, otherwise app will crash - return if (Build.VERSION.SDK_INT > Build.VERSION_CODES.TIRAMISU) - ForegroundInfo(id, notification, ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC) - else - ForegroundInfo(id, notification) - } -} - -// This is outside and synchronized to ensure consistent status across workers -@Synchronized -private fun validateConnectivity(context: Context, wifiOnly: Boolean): Connectivity { - val manager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager - val network = manager.activeNetwork - val capabilities = manager.getNetworkCapabilities(network) - - val hasInternet = capabilities?.hasCapability(NET_CAPABILITY_VALIDATED) == true - - // not wifiOnly, return early - if (!wifiOnly) return if (hasInternet) Connectivity.Ok else Connectivity.NoInternet - - // handle wifiOnly - return if (hasInternet && capabilities?.hasTransport(TRANSPORT_WIFI) == true) - Connectivity.Ok - else - Connectivity.NoWifi // don't return NoInternet here, more direct to request to join wifi -} - - -private fun openAppIntent(context: Context): PendingIntent? { - val intent = Intent(context, NotificationReceiver::class.java) - val flags = PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE - return PendingIntent.getBroadcast(context, "RNFileUpload-notification".hashCode(), intent, flags) + override suspend fun getForegroundInfo(): ForegroundInfo = + uploadForegroundInfo(config, buildUploadNotification(context, config, connectivity)) } diff --git a/android/src/main/java/ai/openspace/backgroundupload/UploaderModule.kt b/android/src/main/java/ai/openspace/backgroundupload/UploaderModule.kt index aed7f4e3..206253d7 100644 --- a/android/src/main/java/ai/openspace/backgroundupload/UploaderModule.kt +++ b/android/src/main/java/ai/openspace/backgroundupload/UploaderModule.kt @@ -13,6 +13,9 @@ import com.facebook.react.bridge.ReadableArray import com.facebook.react.bridge.ReadableMap import com.facebook.react.bridge.WritableMap import com.google.gson.Gson +import java.io.File +import java.nio.file.Files +import java.nio.file.StandardCopyOption import java.util.UUID @@ -114,7 +117,19 @@ class UploaderModule(context: ReactApplicationContext) : override fun ackEvents(ids: ReadableArray, promise: Promise) { try { val eventIds = (0 until ids.size()).mapNotNull { ids.getString(it) } - EventJournal.get(reactApplicationContext).ack(eventIds) + val journal = EventJournal.get(reactApplicationContext) + // An acknowledged 'completed' is the ONE moment when a chunked upload's + // manifest and moved bytes may be deleted. Every other terminal keeps + // them for a resume. Resolve which uploads those are before the entries + // are removed. + val completedUploadIds = journal.unacknowledged() + .filter { it.type == "completed" && eventIds.contains(it.eventId) } + .map { it.uploadId } + journal.ack(eventIds) + releaseAckedCompletions( + completedUploadIds, + ChunkedManifestStore.get(reactApplicationContext), + ) { id -> workManager.cancelUniqueWork(id) } promise.resolve(true) } catch (exc: Throwable) { Log.e(TAG, exc.message, exc) @@ -124,30 +139,51 @@ class UploaderModule(context: ReactApplicationContext) : /** - * Enumerates uploads WorkManager still knows about, as [{ id, state }]. - * WorkManager auto-prunes finished work after roughly a day, so this is for - * reconciling live/recent uploads — terminal outcomes must be read from - * getUnacknowledgedEvents, which is durable until acknowledged. + * Enumerates the uploads that WorkManager still knows about, as + * [{ id, state }]. Chunked uploads also carry the aggregate + * { bytesSent, totalBytes }, and they are listed from their durable + * manifests, even after WorkManager prunes finished work (in roughly a day). + * Terminal outcomes must be read from getUnacknowledgedEvents, which is + * durable until acknowledged. */ override fun getAllUploads(promise: Promise) { try { - val infos = workManager.getWorkInfosByTag(WORKER_TAG).get() + val manifests = ChunkedManifestStore.get(reactApplicationContext).all() + .associateBy { it.id } + // Several WorkInfo rows can exist for one upload id. Finished chains + // linger until they are pruned (in roughly a day), and APPEND_OR_REPLACE + // resumes add rows. Thus the rows are grouped, and each id gets exactly + // ONE entry, like iOS (one row per upload, with the same state + // vocabulary). + val statesById = workManager.getWorkInfosByTag(WORKER_TAG).get() + .groupBy( + { info -> + info.tags.firstOrNull { it.startsWith(ID_TAG_PREFIX) } + ?.removePrefix(ID_TAG_PREFIX) + }, + { it.state }, + ) val arr = Arguments.createArray() - for (info in infos) { - val id = info.tags.firstOrNull { it.startsWith(ID_TAG_PREFIX) } - ?.removePrefix(ID_TAG_PREFIX) ?: continue + for ((id, states) in statesById) { + if (id == null || id in manifests) continue arr.pushMap(Arguments.createMap().apply { putString("id", id) + putString("state", simpleUploadState(states)) + }) + } + // Chunked uploads are listed from their durable manifests, which outlive + // the WorkManager rows. Only a live row contributes (running or pending). + // A lingering finished row never speaks for a manifest that is really + // "finished, awaiting its ack" or "stalled, awaiting a startUpload + // resume". + for (manifest in manifests.values) { + arr.pushMap(Arguments.createMap().apply { + putString("id", manifest.id) putString( "state", - when (info.state) { - WorkInfo.State.ENQUEUED, WorkInfo.State.BLOCKED -> "pending" - WorkInfo.State.RUNNING -> "running" - WorkInfo.State.SUCCEEDED -> "completed" - WorkInfo.State.FAILED -> "error" - WorkInfo.State.CANCELLED -> "cancelled" - }, + chunkedUploadState(statesById[manifest.id].orEmpty(), manifest.allAccepted), ) + putChunkedBytes(manifest) }) } promise.resolve(arr) @@ -159,12 +195,13 @@ class UploaderModule(context: ReactApplicationContext) : /** - * iOS-only: there is no per-task byte counter to read on Android, where uploads - * are WorkManager jobs rather than URLSession tasks. Use getAllUploads for - * liveness and the progress event for bytes. + * Saves the notification configuration (see [NotificationConfig]). Thus a + * worker that WorkManager relaunches with no JS can read it. Each call + * replaces the full configuration. An omitted field goes back to the library + * default. */ - override fun getUploadStatus(id: String, promise: Promise) { - promise.resolve(null) + override fun configure(options: ReadableMap) { + NotificationConfig.save(reactApplicationContext, NotificationConfig.fromReadableMap(options)) } @@ -192,7 +229,7 @@ class UploaderModule(context: ReactApplicationContext) : val upload = Upload.fromReadableMap(options) val data = Gson().toJson(upload) - // Clear any stale user-cancel mark for this (possibly reused customUploadId) + // Clear any stale user-cancel mark for this (possibly reused) id // from a prior life, so a later system stop of this fresh upload isn't // misreported as a user cancel. Done here (before enqueue), never in the // worker, so a real cancel arriving as the worker starts can't be erased. @@ -216,6 +253,138 @@ class UploaderModule(context: ReactApplicationContext) : } + /** + * Starts, or resumes, a chunked upload. It is idempotent against the durable + * [ChunkedManifest]. A first call takes ownership of the source file (an + * O(1) rename into the library's directory) and persists the manifest. A + * re-call with the same id reconciles instead: identical parts are required, + * the stored headers are replaced, and the accepted parts are skipped. Crash + * recovery, a resume after a stop, and a resume with fresh auth are all this + * same call. + */ + override fun startChunkedUpload(options: ReadableMap, promise: Promise) { + try { + promise.resolve(enqueueChunkedUpload(options)) + } catch (exc: Throwable) { + if (exc !is IllegalArgumentException) { + exc.printStackTrace() + Log.e(TAG, exc.message, exc) + } + promise.reject(exc) + } + } + + private fun enqueueChunkedUpload(options: ReadableMap): String { + val store = ChunkedManifestStore.get(reactApplicationContext) + val id = options.getString("id") + ?: throw Upload.MissingOptionException("id") + val blob = store.blobFile(id) + val incoming = ChunkedManifest.fromReadableMap( + options, + sourcePath = blob.absolutePath, + createdAt = System.currentTimeMillis(), + ) + + // One atomic store operation, persisted BEFORE the work is enqueued. The + // manifest is what a worker relaunched with no JS runs from. The store + // lock spans load, reconcile, and save. Thus a running worker's + // markAccepted can never land between them and be erased. The running flag + // inside the lock is race-free too. A worker acquires ChunkedWorkerGate + // before its first manifest read. Thus it either registers first (and the + // recreate is rejected), or it reads the manifest that this call saved. + store.compute(id) { existing -> + if (existing == null) { + val path = options.getString("path") ?: throw Upload.MissingOptionException("path") + takeOwnership(File(path), blob) + incoming + } else { + // `path` is deliberately ignored here. When a manifest exists, the + // owned bytes are the source of truth. + existing.reconcile( + incoming, + running = ChunkedWorkerGate.isRunning(id), + blobSize = File(existing.sourcePath).length(), + ) + } + } + + // The stale-mark reasoning is the same as in enqueueUpload. + UserCancellations.consume(id) + + // A queued successor (an unfinished row that is not RUNNING) already + // guarantees a run after the current one finishes. An appended second run + // would only stack duplicate no-op runs. The manifest reconcile above + // still landed. That is how this call's fresh headers reach the queued + // run. + val states = workManager.getWorkInfosForUniqueWork(id).get().map { it.state } + if (hasQueuedSuccessor(states)) return id + + val request = OneTimeWorkRequestBuilder() + .addTag(WORKER_TAG) + .addTag(ID_TAG_PREFIX + id) + .setInputData(workDataOf(ChunkedUploadWorker.ID_KEY to id)) + .build() + + // APPEND_OR_REPLACE, not KEEP. A worker journals its terminal error before + // doWork returns. Thus a consumer that resumes from the error handler can + // arrive while that run's row is still RUNNING. KEEP would silently drop + // the resume, and nothing would ever run it. An append keeps the runs + // strictly sequential, and a trailing run over an already-settled manifest + // is a clean no-op (see ChunkedEngine.startAction). Appended work is a + // chain DEPENDENT: WorkManager marks the dependents of a failed + // prerequisite FAILED without a run. That is why ChunkedUploadWorker + // always returns Result.success(), even after it journals a terminal error + // (see terminalErrorResult). The OR_REPLACE half only rescues enqueues + // that arrive AFTER the chain already settled failed or cancelled: it + // starts a fresh sequence. A re-call while the worker runs still never + // restarts it. The running worker re-reads the stored manifest before + // every part attempt, so a resume's fresh headers reach it. + workManager + .beginUniqueWork(id, ExistingWorkPolicy.APPEND_OR_REPLACE, request) + .enqueue() + + return id + } + + private fun takeOwnership(source: File, blob: File) { + if (!source.exists()) { + // A crash between the rename and the manifest save leaves the bytes at + // the blob path with no manifest. Adopt them. Do not fail the retry. + if (blob.exists()) return + throw IllegalArgumentException("chunked source file does not exist: ${source.path}") + } + blob.parentFile?.mkdirs() + if (blob.exists()) blob.delete() + if (source.renameTo(blob)) return + // renameTo cannot cross filesystems. Files.move falls back to copy+delete. + Files.move(source.toPath(), blob.toPath(), StandardCopyOption.REPLACE_EXISTING) + } + + + /** + * Releases an upload's stored state. It cancels the scheduled or running + * work, then deletes the chunked manifest and the moved bytes. It is safe on + * any id. A simple upload has nothing stored, so the call reduces to the + * work cancel. There is deliberately no 'cancelled' event. This is an + * explicit release by the consumer, not an outcome that the consumer awaits. + */ + override fun removeUpload(id: String, promise: Promise) { + try { + workManager.cancelUniqueWork(id) + // No user-cancel mark was set, so a running worker's stop handler + // reports nothing. The consume call clears a stale mark from a prior + // life. + UserCancellations.consume(id) + ChunkedManifestStore.get(reactApplicationContext).remove(id) + promise.resolve(null) + } catch (exc: Throwable) { + exc.printStackTrace() + Log.e(TAG, exc.message, exc) + promise.reject(exc) + } + } + + /* * Cancels file upload * Accepts upload ID as a first argument, this upload will be cancelled @@ -223,38 +392,42 @@ class UploaderModule(context: ReactApplicationContext) : */ override fun cancelUpload(id: String, promise: Promise) { try { - val active = workManager.getWorkInfosForUniqueWork(id).get() - .firstOrNull { !it.state.isFinished } + val activeStates = workManager.getWorkInfosForUniqueWork(id).get() + .map { it.state } + .filter { !it.isFinished } - if (active == null) { - // Nothing to cancel. Drop any mark so a later upload reusing this - // customUploadId can't be misreported as a user cancel. + if (activeStates.isEmpty()) { + // Nothing to cancel. Drop any mark so a later upload reusing this id + // can't be misreported as a user cancel. UserCancellations.consume(id) promise.resolve(false) return } - // Record intent BEFORE cancelling so the worker's stop handler can tell - // this apart from a system stop and report cancelReason 'user'. + // Record the intent BEFORE the cancel. Then a running worker's stop + // handler can tell this apart from a system stop, and it reports + // cancelReason 'user'. UserCancellations.mark(id) workManager.cancelUniqueWork(id) - if (active.state == WorkInfo.State.ENQUEUED) { - // The worker never started, so it will never run its own stop handler and - // nothing else would ever report this cancellation — leaving a consumer - // awaiting this upload's outcome forever. Report it here instead, and - // consume the mark so it cannot leak. + if (cancelReportsFromModule(activeStates)) { + // No worker ever started: the rows are only ENQUEUED, or BLOCKED + // behind an appended chain. Thus no stop handler will ever run, and + // nothing else would ever report this cancellation. A consumer would + // then await this upload's outcome forever. Report it here instead, + // and consume the mark so it cannot leak. UserCancellations.consume(id) - val entry = EventJournal.Entry( - eventId = UUID.randomUUID().toString(), - uploadId = id, - type = "cancelled", - timestamp = System.currentTimeMillis(), - cancelReason = "user", + EventReporter.journalAndEmit( + reactApplicationContext, + EventJournal.Entry( + eventId = UUID.randomUUID().toString(), + uploadId = id, + type = "cancelled", + timestamp = System.currentTimeMillis(), + cancelReason = "user", + ), ) - EventJournal.get(reactApplicationContext).append(entry) - EventReporter.emit(entry) } promise.resolve(true) @@ -265,3 +438,81 @@ class UploaderModule(context: ReactApplicationContext) : } } } + +/** + * Releases the uploads whose 'completed' events were just acknowledged. That + * is the ONE moment when a chunked upload's manifest and moved bytes may be + * deleted. The allAccepted guard protects a recreate: the id may have been + * RECREATED (a different parts array under the same id) and run again over + * these bytes. An ack of the old life's completion must not cancel that work, + * and it must not delete the blob under it. Simple uploads have no manifest + * and fall through untouched. A cancel of their unique work could kill an + * unrelated new upload that reuses the id. + */ +internal fun releaseAckedCompletions( + uploadIds: List, + store: ChunkedManifestStore, + cancelWork: (String) -> Unit, +) { + uploadIds.forEach { id -> + val manifest = store.load(id) ?: return@forEach + if (!manifest.allAccepted) return@forEach + // Cancel a still-enqueued trailing run BEFORE the delete. A worker that + // starts after the delete finds nothing. It exits silently, but there is + // no reason to run it at all. + cancelWork(id) + store.remove(id) + } +} + +/** + * Whether cancelUpload must journal and emit the 'cancelled' event itself. + * That is the case only when NO row is RUNNING. A never-started row (ENQUEUED, + * or BLOCKED as an appended chain's dependent) has no worker to run a stop + * handler. A RUNNING worker's stop handler owns the report, including a worker + * that still waits on the ChunkedWorkerGate. + */ +internal fun cancelReportsFromModule(unfinishedStates: List): Boolean = + unfinishedStates.isNotEmpty() && unfinishedStates.none { it == WorkInfo.State.RUNNING } + +/** An unfinished row that is not RUNNING: a queued run that did not start yet. */ +internal fun hasQueuedSuccessor(states: List): Boolean = + states.any { !it.isFinished && it != WorkInfo.State.RUNNING } + +// The aggregate byte fields for a chunked upload's snapshot. bytesSent counts +// accepted parts only. That is the durable number, and it has meaning even in +// a process where the worker does not run. +private fun WritableMap.putChunkedBytes(manifest: ChunkedManifest) { + putDouble("bytesSent", manifest.acceptedBytes.toDouble()) + putDouble("totalBytes", manifest.totalBytes.toDouble()) +} + +/** + * One state for a chunked upload id, from all its WorkInfo rows plus the + * durable manifest. A live row wins. With no live row, the manifest speaks. + * The state is never "cancelled". iOS getAllUploads has no lingering cancelled + * rows (a cancelled task leaves the session). And on Android, a cancelled + * chunked upload keeps its manifest. Its truthful state is + * stalled-awaiting-resume, that is, "error". + */ +internal fun chunkedUploadState(states: List, allAccepted: Boolean): String = + when { + WorkInfo.State.RUNNING in states -> "running" + states.any { !it.isFinished } -> "pending" + allAccepted -> "completed" + else -> "error" + } + +/** + * One state for a simple upload id, from all its WorkInfo rows. An id can have + * a lingering finished chain next to a live one. A live row wins. Otherwise + * the most conclusive finished state wins. + */ +internal fun simpleUploadState(states: List): String = + when { + WorkInfo.State.RUNNING in states -> "running" + states.any { !it.isFinished } -> "pending" + WorkInfo.State.SUCCEEDED in states -> "completed" + WorkInfo.State.FAILED in states -> "error" + else -> "cancelled" + } diff --git a/android/src/test/java/ai/openspace/backgroundupload/AckReleaseTest.kt b/android/src/test/java/ai/openspace/backgroundupload/AckReleaseTest.kt new file mode 100644 index 00000000..0afe1e88 --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/AckReleaseTest.kt @@ -0,0 +1,69 @@ +package ai.openspace.backgroundupload + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder + +// An acknowledged 'completed' is the one moment when a chunked upload's stored +// state may be released. But only the completed life's state may go. A recreate +// under the same id can run over the same bytes, and it must survive the old +// life's ack. +class AckReleaseTest { + @get:Rule + val tmp = TemporaryFolder() + + private fun manifest(id: String, accepted: Boolean) = ChunkedManifest( + id = id, + sourcePath = "/data/blob", + parts = listOf( + ChunkedManifest.Part( + url = "https://example.com/1", + headers = emptyMap(), + start = 0, + end = 100, + accepted = accepted, + ), + ), + accept = emptyList(), + expiresAt = 5_000, + wifiOnly = false, + noNotification = false, + createdAt = 1_000, + ) + + @Test + fun `releases a completed upload's manifest and cancels its trailing runs`() { + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest("u1", accepted = true)) + val cancelled = mutableListOf() + releaseAckedCompletions(listOf("u1"), store) { cancelled.add(it) } + assertNull(store.load("u1")) + assertEquals(listOf("u1"), cancelled) + } + + @Test + fun `spares a recreate running under the same id`() { + // The acknowledged completion belongs to the id's PREVIOUS life. The + // manifest now holds a recreate's unaccepted parts, and a worker can be + // mid-transfer. A work cancel or a blob delete here would destroy its + // bytes. + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest("u1", accepted = false)) + val cancelled = mutableListOf() + releaseAckedCompletions(listOf("u1"), store) { cancelled.add(it) } + assertNotNull(store.load("u1")) + assertTrue(cancelled.isEmpty()) + } + + @Test + fun `ignores ids with no manifest (simple uploads)`() { + val store = ChunkedManifestStore(tmp.newFolder()) + val cancelled = mutableListOf() + releaseAckedCompletions(listOf("raw-upload"), store) { cancelled.add(it) } + assertTrue(cancelled.isEmpty()) + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/ChunkedEngineTest.kt b/android/src/test/java/ai/openspace/backgroundupload/ChunkedEngineTest.kt new file mode 100644 index 00000000..8c909d62 --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/ChunkedEngineTest.kt @@ -0,0 +1,181 @@ +package ai.openspace.backgroundupload + +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.yield +import org.junit.Assert.assertEquals +import org.junit.Assert.assertThrows +import org.junit.Assert.assertTrue +import org.junit.Test +import java.util.concurrent.ConcurrentHashMap + +class ChunkedEngineTest { + + // runBlocking is single-threaded, so the overlap is deterministic. Every + // executor suspends at yield(). Thus all launchable siblings start before any + // executor finishes. + private class Tracker { + var inFlight = 0 + var maxInFlight = 0 + val executions = ConcurrentHashMap() + + suspend fun execute(index: Int) { + executions.merge(index, 1, Int::plus) + inFlight++ + maxInFlight = maxOf(maxInFlight, inFlight) + yield() + yield() + inFlight-- + } + } + + @Test + fun `runs every part exactly once`() = runBlocking { + val tracker = Tracker() + ChunkedEngine.run((0 until 10).toList()) { tracker.execute(it) } + assertEquals((0 until 10).associateWith { 1 }, tracker.executions.toMap()) + } + + @Test + fun `never more than WINDOW parts in flight`() = runBlocking { + val tracker = Tracker() + ChunkedEngine.run((0 until 10).toList()) { tracker.execute(it) } + assertEquals(ChunkedEngine.WINDOW, tracker.maxInFlight) + } + + @Test + fun `respects a smaller window`() = runBlocking { + val tracker = Tracker() + ChunkedEngine.run((0 until 5).toList(), window = 1) { tracker.execute(it) } + assertEquals(1, tracker.maxInFlight) + } + + @Test + fun `an empty part list completes immediately`() = runBlocking { + ChunkedEngine.run(emptyList()) { throw AssertionError("must not execute") } + } + + @Test + fun `a terminal part failure propagates and cancels the remaining parts`() { + val tracker = Tracker() + val thrown = assertThrows(IllegalStateException::class.java) { + runBlocking { + ChunkedEngine.run((0 until 10).toList()) { index -> + if (index == 0) throw IllegalStateException("part rejected") + tracker.execute(index) + } + } + } + assertEquals("part rejected", thrown.message) + assertTrue(tracker.executions.size < 10) + } + + @Test + fun `backoff grows exponentially and caps`() { + assertEquals(1_000, ChunkedEngine.backoffMs(1)) + assertEquals(2_000, ChunkedEngine.backoffMs(2)) + assertEquals(4_000, ChunkedEngine.backoffMs(3)) + assertEquals(60_000, ChunkedEngine.backoffMs(7)) + assertEquals(60_000, ChunkedEngine.backoffMs(100)) + // Defensive: a nonsense attempt number must not shift into a huge delay. + assertEquals(1_000, ChunkedEngine.backoffMs(0)) + } + + // MARK: - startAction + + private fun manifest(vararg accepted: Boolean) = ChunkedManifest( + id = "u1", + sourcePath = "/data/blob", + parts = accepted.mapIndexed { i, a -> + ChunkedManifest.Part( + url = "https://example.com/part?n=$i", + headers = emptyMap(), + start = i * 100L, + end = (i + 1) * 100L, + accepted = a, + ) + }, + accept = emptyList(), + expiresAt = 5_000, + wifiOnly = false, + noNotification = false, + createdAt = 1_000, + ) + + @Test + fun `no manifest at start is a silent success, never a journaled error`() { + // A completed ack or removeUpload deleted the manifest while this run sat + // in the queue. That is a legitimate end, already settled. + assertEquals(ChunkedEngine.StartAction.NO_MANIFEST, ChunkedEngine.startAction(null)) + } + + @Test + fun `an all-accepted manifest re-reports completion instead of running`() { + assertEquals( + ChunkedEngine.StartAction.ALREADY_COMPLETE, + ChunkedEngine.startAction(manifest(true, true)), + ) + } + + @Test + fun `pending parts run the engine`() { + assertEquals(ChunkedEngine.StartAction.RUN, ChunkedEngine.startAction(manifest(true, false))) + } + + // MARK: - completionReport + + private fun completedEntry(uploadId: String) = EventJournal.Entry( + eventId = "e-$uploadId", + uploadId = uploadId, + type = "completed", + timestamp = 1, + ) + + @Test + fun `an unacked completed entry is re-emitted, never minted twice`() { + assertEquals( + ChunkedEngine.CompletionReport.ReEmit(completedEntry("u1")), + ChunkedEngine.completionReport(listOf(completedEntry("u1")), "u1", freshCompletion = false), + ) + assertEquals( + ChunkedEngine.CompletionReport.ReEmit(completedEntry("u1")), + ChunkedEngine.completionReport(listOf(completedEntry("u1")), "u1", freshCompletion = true), + ) + } + + @Test + fun `a fresh completion with nothing journaled mints a new entry`() { + assertEquals( + ChunkedEngine.CompletionReport.Mint, + ChunkedEngine.completionReport(emptyList(), "u1", freshCompletion = true), + ) + } + + @Test + fun `a trailing run over an acked completion reports nothing`() { + // The trailing run raced ackEvents. The journal entry is already gone, but + // the manifest still exists for a moment. An acknowledged completion means + // that nobody is owed an event. A minted event would be a duplicate + // 'completed' for an upload that the consumer already settled. + assertEquals( + ChunkedEngine.CompletionReport.None, + ChunkedEngine.completionReport(emptyList(), "u1", freshCompletion = false), + ) + } + + @Test + fun `another upload's completed entry does not satisfy the lookup`() { + assertEquals( + ChunkedEngine.CompletionReport.None, + ChunkedEngine.completionReport(listOf(completedEntry("other")), "u1", freshCompletion = false), + ) + } + + @Test + fun `only 5xx responses are transient`() { + assertTrue(ChunkedEngine.isTransientHttp(500)) + assertTrue(ChunkedEngine.isTransientHttp(599)) + for (code in listOf(400, 401, 403, 404, 409, 429, 499, 600)) { + assertEquals("code $code", false, ChunkedEngine.isTransientHttp(code)) + } + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/ChunkedManifestTest.kt b/android/src/test/java/ai/openspace/backgroundupload/ChunkedManifestTest.kt new file mode 100644 index 00000000..798852ef --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/ChunkedManifestTest.kt @@ -0,0 +1,384 @@ +package ai.openspace.backgroundupload + +import ai.openspace.backgroundupload.UploadOutcome.AcceptRule +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.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import java.io.File +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit + +class ChunkedManifestTest { + @get:Rule + val tmp = TemporaryFolder() + + private fun manifest( + id: String = "u1", + parts: List = listOf( + part(0, 100), + part(100, 250), + ), + expiresAt: Long = 5_000, + ) = ChunkedManifest( + id = id, + sourcePath = "/data/blob", + parts = parts, + accept = listOf(AcceptRule(409, "already completed")), + expiresAt = expiresAt, + wifiOnly = false, + noNotification = false, + createdAt = 1_000, + ) + + private fun part(start: Long, end: Long, accepted: Boolean = false) = + ChunkedManifest.Part( + url = "https://example.com/part?start=$start", + headers = mapOf("Authorization" to "Bearer old"), + start = start, + end = end, + accepted = accepted, + ) + + // MARK: - Model + + @Test + fun `byte math is range-based`() { + val m = manifest(parts = listOf(part(0, 100, accepted = true), part(100, 250))) + assertEquals(250, m.totalBytes) + assertEquals(100, m.acceptedBytes) + } + + @Test + fun `completed only when every part is accepted`() { + val none = manifest() + assertFalse(none.allAccepted) + val partial = none.withPartAccepted(0) + assertFalse(partial.allAccepted) + val all = partial.withPartAccepted(1) + assertTrue(all.allAccepted) + assertEquals(emptyList(), all.pendingIndexes()) + assertEquals(listOf(1), partial.pendingIndexes()) + } + + @Test + fun `expiry is inclusive of the deadline`() { + val m = manifest(expiresAt = 5_000) + assertFalse(m.isExpired(4_999)) + assertTrue(m.isExpired(5_000)) + assertTrue(m.isExpired(5_001)) + } + + // MARK: - Reconcile: resume (same parts array) + + private val blobSize = 250L + + @Test + fun `resume replaces headers and deadline, keeps accepted parts and the moved source`() { + val stored = manifest().withPartAccepted(0) + val fresh = manifest(expiresAt = 99_000).copy( + sourcePath = "/ignored/by/reconcile", + createdAt = 42, + accept = listOf(AcceptRule(208)), + wifiOnly = true, + parts = stored.parts.map { it.copy(headers = mapOf("Authorization" to "Bearer new"), accepted = false) }, + ) + + val merged = stored.reconcile(fresh, running = false, blobSize = blobSize) + + assertEquals("Bearer new", merged.parts[0].headers["Authorization"]) + assertEquals(99_000, merged.expiresAt) + assertEquals(listOf(AcceptRule(208)), merged.accept) + assertTrue(merged.wifiOnly) + // Accepted statuses, ownership, and identity survive from the stored copy. + assertTrue(merged.parts[0].accepted) + assertFalse(merged.parts[1].accepted) + assertEquals("/data/blob", merged.sourcePath) + assertEquals(1_000, merged.createdAt) + } + + @Test + fun `resume is allowed while the upload is running`() { + // Fresh auth must reach a running worker's stalled parts. + val stored = manifest().withPartAccepted(0) + val fresh = manifest().copy( + parts = stored.parts.map { it.copy(headers = mapOf("Authorization" to "Bearer new"), accepted = false) }, + ) + val merged = stored.reconcile(fresh, running = true, blobSize = blobSize) + assertTrue(merged.parts[0].accepted) + assertEquals("Bearer new", merged.parts[1].headers["Authorization"]) + } + + @Test + fun `resume matches the same parts authored in a different order`() { + // Identical tiles, reordered, are the SAME upload: a resume, never a + // recreate (running = true would reject a recreate). Accepted flags follow + // the range, not the array index. + val stored = manifest().withPartAccepted(0) + val fresh = manifest().copy( + parts = listOf(stored.parts[1], stored.parts[0]).map { + it.copy(headers = mapOf("Authorization" to "Bearer new"), accepted = false) + }, + ) + val merged = stored.reconcile(fresh, running = true, blobSize = blobSize) + assertTrue(merged.parts.first { it.start == 0L }.accepted) + assertFalse(merged.parts.first { it.start == 100L }.accepted) + assertEquals("Bearer new", merged.parts[0].headers["Authorization"]) + } + + // MARK: - Reconcile: recreate (different parts array) + + @Test + fun `recreate from a stalled upload replaces parts and resets every status`() { + // The consumer re-authored under a fresh server uploadId: new urls, a new + // split, and fresh headers, accept, and expiresAt. The owned bytes stay. + val stored = manifest().withPartAccepted(0) + val fresh = manifest( + parts = listOf( + part(0, 120).copy(url = "https://example.com/v2?part=1"), + part(120, 250).copy(url = "https://example.com/v2?part=2"), + ), + expiresAt = 99_000, + ).copy(sourcePath = "/ignored/by/reconcile", createdAt = 42, accept = listOf(AcceptRule(208))) + + val recreated = stored.reconcile(fresh, running = false, blobSize = blobSize) + + assertTrue(recreated.parts.none { it.accepted }) + assertEquals(listOf("https://example.com/v2?part=1", "https://example.com/v2?part=2"), recreated.parts.map { it.url }) + assertEquals(99_000, recreated.expiresAt) + assertEquals(listOf(AcceptRule(208)), recreated.accept) + // Ownership survives. The blob is reused for the full re-upload. + assertEquals("/data/blob", recreated.sourcePath) + assertEquals(1_000, recreated.createdAt) + } + + @Test + fun `recreate with the same ranges but new urls also resets statuses`() { + // New part urls embed a new server uploadId, even when the split is + // identical. Nothing sent under the old id counts for the new one. + val stored = manifest().withPartAccepted(0) + val fresh = manifest( + parts = listOf( + part(0, 100).copy(url = "https://example.com/v2?part=1"), + part(100, 250).copy(url = "https://example.com/v2?part=2"), + ), + ) + val recreated = stored.reconcile(fresh, running = false, blobSize = blobSize) + assertTrue(recreated.parts.none { it.accepted }) + } + + @Test + fun `recreate is rejected while the upload is running`() { + val stored = manifest() + val fresh = manifest(parts = listOf(part(0, 250).copy(url = "https://example.com/v2"))) + assertThrows(ChunkedManifest.ReconcileException::class.java) { + stored.reconcile(fresh, running = true, blobSize = blobSize) + } + } + + @Test + fun `recreate rejects parts that do not tile the blob exactly`() { + val stored = manifest() + for ( + bad in listOf( + listOf(part(0, 100), part(150, 250)), // gap + listOf(part(0, 150), part(100, 250)), // overlap + listOf(part(50, 250)), // does not start at 0 + listOf(part(0, 200)), // short of the blob size + listOf(part(0, 100), part(100, 251)), // past the blob size + ) + ) { + assertThrows(ChunkedManifest.ReconcileException::class.java) { + stored.reconcile(manifest(parts = bad), running = false, blobSize = blobSize) + } + } + } + + @Test + fun `recreate accepts parts authored in any order`() { + val stored = manifest() + val fresh = manifest(parts = listOf(part(100, 250), part(0, 100)).map { it.copy(url = it.url + "&v=2") }) + val recreated = stored.reconcile(fresh, running = false, blobSize = blobSize) + assertEquals(2, recreated.parts.size) + } + + @Test + fun `tilesExactly covers the edge shapes`() { + assertTrue(ChunkedManifest.tilesExactly(listOf(part(0, 250)), 250)) + assertFalse(ChunkedManifest.tilesExactly(emptyList(), 0)) + assertFalse(ChunkedManifest.tilesExactly(listOf(part(0, 0)), 0)) // empty range + assertFalse(ChunkedManifest.tilesExactly(listOf(part(0, 250)), 300)) + } + + // MARK: - Create validation + + @Test + fun `create accepts parts that tile the blob exactly`() { + val m = manifest() + assertEquals(m, ChunkedManifest.validatedForCreate(m, blobSize)) + } + + @Test + fun `create rejects parts that do not tile the blob`() { + for ( + bad in listOf( + listOf(part(0, 100), part(150, 250)), // gap + listOf(part(0, 150), part(100, 250)), // overlap + listOf(part(50, 250)), // does not start at 0 + listOf(part(0, 200)), // short of the blob size + listOf(part(0, 100), part(100, 251)), // past the blob size + ) + ) { + assertThrows(ChunkedManifest.ReconcileException::class.java) { + ChunkedManifest.validatedForCreate(manifest(parts = bad), blobSize) + } + } + } + + @Test + fun `a rejected create writes no manifest, leaving the blob adoptable`() { + // startUpload validates AFTER takeOwnership moved the bytes. The throw + // propagates out of compute before a save. Thus the blob sits ownerless at + // its path. That is exactly what takeOwnership's orphan branch adopts on + // the corrected retry. + val store = ChunkedManifestStore(tmp.newFolder()) + store.blobFile("u1").apply { parentFile!!.mkdirs() }.writeText("owned bytes") + assertThrows(ChunkedManifest.ReconcileException::class.java) { + store.compute("u1") { ChunkedManifest.validatedForCreate(manifest(), 999L) } + } + assertNull(store.load("u1")) + assertTrue(store.blobFile("u1").exists()) + } + + // MARK: - Store + + @Test + fun `save then load round-trips, across store instances`() { + val dir = tmp.newFolder() + val m = manifest().withPartAccepted(1) + ChunkedManifestStore(dir).save(m) + // A new instance over the same dir is what a process relaunch looks like. + assertEquals(m, ChunkedManifestStore(dir).load("u1")) + } + + @Test + fun `load returns null for an unknown id`() { + assertNull(ChunkedManifestStore(tmp.newFolder()).load("nope")) + } + + @Test + fun `update persists the transformed manifest`() { + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest()) + val updated = store.update("u1") { it.withPartAccepted(0) } + assertTrue(updated!!.parts[0].accepted) + assertTrue(store.load("u1")!!.parts[0].accepted) + } + + @Test + fun `update of a missing manifest returns null`() { + assertNull(ChunkedManifestStore(tmp.newFolder()).update("nope") { it }) + } + + @Test + fun `compute creates when no manifest exists`() { + val store = ChunkedManifestStore(tmp.newFolder()) + val created = store.compute("u1") { existing -> + assertNull(existing) + manifest() + } + assertEquals(created, store.load("u1")) + } + + @Test + fun `compute holds the store lock across load, transform, and save`() { + // The startUpload reconcile and a running worker's markAccepted race. If + // the lock did not span all three steps, the update below could land + // between compute's load and save, and it would be erased from disk. When + // they are serialized, both effects must survive. + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest()) + val inTransform = CountDownLatch(1) + val computing = Thread { + store.compute("u1") { existing -> + inTransform.countDown() + Thread.sleep(300) // hold the lock with load done and save not yet run + existing!!.copy(expiresAt = 99_000) + } + }.apply { start() } + assertTrue(inTransform.await(5, TimeUnit.SECONDS)) + val updating = Thread { store.update("u1") { it.withPartAccepted(0) } }.apply { start() } + computing.join() + updating.join() + val final = store.load("u1")!! + assertEquals(99_000, final.expiresAt) + assertTrue(final.parts[0].accepted) + } + + @Test + fun `a throwing compute transform propagates and writes nothing`() { + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest()) + assertThrows(ChunkedManifest.ReconcileException::class.java) { + store.compute("u1") { throw ChunkedManifest.ReconcileException("rejected") } + } + assertEquals(manifest(), store.load("u1")) + } + + @Test + fun `contains tracks save and remove`() { + val store = ChunkedManifestStore(tmp.newFolder()) + assertFalse(store.contains("u1")) + store.save(manifest()) + assertTrue(store.contains("u1")) + store.remove("u1") + assertFalse(store.contains("u1")) + } + + @Test + fun `remove deletes the manifest and the blob`() { + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest()) + store.blobFile("u1").writeText("bytes") + store.remove("u1") + assertNull(store.load("u1")) + assertFalse(store.blobFile("u1").exists()) + } + + @Test + fun `remove of an unknown id is a no-op`() { + ChunkedManifestStore(tmp.newFolder()).remove("simple-upload-id") + } + + @Test + fun `ids with filesystem-hostile characters round-trip`() { + val store = ChunkedManifestStore(tmp.newFolder()) + val id = "a/b:c dü..\\e" + store.save(manifest(id = id)) + assertEquals(id, store.load(id)!!.id) + store.remove(id) + assertNull(store.load(id)) + } + + @Test + fun `a corrupt manifest reads as absent, not fatal`() { + val dir = tmp.newFolder() + val store = ChunkedManifestStore(dir) + store.save(manifest()) + File(File(dir, dir.list()!!.first()), "manifest.json").writeText("{not json") + assertNull(store.load("u1")) + assertEquals(emptyList(), store.all()) + } + + @Test + fun `all lists every stored manifest`() { + val store = ChunkedManifestStore(tmp.newFolder()) + store.save(manifest(id = "u1")) + store.save(manifest(id = "u2").withPartAccepted(0)) + assertEquals(setOf("u1", "u2"), store.all().map { it.id }.toSet()) + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/ChunkedWorkerGateTest.kt b/android/src/test/java/ai/openspace/backgroundupload/ChunkedWorkerGateTest.kt new file mode 100644 index 00000000..0b65d54b --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/ChunkedWorkerGateTest.kt @@ -0,0 +1,57 @@ +package ai.openspace.backgroundupload + +import org.junit.After +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test + +class ChunkedWorkerGateTest { + private val a = Any() + private val b = Any() + + @After + fun tearDown() { + // The gate is a process-wide singleton. Leave nothing for other tests. + ChunkedWorkerGate.release("u1", a) + ChunkedWorkerGate.release("u1", b) + ChunkedWorkerGate.release("u2", a) + } + + @Test + fun `a second worker for the same id must wait`() { + assertTrue(ChunkedWorkerGate.tryAcquire("u1", a)) + // The replacement worker after a cancel-then-start: it must not run a part + // PUT while the cancelled worker still holds the id. + assertFalse(ChunkedWorkerGate.tryAcquire("u1", b)) + ChunkedWorkerGate.release("u1", a) + assertTrue(ChunkedWorkerGate.tryAcquire("u1", b)) + } + + @Test + fun `reacquiring with the same token is idempotent`() { + assertTrue(ChunkedWorkerGate.tryAcquire("u1", a)) + assertTrue(ChunkedWorkerGate.tryAcquire("u1", a)) + } + + @Test + fun `a stale release cannot evict a successor`() { + assertTrue(ChunkedWorkerGate.tryAcquire("u1", a)) + ChunkedWorkerGate.release("u1", a) + assertTrue(ChunkedWorkerGate.tryAcquire("u1", b)) + ChunkedWorkerGate.release("u1", a) // the old worker's finally, arriving late + assertTrue(ChunkedWorkerGate.isRunning("u1")) + assertFalse(ChunkedWorkerGate.tryAcquire("u1", a)) + } + + @Test + fun `ids are independent and isRunning tracks the holder`() { + assertFalse(ChunkedWorkerGate.isRunning("u1")) + assertTrue(ChunkedWorkerGate.tryAcquire("u1", a)) + assertTrue(ChunkedWorkerGate.isRunning("u1")) + assertFalse(ChunkedWorkerGate.isRunning("u2")) + assertTrue(ChunkedWorkerGate.tryAcquire("u2", a)) + ChunkedWorkerGate.release("u1", a) + assertFalse(ChunkedWorkerGate.isRunning("u1")) + assertTrue(ChunkedWorkerGate.isRunning("u2")) + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/NotificationConfigTest.kt b/android/src/test/java/ai/openspace/backgroundupload/NotificationConfigTest.kt new file mode 100644 index 00000000..d3545c78 --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/NotificationConfigTest.kt @@ -0,0 +1,53 @@ +package ai.openspace.backgroundupload + +import com.google.gson.Gson +import org.junit.Assert.assertEquals +import org.junit.Test + +class NotificationConfigTest { + private val gson = Gson() + + private val configured = NotificationConfig( + notificationId = "my-id", + notificationTitle = "Backing up…", + notificationTitleNoInternet = "Offline", + notificationTitleNoWifi = "No wifi", + notificationChannel = "my-channel", + ) + + @Test + fun `no stored config yields the defaults`() { + assertEquals(NotificationConfig.DEFAULTS, NotificationConfig.fromJson(null)) + } + + @Test + fun `a configured blob survives a persistence round trip`() { + assertEquals(configured, NotificationConfig.fromJson(gson.toJson(configured))) + } + + @Test + fun `a corrupt blob degrades to the defaults`() { + assertEquals(NotificationConfig.DEFAULTS, NotificationConfig.fromJson("][")) + } + + // A blob from a build with fewer fields must not make the other fields null. + @Test + fun `fields missing from a stored blob fall back individually`() { + val config = NotificationConfig.fromJson("""{"notificationTitle":"Custom"}""") + assertEquals("Custom", config.notificationTitle) + assertEquals(NotificationConfig.DEFAULTS.notificationChannel, config.notificationChannel) + assertEquals(NotificationConfig.DEFAULTS.notificationTitleNoWifi, config.notificationTitleNoWifi) + assertEquals( + NotificationConfig.DEFAULTS.notificationTitleNoInternet, + config.notificationTitleNoInternet, + ) + } + + // This is the same derivation that v8 applied to the per-upload option. Thus + // an app that gives its old notificationId to configure() keeps the same + // system notification. + @Test + fun `the system notification id derives from the configured string`() { + assertEquals("my-id".hashCode(), configured.systemNotificationId) + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/UploadOutcomeTest.kt b/android/src/test/java/ai/openspace/backgroundupload/UploadOutcomeTest.kt index b9375ce0..187349ed 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/UploadOutcomeTest.kt +++ b/android/src/test/java/ai/openspace/backgroundupload/UploadOutcomeTest.kt @@ -1,5 +1,6 @@ package ai.openspace.backgroundupload +import ai.openspace.backgroundupload.UploadOutcome.AcceptRule import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertTrue @@ -10,28 +11,51 @@ class UploadOutcomeTest { @Test fun `2xx is accepted`() { - assertTrue(UploadOutcome.isAccepted(200, listOf())) - assertTrue(UploadOutcome.isAccepted(204, listOf())) - assertTrue(UploadOutcome.isAccepted(299, listOf())) + assertTrue(UploadOutcome.isAccepted(200, "", listOf())) + assertTrue(UploadOutcome.isAccepted(204, "", listOf())) + assertTrue(UploadOutcome.isAccepted(299, "", listOf())) } @Test fun `non-2xx is not accepted by default`() { - assertFalse(UploadOutcome.isAccepted(199, listOf())) - assertFalse(UploadOutcome.isAccepted(300, listOf())) - assertFalse(UploadOutcome.isAccepted(404, listOf())) - assertFalse(UploadOutcome.isAccepted(500, listOf())) + assertFalse(UploadOutcome.isAccepted(199, "", listOf())) + assertFalse(UploadOutcome.isAccepted(300, "", listOf())) + assertFalse(UploadOutcome.isAccepted(404, "", listOf())) + assertFalse(UploadOutcome.isAccepted(500, "", listOf())) } @Test - fun `non-2xx listed in acceptStatus is accepted`() { - assertTrue(UploadOutcome.isAccepted(409, listOf(409))) - assertTrue(UploadOutcome.isAccepted(404, listOf(404, 409))) + fun `a status-only rule accepts its status regardless of body`() { + val rules = listOf(AcceptRule(409)) + assertTrue(UploadOutcome.isAccepted(409, "anything", rules)) + assertTrue(UploadOutcome.isAccepted(409, null, rules)) } @Test - fun `acceptStatus does not accept unlisted codes`() { - assertFalse(UploadOutcome.isAccepted(500, listOf(409))) + fun `rules do not accept unlisted codes`() { + assertFalse(UploadOutcome.isAccepted(500, "", listOf(AcceptRule(409)))) + } + + // The backend's 409 carries several meanings, and only the message shows the + // difference. bodyIncludes is what separates the success meanings from the + // bug meanings. + @Test + fun `bodyIncludes narrows a rule to matching bodies`() { + val rules = listOf(AcceptRule(409, bodyIncludes = "already completed")) + assertTrue(UploadOutcome.isAccepted(409, "part already completed", rules)) + assertFalse(UploadOutcome.isAccepted(409, "part number mismatch", rules)) + assertFalse(UploadOutcome.isAccepted(409, null, rules)) + } + + @Test + fun `any matching rule accepts`() { + val rules = listOf( + AcceptRule(409, bodyIncludes = "already completed"), + AcceptRule(208), + ) + assertTrue(UploadOutcome.isAccepted(208, "", rules)) + assertTrue(UploadOutcome.isAccepted(409, "already completed", rules)) + assertFalse(UploadOutcome.isAccepted(410, "already completed", rules)) } @Test diff --git a/android/src/test/java/ai/openspace/backgroundupload/UploadStatesTest.kt b/android/src/test/java/ai/openspace/backgroundupload/UploadStatesTest.kt new file mode 100644 index 00000000..bef1858c --- /dev/null +++ b/android/src/test/java/ai/openspace/backgroundupload/UploadStatesTest.kt @@ -0,0 +1,84 @@ +package ai.openspace.backgroundupload + +import androidx.work.WorkInfo.State.BLOCKED +import androidx.work.WorkInfo.State.CANCELLED +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 androidx.work.ListenableWorker +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test + +// getAllUploads must return ONE row per upload id, although WorkManager can +// hold several rows for it (finished chains linger for roughly a day, and +// APPEND_OR_REPLACE resumes add rows). The state vocabulary is the same as +// iOS's. +class UploadStatesTest { + + @Test + fun `a live row wins for a chunked upload`() { + assertEquals("running", chunkedUploadState(listOf(CANCELLED, RUNNING), allAccepted = false)) + assertEquals("running", chunkedUploadState(listOf(RUNNING, BLOCKED), allAccepted = false)) + assertEquals("pending", chunkedUploadState(listOf(FAILED, ENQUEUED), allAccepted = false)) + assertEquals("pending", chunkedUploadState(listOf(BLOCKED), allAccepted = false)) + } + + @Test + fun `with no live row the manifest speaks, never a lingering finished row`() { + // A cancelled chunked upload keeps its manifest. Its truthful state is + // stalled-awaiting-resume ("error"), not "cancelled". iOS's getAllUploads + // never reports "cancelled" for a lingering upload. + assertEquals("error", chunkedUploadState(listOf(CANCELLED), allAccepted = false)) + assertEquals("error", chunkedUploadState(listOf(FAILED), allAccepted = false)) + assertEquals("error", chunkedUploadState(emptyList(), allAccepted = false)) + assertEquals("completed", chunkedUploadState(listOf(SUCCEEDED), allAccepted = true)) + assertEquals("completed", chunkedUploadState(emptyList(), allAccepted = true)) + // The row that a run leaves after it journals a terminal error is + // SUCCEEDED (see terminalErrorResult). The manifest, not the row, carries + // the outcome. + assertEquals("error", chunkedUploadState(listOf(SUCCEEDED), allAccepted = false)) + } + + @Test + fun `a journaled terminal error still succeeds the row`() { + // WorkManager marks the dependents of a FAILED prerequisite FAILED without + // a run. Thus a resume appended during a failing run's teardown would + // silently never run. The journal and the manifest are the outcome record, + // never the row state. + assertTrue(terminalErrorResult() is ListenableWorker.Result.Success) + } + + @Test + fun `cancel reports from the module only when no worker is running`() { + assertTrue(cancelReportsFromModule(listOf(ENQUEUED))) + // An appended chain's dependent is BLOCKED, not ENQUEUED. It is still + // never-started, and it is still owed a module-side 'cancelled'. + assertTrue(cancelReportsFromModule(listOf(BLOCKED))) + assertTrue(cancelReportsFromModule(listOf(ENQUEUED, BLOCKED))) + // A RUNNING worker's stop handler owns the report. + assertFalse(cancelReportsFromModule(listOf(RUNNING))) + assertFalse(cancelReportsFromModule(listOf(RUNNING, BLOCKED))) + assertFalse(cancelReportsFromModule(emptyList())) + } + + @Test + fun `a queued successor suppresses another append`() { + assertTrue(hasQueuedSuccessor(listOf(RUNNING, BLOCKED))) + assertTrue(hasQueuedSuccessor(listOf(ENQUEUED))) + assertFalse(hasQueuedSuccessor(listOf(RUNNING))) + assertFalse(hasQueuedSuccessor(listOf(SUCCEEDED, FAILED, CANCELLED))) + assertFalse(hasQueuedSuccessor(emptyList())) + } + + @Test + fun `a simple upload reports its live row first, then the most conclusive finished one`() { + assertEquals("running", simpleUploadState(listOf(CANCELLED, RUNNING))) + assertEquals("pending", simpleUploadState(listOf(SUCCEEDED, ENQUEUED))) + assertEquals("completed", simpleUploadState(listOf(CANCELLED, SUCCEEDED))) + assertEquals("error", simpleUploadState(listOf(CANCELLED, FAILED))) + assertEquals("cancelled", simpleUploadState(listOf(CANCELLED))) + } +} diff --git a/android/src/test/java/ai/openspace/backgroundupload/UploadTest.kt b/android/src/test/java/ai/openspace/backgroundupload/UploadTest.kt index de856578..a99178d1 100644 --- a/android/src/test/java/ai/openspace/backgroundupload/UploadTest.kt +++ b/android/src/test/java/ai/openspace/backgroundupload/UploadTest.kt @@ -1,6 +1,7 @@ package ai.openspace.backgroundupload import com.google.gson.Gson +import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertTrue import org.junit.Test @@ -13,15 +14,9 @@ class UploadTest { url = "https://example.com/upload", path = "/tmp/file", method = "POST", - maxRetries = 5, wifiOnly = false, - acceptStatus = listOf(), + accept = listOf(), headers = mapOf(), - notificationId = 1, - notificationTitle = "Uploading…", - notificationTitleNoInternet = "Waiting for connection…", - notificationTitleNoWifi = "Waiting for Wi-Fi…", - notificationChannel = "background-upload", noNotification = noNotification, ) @@ -46,4 +41,72 @@ class UploadTest { json.remove(Upload::noNotification.name) assertTrue(gson.fromJson(json, Upload::class.java).showsNotification) } + + // One build can enqueue a WorkManager job, and the next build can replay it. + // This is the exact JSON shape that a v8 build serialized into input data + // (Gson.toJson of the v8 Upload model): `acceptStatus: List`, and no + // `accept`. Gson does not use the constructor. Thus, without normalized(), + // the replayed object's `accept` is NULL, and the worker NPEs after the file + // has fully transmitted. WorkManager then re-runs it and re-sends the whole + // file. + private val v8JobJson = """ + { + "id": "u1", + "url": "https://example.com/upload", + "path": "/tmp/file", + "method": "PUT", + "maxRetries": 5, + "wifiOnly": false, + "acceptStatus": [409, 208], + "headers": {"Authorization": "Bearer t"}, + "notificationId": 123456, + "notificationTitle": "Uploading…", + "notificationTitleNoInternet": "Waiting for connection…", + "notificationTitleNoWifi": "Waiting for Wi-Fi…", + "notificationChannel": "background-upload", + "noNotification": false + } + """ + + @Test + fun `a replayed v8 job maps acceptStatus to accept rules and is safe to run`() { + val replayed = gson.fromJson(v8JobJson, Upload::class.java).normalized() + assertEquals( + listOf(UploadOutcome.AcceptRule(409), UploadOutcome.AcceptRule(208)), + replayed.accept, + ) + // The worker-facing calls that NPE'd on the un-normalized object. + assertTrue(UploadOutcome.isAccepted(409, "duplicate", replayed.accept)) + assertFalse(UploadOutcome.isAccepted(400, "", replayed.accept)) + assertEquals("u1", replayed.id) + assertEquals(mapOf("Authorization" to "Bearer t"), replayed.headers) + assertTrue(replayed.showsNotification) + } + + @Test + fun `a replayed v8 job with an empty acceptStatus gets no rules`() { + val json = gson.fromJson(v8JobJson, com.google.gson.JsonObject::class.java) + json.add("acceptStatus", com.google.gson.JsonArray()) + val replayed = gson.fromJson(json, Upload::class.java).normalized() + assertEquals(emptyList(), replayed.accept) + assertTrue(UploadOutcome.isAccepted(200, "", replayed.accept)) + } + + @Test + fun `a job with neither accept nor acceptStatus normalizes to no rules`() { + val json = gson.fromJson(v8JobJson, com.google.gson.JsonObject::class.java) + json.remove("acceptStatus") + val replayed = gson.fromJson(json, Upload::class.java).normalized() + assertEquals(emptyList(), replayed.accept) + assertFalse(UploadOutcome.isAccepted(409, "duplicate", replayed.accept)) + } + + @Test + fun `normalized passes a current-shape job through unchanged`() { + val current = upload(noNotification = true).copy( + accept = listOf(UploadOutcome.AcceptRule(409, "already completed")), + ) + val replayed = gson.fromJson(gson.toJson(current), Upload::class.java).normalized() + assertEquals(current, replayed) + } } diff --git a/example/RNBGUExample/App.tsx b/example/RNBGUExample/App.tsx index d3147e02..f7dbc9f7 100644 --- a/example/RNBGUExample/App.tsx +++ b/example/RNBGUExample/App.tsx @@ -17,9 +17,11 @@ import { Button, } from 'react-native'; import notifee, {AndroidImportance} from '@notifee/react-native'; -import {Colors} from 'react-native/Libraries/NewAppScreen'; -import Upload, {UploadOptions} from 'react-native-background-upload'; +import Upload, { + ChunkedUploadOptions, + UploadOptions, +} from 'react-native-background-upload'; import * as RNFS from 'react-native-fs'; @@ -27,6 +29,8 @@ const TEST_FILE = `${RNFS.DocumentDirectoryPath}/1MB.bin`; const TEST_FILE_URL = 'https://gist.githubusercontent.com/khaykov/a6105154becce4c0530da38e723c2330/raw/41ab415ac41c93a198f7da5b47d604956157c5c3/gistfile1.txt'; const UPLOAD_URL = 'https://httpbin.org/post'; +const CHUNKED_UPLOAD_URL = 'https://httpbin.org/put'; +const NOTIFICATION_CHANNEL = 'RNBGUExample'; const App = () => { const [uploadId, setUploadId] = useState(); @@ -36,16 +40,31 @@ const App = () => { >(); useEffect(() => { - Upload.addListener('progress', null, data => { + // One-time notification configuration. The library keeps it in native + // storage. Thus a headless WorkManager relaunch shows the same text. The + // call does nothing on iOS. + Upload.configure({ + android: { + notificationId: NOTIFICATION_CHANNEL, + notificationTitle: NOTIFICATION_CHANNEL, + notificationTitleNoWifi: 'No wifi', + notificationTitleNoInternet: 'No internet', + notificationChannel: NOTIFICATION_CHANNEL, + }, + }); + }, []); + + useEffect(() => { + Upload.addListener('progress', data => { setProgress(data.progress); }); - Upload.addListener('error', null, data => { + Upload.addListener('error', data => { console.log('Error!', JSON.stringify(data)); }); - Upload.addListener('completed', null, data => { + Upload.addListener('completed', data => { console.log('Completed!', JSON.stringify(data)); }); - Upload.addListener('cancelled', null, data => { + Upload.addListener('cancelled', data => { console.log('Cancelled!', JSON.stringify(data)); }); }, []); @@ -62,24 +81,20 @@ const App = () => { .then(() => setTestFileDownload('downloaded')); }, []); - const onPressUpload = async () => { + const ensureNotificationChannel = async () => { await notifee.requestPermission({alert: true, sound: true}); - const channelId = 'RNBGUExample'; await notifee.createChannel({ - id: channelId, - name: channelId, + id: NOTIFICATION_CHANNEL, + name: NOTIFICATION_CHANNEL, importance: AndroidImportance.LOW, }); + }; + + const onPressUpload = async () => { + await ensureNotificationChannel(); const uploadOpts: UploadOptions = { - android: { - notificationId: channelId, - notificationTitle: channelId, - notificationTitleNoWifi: 'No wifi', - notificationTitleNoInternet: 'No internet', - notificationChannel: channelId, - }, type: 'raw', url: UPLOAD_URL, path: TEST_FILE, @@ -102,6 +117,53 @@ const App = () => { }); }; + const onPressChunkedUpload = async () => { + await ensureNotificationChannel(); + + // The library takes ownership of a chunked upload's file. It renames the + // file into its own directory. Thus we upload a copy, and the test file + // stays available. + const chunkedFile = `${RNFS.DocumentDirectoryPath}/chunked.bin`; + if (await RNFS.exists('file://' + chunkedFile)) { + await RNFS.unlink(chunkedFile); + } + await RNFS.copyFile(TEST_FILE, chunkedFile); + + // A small min and max, so the 1MB test file still splits into some parts. + // Production callers use the server's real part-size limits. + const {size} = await RNFS.stat(chunkedFile); + const ranges = Upload.chunkPlan(size, {min: 128 * 1024, max: 256 * 1024}); + + const uploadOpts: ChunkedUploadOptions = { + type: 'chunked', + id: 'chunked-demo', + path: chunkedFile, + parts: ranges.map((range, i) => ({ + url: `${CHUNKED_UPLOAD_URL}?partNum=${i + 1}`, + headers: { + 'Content-Type': 'application/octet-stream', + 'Content-Range': `bytes ${range.start}-${range.end - 1}/${size}`, + }, + range, + })), + expiresAt: Date.now() + 24 * 60 * 60 * 1000, + }; + + Upload.startUpload(uploadOpts) + .then(uploadId => { + console.log( + `Chunked upload started: ${uploadId} (${ranges.length} parts)`, + ); + setUploadId(uploadId); + setProgress(0); + }) + .catch(function (err) { + setUploadId(undefined); + setProgress(undefined); + console.log('Chunked upload error!', err); + }); + }; + return ( <> @@ -118,6 +180,7 @@ const App = () => {