diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4811236e..b3a475a6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -116,6 +116,35 @@ pending until the returned function is called, so a test can model a native
that is slow to answer. Later enqueues wait behind it, as they do on a
device.
+iOS no longer makes a `mutate()` wait while a chunked upload copies all of
+its parts. Since 11.0.0, a chunked `mutate()` copied every part out of the
+file and created every part task in one block on the queue, so the next
+`mutate()` waited for all of it. For a 2 GiB file on an older iPhone this
+could pass the 10 s enqueue timeout. The parts are now built in slices of
+about 32 MiB, and other calls run between two slices: a `mutate()` waits for
+at most one slice per chunked upload that is building its parts. Each slice
+writes the task map once; before, every part task wrote the whole task map.
+The part copy frees each 1 MiB read buffer before the next read. The parts
+still get their tasks right after `mutate()`, one slice after the other. A
+background wake keeps its completion handler until the last slice is built,
+or until less than 5 s of background time is left. Replacing the part tasks
+created in the background, when the app becomes active, is sliced the same
+way, and each part keeps its old task until its new one exists, unless
+that slice's build fails. Each slice checks again that an old task has not
+started sending; one that has is kept. Replacing stops when the app leaves
+the foreground again.
+
+iOS now resumes every task it adopts at launch. A crash between writing a
+task's task map entry and resuming it left a suspended task that held its
+part, or its request, and never sent.
+
+On iOS, a running chunked upload past its `expiresAt` now settles `expired`
+when the app becomes active and its background-created part tasks are due
+for replacement. Before, it got new part tasks.
+
+iOS adds `os_signpost` intervals for each enqueue, issue and part slice, so
+Instruments can show how long each one holds the queue.
+
## 11.1.0
Added `takeDiagnostics()`. On Android it returns the module's recent
@@ -151,8 +180,8 @@ A replaced simple request whose body landed anyway now completes its entry,
even when its replacement failed and the entry moved to a later attempt.
Before, the late success was dropped and the request was sent again.
-A chunked upload saves its entry and the task map once per batch of part
-tasks instead of once per part.
+A chunked upload saves its entry once per batch of part tasks instead of
+once per part.
## 11.0.0
diff --git a/README.md b/README.md
index 6e84bf0d..4c5dca1b 100644
--- a/README.md
+++ b/README.md
@@ -55,6 +55,17 @@ relaunched with no JS running, so queued completions get journaled. `RNFileUploa
is the TurboModule and is deliberately not reachable from plain Objective-C, because its
generated header is Objective-C++ only. So the handler lives on `RNBackgroundUpload`.
+The library calls the handler when its work for the wake is done: after the
+relaunch reconcile, its grace waits, and every queued slice of chunked part
+tasks (see "Every part is queued at once"). A part slice can take long on an
+older device, so the library stops waiting for slices when less than 5 s of
+background time is left, and calls the handler then. The parts not built yet
+get their tasks at the next wake or launch. Usually that wake comes, because
+the upload already has the tasks of its first slice and the system wakes the
+app when one of them finishes. It may not come when no slice of the upload
+could be built (a full disk, for example): then only an in-process timer
+retries the build, and the parts wait until the app next runs.
+
# Usage
The library owns a durable queue of HTTP requests: JSON bodies, multipart
@@ -226,16 +237,25 @@ two background sessions (cellular allowed, and Wi-Fi only) sets
`httpMaximumConnectionsPerHost = 4`. That caps connections to one host, not
requests: over HTTP/2, every queued part can be in flight at once.
-**Every part is queued at once.** A chunked upload creates a task for every
-part at `mutate()`. Each part task reads its own part file, so a chunked
+**Every part is queued at once.** A chunked upload starts creating a task
+for every part at `mutate()`. It copies the parts and creates their tasks in
+slices of about 32 MiB, one slice at a time, and other calls run between two
+slices. So a `mutate()` right after a large chunked one waits for at most one
+slice per chunked upload that is building its parts, not for whole files. Each part task reads its own part file, so a chunked
upload needs the file's size again in free space until its parts are sent.
+**Instruments.** Each `enqueue`, each `issue` and each part slice is an
+`os_signpost` interval (subsystem `ai.openspace.rnbgupload`, Points of
+Interest), with the entry id and the part count.
+
**Not at the system's discretion.** Both sessions set
`isDiscretionary = false`, but iOS only honors that for a task created while
the app is in the foreground. It starts a task created in the background at
its own discretion, for example when the device charges. Tasks are often
created in the background: a retry scheduled while the app is there, a
-`mutate()` from a background wake, and the tasks a relaunch rebuilds. So the library records whether the app was active when it created
+`mutate()` from a background wake, the tasks a relaunch rebuilds, and the
+part tasks of a chunked upload whose slice runs after the app leaves the
+foreground. So the library records whether the app was active when it created
each task. Two seconds after the app becomes active, it replaces every task
created while the app was not active, unless the task has sent a byte of its
body or has a response. The delay lets the system report what the task did
diff --git a/example/RNBGUExample/App.tsx b/example/RNBGUExample/App.tsx
index 8ce289bf..9a954cfd 100644
--- a/example/RNBGUExample/App.tsx
+++ b/example/RNBGUExample/App.tsx
@@ -29,6 +29,7 @@ import {
remove,
type TestFile,
} from './harness/files';
+import {LATENCY_RUNS, measureEnqueueLatency} from './harness/latency';
import {clearLog, clock, getLog, log, subscribeLog} from './harness/log';
import type {LogEntry, LogKind} from './harness/log';
import {saveSettings, settings} from './harness/settings';
@@ -210,6 +211,9 @@ const App = () => {
...[1, 2, 3].map(() => enqueueJson('slow', undefined)),
]);
}),
+ latency: run(`Enqueue latency x${LATENCY_RUNS}`, () =>
+ measureEnqueueLatency(settings.host),
+ ),
stats: run('Server stats', async () =>
(await fetch(`${settings.host}/stats`)).json(),
),
@@ -386,6 +390,13 @@ const App = () => {
disabled={!isReady}
/>
+
+
+
({
exists: jest.fn(() => Promise.resolve(false)),
readFile: jest.fn(() => Promise.resolve('{}')),
writeFile: jest.fn(() => Promise.resolve()),
+ stat: jest.fn(() => Promise.resolve({size: 2048 * 1024 * 1024})),
+ copyFile: jest.fn(() => Promise.resolve()),
}));
jest.mock('@notifee/react-native', () => ({
__esModule: true,
@@ -112,3 +114,43 @@ it('rejects a GET with a body before native sees it', async () => {
await press(tree, 'GET with a body');
expect(mockNative.enqueue.mock.calls.length).toBe(before);
});
+
+it('runs the enqueue latency measurement 7 times on /slow', async () => {
+ const fs = require('react-native-fs') as {
+ exists: jest.Mock<(path: string) => Promise>;
+ copyFile: jest.Mock;
+ };
+ // The 2 GiB source is already made.
+ fs.exists.mockImplementation(path =>
+ Promise.resolve(path.endsWith('/latency-src.bin')),
+ );
+ const before = mockNative.enqueue.mock.calls.length;
+ const resumes = mockNative.resume.mock.calls.length;
+ try {
+ await press(tree, 'Enqueue latency x7: 2 GiB, 20 MiB parts, on /slow');
+ for (let i = 0; i < 100 && mockNative.cancel.mock.calls.length < 14; i++) {
+ await act(async () => {
+ await new Promise(r => setTimeout(r, 0));
+ });
+ }
+ } finally {
+ fs.exists.mockImplementation(() => Promise.resolve(false));
+ }
+ const entries = mockNative.enqueue.mock.calls
+ .slice(before)
+ .map(c => c[0] as unknown as {key: string; descriptor: any});
+ expect(entries.map(e => e.key)).toEqual(
+ Array(7).fill(['harness.chunked20', 'harness.json']).flat(),
+ );
+ const parts = entries[0].descriptor.parts as Array<{
+ url: string;
+ headers: Record;
+ }>;
+ expect(parts).toHaveLength(103);
+ expect(parts[0].url).toMatch(/\/slow\/chunk\/[\w-]+\/1$/);
+ expect(parts[0].headers['Content-Range']).toBe('0-20971519/2147483648');
+ expect(entries[1].descriptor.url).toMatch(/\/slow\/json\//);
+ expect(fs.copyFile).toHaveBeenCalledTimes(7);
+ expect(mockNative.cancel).toHaveBeenCalledTimes(14);
+ expect(mockNative.resume.mock.calls.length).toBe(resumes + 1);
+});
diff --git a/example/RNBGUExample/harness/files.ts b/example/RNBGUExample/harness/files.ts
index 349dcdd3..1d1a7a8e 100644
--- a/example/RNBGUExample/harness/files.ts
+++ b/example/RNBGUExample/harness/files.ts
@@ -47,8 +47,15 @@ const ensureDir = async () => {
export type TestFile = {path: string; size: number; md5: string};
-/** Writes a new file of `mb` MiB and returns its path, size, and MD5. */
-export const makeFile = async (name: string, mb: number): Promise => {
+/**
+ * Writes a new file of `mb` MiB and returns its path, size, and MD5. With
+ * `md5: false` the MD5 is '' (hashing a large file takes long).
+ */
+export const makeFile = async (
+ name: string,
+ mb: number,
+ options: {md5?: boolean} = {},
+): Promise => {
await ensureDir();
const path = `${TEST_DIR}/${name}`;
await RNFS.writeFile(path, block(0), 'utf8');
@@ -56,7 +63,8 @@ export const makeFile = async (name: string, mb: number): Promise => {
await RNFS.appendFile(path, block(i), 'utf8');
}
const {size} = await RNFS.stat(path);
- return {path, size: Number(size), md5: await RNFS.hash(path, 'md5')};
+ const md5 = options.md5 === false ? '' : await RNFS.hash(path, 'md5');
+ return {path, size: Number(size), md5};
};
/** Writes a new 1x1 JPEG and returns its path. */
diff --git a/example/RNBGUExample/harness/latency.ts b/example/RNBGUExample/harness/latency.ts
new file mode 100644
index 00000000..3bda828d
--- /dev/null
+++ b/example/RNBGUExample/harness/latency.ts
@@ -0,0 +1,113 @@
+import * as RNFS from 'react-native-fs';
+import Upload from 'react-native-background-upload';
+
+import {MB, TEST_DIR, makeFile, remove} from './files';
+import {log} from './log';
+import {postJson, putChunked20} from './uploads';
+
+// The enqueue latency run. It measures what Diana does after a capture: the
+// file mutate(), then at once the metadata mutate(). nextMs is how long the
+// second mutate() waits. Before 11.2.0 it waited while iOS copied every part
+// of the file, and a 2 GiB file could pass the 10 s enqueue timeout.
+
+export const LATENCY_RUNS = 7;
+const SOURCE_MB = 2048;
+const SOURCE_NAME = 'latency-src.bin';
+const SOURCE = `${TEST_DIR}/${SOURCE_NAME}`;
+
+/** The 2 GiB source, made one time and kept. Making it takes minutes. */
+const source = async (): Promise => {
+ const size = SOURCE_MB * MB;
+ if (await RNFS.exists(SOURCE)) {
+ const stat = await RNFS.stat(SOURCE);
+ if (Number(stat.size) === size) {
+ return size;
+ }
+ }
+ log('action', `latency: making the ${SOURCE_MB} MiB source, one time`);
+ await makeFile(SOURCE_NAME, SOURCE_MB, {md5: false});
+ return size;
+};
+
+const sleep = (ms: number) => new Promise(r => setTimeout(r, ms));
+
+/**
+ * Waits until native forgets `ids` (cancel, then the automatic ack). Logs a
+ * warning when they are still there after 60 s.
+ */
+const waitGone = async (ids: string[]) => {
+ const until = Date.now() + 60_000;
+ const left = () => Upload.getRequests().filter(r => ids.includes(r.id));
+ while (left().length > 0) {
+ if (Date.now() >= until) {
+ log(
+ 'warn',
+ `latency: still in the queue after 60 s: ${left()
+ .map(r => `${r.id} ${r.state}`)
+ .join(', ')}`,
+ );
+ return;
+ }
+ await sleep(200);
+ }
+};
+
+export type LatencyResult = {nextMs: number[]; median: number; max: number};
+
+/**
+ * LATENCY_RUNS runs. Each: copy the source, mutate() it as a chunked upload
+ * in 20 MiB parts, then at once mutate() a small JSON request. Both go to
+ * the server's `slow` mode, so no part finishes during the run. The run
+ * calls resume() first: a paused entry builds no parts. Then both are
+ * cancelled, so the next run has the disk space.
+ */
+export const measureEnqueueLatency = async (
+ host: string,
+): Promise => {
+ const size = await source();
+ // A paused entry builds no parts.
+ await Upload.resume();
+ const nextMs: number[] = [];
+ for (let i = 1; i <= LATENCY_RUNS; i++) {
+ const run = `${Date.now().toString(36)}-${i}`;
+ const file = `${TEST_DIR}/latency-${run}.bin`;
+ const c0 = Date.now();
+ await RNFS.copyFile(SOURCE, file);
+ const copyMs = Date.now() - c0;
+ const t0 = Date.now();
+ const cap = await putChunked20.mutate({
+ host,
+ path: `/slow/chunk/${run}`,
+ file,
+ size,
+ });
+ const t1 = Date.now();
+ const meta = await postJson.mutate({
+ host,
+ path: `/slow/json/${run}`,
+ n: i,
+ text: 'latency',
+ });
+ const t2 = Date.now();
+ nextMs.push(t2 - t1);
+ log(
+ 'action',
+ `latency run ${i}/${LATENCY_RUNS}: copyMs=${copyMs} fileMs=${
+ t1 - t0
+ } nextMs=${t2 - t1}`,
+ );
+ await Upload.cancel(cap.id);
+ await Upload.cancel(meta.id);
+ await remove(file); // the library moved it; this is a guard
+ await waitGone([cap.id, meta.id]);
+ }
+ const sorted = [...nextMs].sort((a, b) => a - b);
+ const median = sorted[Math.floor(sorted.length / 2)];
+ const max = sorted[sorted.length - 1];
+ log(
+ 'action',
+ `latency nextMs median=${median} max=${max} ` +
+ `(target: median < 500, max < 2000) all=${nextMs.join(',')}`,
+ );
+ return {nextMs, median, max};
+};
diff --git a/example/RNBGUExample/harness/uploads.ts b/example/RNBGUExample/harness/uploads.ts
index 45bd84a1..182b9150 100644
--- a/example/RNBGUExample/harness/uploads.ts
+++ b/example/RNBGUExample/harness/uploads.ts
@@ -153,35 +153,55 @@ export const putFile = Upload.define({
export type ChunkedVars = Target & {file: string; size: number};
/**
- * A chunked PUT. The library moves the file at mutate(). Part URLs are
- * `${path}/${n}`, 1-indexed, with a Diana-style Content-Range. A 404 on a
- * part is terminal, as in Diana, so onError could recreate the upload.
+ * A chunked PUT over `ranges`. The library moves the file at mutate(). Part
+ * URLs are `${path}/${n}`, 1-indexed, with a Diana-style Content-Range. A 404
+ * on a part is terminal, as in Diana, so onError could recreate the upload.
*/
+const chunkedRequest = (
+ v: ChunkedVars,
+ ranges: Array<{start: number; end: number}>,
+) => ({
+ method: 'PUT' as const,
+ file: v.file,
+ headers: {'Content-Type': 'application/octet-stream'},
+ parts: ranges.map((range, i) => ({
+ url: `${v.host}${v.path}/${i + 1}`,
+ range,
+ headers: {
+ 'Content-Range': `${range.start}-${range.end - 1}/${v.size}`,
+ },
+ })),
+ accept: ALREADY_DONE,
+ retry: {terminalHttp: {exempt: []}},
+ ...common(v),
+});
+
+/** A chunked PUT in 5 MiB parts. */
export const putChunked = Upload.define({
key: 'harness.chunked',
- request: (v: ChunkedVars) => {
- const ranges = Upload.chunkPlan(v.size, {min: PART_BYTES, max: PART_BYTES});
- return {
- method: 'PUT',
- file: v.file,
- headers: {'Content-Type': 'application/octet-stream'},
- parts: ranges.map((range, i) => ({
- url: `${v.host}${v.path}/${i + 1}`,
- range,
- headers: {
- 'Content-Range': `${range.start}-${range.end - 1}/${v.size}`,
- },
- })),
- accept: ALREADY_DONE,
- retry: {terminalHttp: {exempt: []}},
- ...common(v),
- };
- },
+ request: (v: ChunkedVars) =>
+ chunkedRequest(
+ v,
+ Upload.chunkPlan(v.size, {min: PART_BYTES, max: PART_BYTES}),
+ ),
onSuccess: (raw, v, meta) =>
outcome('ok', meta, `${v.path} chunked status=${raw.status ?? 'none'}`),
onError: (e, v, meta) => outcome('error', meta, errorText(e, v)),
});
+/**
+ * A chunked PUT in Diana's parts: 20 MiB, at least 8 MiB. A 2 GiB file gives
+ * 103 parts. The enqueue latency run uses it.
+ */
+export const putChunked20 = Upload.define({
+ key: 'harness.chunked20',
+ request: (v: ChunkedVars) =>
+ chunkedRequest(v, Upload.chunkPlan(v.size, {min: 8 * MB, max: 20 * MB})),
+ onSuccess: (raw, v, meta) =>
+ outcome('ok', meta, `${v.path} chunked20 status=${raw.status ?? 'none'}`),
+ onError: (e, v, meta) => outcome('error', meta, errorText(e, v)),
+});
+
/**
* A bodiless DELETE. `data: undefined` is spelled out on purpose: Diana's
* generated requests pass it that way.
diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift
index ee5520b8..4e18d48d 100644
--- a/ios/ChunkedCoordinator.swift
+++ b/ios/ChunkedCoordinator.swift
@@ -10,6 +10,11 @@ import Foundation
/// each enqueued part to the key of the one task that owns it. Only
/// enqueueParts creates part tasks.
///
+/// refill builds the part tasks one slice at a time (about 32 MiB of parts),
+/// each slice in its own block on the queue. Another call (the next
+/// enqueue, a delegate callback) waits for at most one slice per entry that
+/// is building, not for whole plans.
+///
/// The entry (QueueEntry) is the durable truth: parts, accepted flags,
/// incarnation. Terminals go through QueueCoordinator.settle.
final class ChunkedCoordinator {
@@ -37,6 +42,21 @@ final class ChunkedCoordinator {
// as nextAttemptAt.
private var partBeginAt: [String: [Int: Double]] = [:]
+ /// About how many part bytes one block on the queue copies.
+ static let defaultSliceBytes: Int64 = 32 << 20
+ /// A test hook: a small budget builds a plan in many slices.
+ var sliceBytes = ChunkedCoordinator.defaultSliceBytes
+ /// Ids with a slice block queued. At most one block per id: the block
+ /// takes its id out, then refills, which queues the next block when parts
+ /// are left. While it is not empty, the coordinator holds the background
+ /// completion handler (QueueCoordinator.whenIdle).
+ private(set) var queuedSlices: Set = []
+ /// A test hook: where a slice block goes. nil puts it on the queue.
+ var dispatchSlice: ((@escaping () -> Void) -> Void)?
+ /// id -> background-created part tasks still to replace, in part order.
+ /// replace() takes one slice of them per block.
+ private var replacing: [String: [Replaced]] = [:]
+
init(_ coordinator: QueueCoordinator) {
q = coordinator
}
@@ -57,6 +77,7 @@ final class ChunkedCoordinator {
/// Forgets the part tasks. The caller cancels them.
func stop(_ id: String) {
+ replacing[id] = nil
inFlight[id] = nil
partSent[id] = nil
partBeginAt[id] = nil
@@ -281,42 +302,82 @@ final class ChunkedCoordinator {
/// superseded, so their callbacks produce no outcome. If one landed before
/// the cancel took effect, partCompleted accepts the part and stops its
/// new task.
+ ///
+ /// It works one slice at a time, like refill. Each old task is cancelled
+ /// in the slice that replaces it, so a part keeps its old task until its
+ /// new one exists, unless that slice's build fails: then the part waits
+ /// for the refill after the backoff. Only while the app is active.
func replace(_ id: String, _ replaced: [Replaced]) {
- guard let e = q.index.entry(id), e.isChunked, e.state == .running,
- !q.settings.isPaused(e.key) else { return }
- let stale = replaced.filter {
+ var queued = replacing[id] ?? []
+ let keys = Set(queued.map(\.key))
+ queued += replaced.filter { !keys.contains($0.key) }
+ replacing[id] = queued.sorted { $0.part < $1.part }
+ replaceSlice(id)
+ updateWait(id)
+ }
+
+ /// Replaces the first slice of `replacing[id]` that is still stale, then
+ /// queues a block for the rest (and for any part with no task).
+ private func replaceSlice(_ id: String) {
+ // A task made now would be a background task too: keep the old ones
+ // and their place in the system's queue. Continue the refill plan.
+ guard q.activity.isActive else {
+ replacing[id] = nil
+ refill(id)
+ return
+ }
+ guard let e = runnable(id) else {
+ replacing[id] = nil
+ return
+ }
+ // Checked again in every slice: an old task the daemon started while
+ // it waited for its slice is sending, so it is kept, not replaced.
+ let stale = (replacing[id] ?? []).filter {
$0.incarnation == e.incarnation && e.parts.indices.contains($0.part) && !e.parts[$0.part].accepted
&& inFlight[id]?[$0.part] == $0.key
+ && q.liveTasks[$0.key].map { $0.task.isLive && $0.task.bytesSent == 0 && !$0.task.hasResponse } == true
}
- guard !stale.isEmpty else { return }
- let requests: [(index: Int, delayMs: Int?)] = stale.map { (index: $0.part, delayMs: remainingMs(id, $0.part)) }
- q.cancelTasks(keys: stale.map(\.key), purpose: .superseded)
- for r in stale {
+ let slice = ChunkedEngine.slice(stale.map(\.part), budget: sliceBytes) { e.parts[$0].size }
+ let now = Array(stale.prefix(slice.count))
+ replacing[id] = stale.count > now.count ? Array(stale.dropFirst(now.count)) : nil
+ // None left to replace (each got a new task meanwhile): the block still
+ // owes the refill plan its next slice.
+ guard !now.isEmpty else {
+ refill(id)
+ return
+ }
+ // A new task keeps what is left of the old one's backoff.
+ let requests: [(index: Int, delayMs: Int?)] = now.map { (index: $0.part, delayMs: remainingMs(id, $0.part)) }
+ q.cancelTasks(keys: now.map(\.key), purpose: .superseded)
+ for r in now {
inFlight[id]?[r.part] = nil
partSent[id]?[r.part] = nil
}
- // On false a part got no task: enqueueParts settled the entry or
- // scheduled a refill, which gives the part a task. partBeginAt still
- // holds the begin date of each part left over, so the refill keeps
- // what is left of its backoff.
- _ = enqueueParts(id, requests)
- updateWait(id)
+ let built = Signpost.interval("part replace slice", "\(id) parts \(now.count) of \(stale.count)") {
+ enqueueParts(id, requests)
+ }
+ // On false the entry settled, or a refill after a backoff follows and
+ // gives the parts of this slice their tasks. The old tasks of the rest
+ // keep running; the next activation replaces them.
+ guard built else {
+ replacing[id] = nil
+ return
+ }
+ refillNextSlice(id)
}
- /// Gives every pending part with no task one. Called at start, after every
- /// part completion, and at the end of every reconcile. After start, it
- /// only replaces a task that ended without its part being accepted (a
- /// lost task, or one that could not be built).
- func refill(_ id: String) {
+ /// The entry when its parts may get tasks now. Otherwise nil, and an
+ /// entry that is done, expired or has a short blob is settled.
+ private func runnable(_ id: String) -> QueueEntry? {
guard q.ready, let e = q.index.entry(id), e.isChunked, e.state == .running,
- !q.settings.isPaused(e.key) else { return }
+ !q.settings.isPaused(e.key) else { return nil }
if e.allAccepted {
q.settle(id, .completed(RawResponseRecord(bodyTruncated: false)))
- return
+ return nil
}
if q.now() >= e.expiresAt {
q.settle(id, .expired)
- return
+ return nil
}
// A blob shorter than a part can never finish: report it now.
let blobSize = e.bodyPath.flatMap { FileIO.size(q.store.fileURL(id, $0)) } ?? 0
@@ -324,12 +385,60 @@ final class ChunkedCoordinator {
q.settle(id, .fileError(
"source blob is \(blobSize) bytes; part \(short) needs "
+ "[\(e.parts[short].start), \(e.parts[short].end))", partIndex: short))
- return
+ return nil
}
+ return e
+ }
+
+ /// Gives every pending part with no task one, a slice at a time. Called at
+ /// start, after every part completion, at the end of every reconcile, and
+ /// from each slice block. After start, it only replaces a task that ended
+ /// without its part being accepted (a lost task, or one that could not be
+ /// built).
+ ///
+ /// Each call checks the entry again and computes "pending minus in
+ /// flight" again. So a part never gets a second task, and a cancel, pause,
+ /// replace, settle or expiry between two slices stops the build.
+ func refill(_ id: String) {
+ guard let e = runnable(id) else { return }
let flight = Set((inFlight[id] ?? [:]).keys)
let indexes = ChunkedEngine.indexesToEnqueue(pending: e.pendingIndexes(), inFlight: flight)
- let requests: [(index: Int, delayMs: Int?)] = indexes.map { (index: $0, delayMs: remainingMs(id, $0)) }
- _ = enqueueParts(id, requests) // settled or deferred inside
+ guard !indexes.isEmpty else { return }
+ let slice = ChunkedEngine.slice(indexes, budget: sliceBytes) { e.parts[$0].size }
+ let requests: [(index: Int, delayMs: Int?)] = slice.map { (index: $0, delayMs: remainingMs(id, $0)) }
+ let built = Signpost.interval("part slice", "\(id) parts \(slice.count) of \(indexes.count)") {
+ enqueueParts(id, requests)
+ }
+ // On false the entry settled, or a refill after a backoff follows.
+ if built && slice.count < indexes.count { refillNextSlice(id) }
+ }
+
+ /// Queues one block for `id`, unless one is queued already. The block
+ /// replaces the next slice of background-created tasks when any are
+ /// left, else it refills. The blocks of other calls already queued run
+ /// first.
+ private func refillNextSlice(_ id: String) {
+ guard queuedSlices.insert(id).inserted else { return }
+ let block = { [weak self] in
+ guard let self else { return }
+ // Before the id leaves queuedSlices, this releases only when a
+ // background wake is short on time: then the completion handler goes
+ // back to the system now. The rest is built at the next wake or
+ // launch, which usually comes because the entry's first slice has
+ // tasks. (A first slice that could not be built has only an
+ // in-process timer.)
+ self.q.releaseIfIdle()
+ self.queuedSlices.remove(id)
+ if self.replacing[id] != nil {
+ self.replaceSlice(id)
+ } else {
+ self.refill(id)
+ }
+ // A later slice may start parts while every earlier one waits.
+ self.updateWait(id)
+ self.q.releaseIfIdle()
+ }
+ if let dispatchSlice { dispatchSlice(block) } else { q.queue.async(execute: block) }
}
/// What is left of a part's begin date, or nil once it has passed.
@@ -348,11 +457,12 @@ final class ChunkedCoordinator {
/// Part tasks for `requests`, in order. Write-ahead: every part file is
/// built, then one save of the entry carries every new attempt ordinal,
- /// before any task exists. Each TaskMap entry is written before its task
- /// resumes. A part that cannot be built stops the batch: a terminal
- /// settles the entry with no task made, and a transient failure keeps the
- /// tasks of the parts before it and refills after a backoff. Returns false
- /// when the entry settled, or a part got no task and a refill follows.
+ /// before any task exists. The tasks are created suspended, their TaskMap
+ /// entries go in one write, and then they resume. A part that cannot be
+ /// built stops the batch: a terminal settles the entry with no task made,
+ /// and a transient failure keeps the tasks of the parts before it and
+ /// refills after a backoff. Returns false when the entry settled, or a
+ /// part got no task and a refill follows.
private func enqueueParts(_ id: String, _ requests: [(index: Int, delayMs: Int?)]) -> Bool {
guard var e = q.index.entry(id) else { return false }
let blob = e.bodyPath ?? ChunkedManifestV9.blobName
@@ -393,6 +503,8 @@ final class ChunkedCoordinator {
refillLater(e, part: e.parts[first.index], delayMs: first.delayMs)
return false
}
+ var made: [(index: Int, delayMs: Int?, task: UploadTask, meta: TaskMap.Meta)] = []
+ var unopened: (index: Int, delayMs: Int?, error: Error)?
for (offset, b) in built.enumerated() {
let part = e.parts[b.index]
let requestId = UUID().uuidString
@@ -400,34 +512,44 @@ final class ChunkedCoordinator {
id: id, partIndex: b.index, incarnation: e.incarnation, attempt: base + offset + 1,
requestId: requestId, headerGeneration: q.settings.headerGeneration,
generation: e.generation, purpose: .attempt, foreground: q.activity.isActive)
- let task: UploadTask
do {
- task = try q.transport.upload(
+ let task = try q.transport.upload(
q.buildRequest(e, url: b.url, requestId: requestId, partHeaders: part.headers),
fromFile: b.file, wifiOnly: q.settings.wifiOnly(e.wifiOnly),
description: ChunkedEngine.taskDescription(id: id, part: b.index, incarnation: e.incarnation),
- beginAt: b.delayMs.map { Date(timeIntervalSince1970: (q.now() + Double($0)) / 1000) },
- beforeResume: { key in self.q.taskMap.set(meta, forKey: key) })
+ beginAt: b.delayMs.map { Date(timeIntervalSince1970: (q.now() + Double($0)) / 1000) })
+ made.append((index: b.index, delayMs: b.delayMs, task: task, meta: meta))
} catch {
- // The session could not open the part file. As for a failed part
- // file build: a short blob can never succeed; otherwise try again
- // later. The ordinals saved for the parts left are not reused.
- let blobSize = FileIO.size(q.store.fileURL(id, blob)) ?? 0
- if blobSize < part.end {
- q.settle(id, .fileError("cannot read part \(b.index): \(error.localizedDescription)",
- partIndex: b.index))
- } else {
- refillLater(e, part: part, delayMs: b.delayMs)
- }
- return false
+ unopened = (index: b.index, delayMs: b.delayMs, error: error)
+ break
+ }
+ }
+ // One TaskMap write for the batch, before any of its tasks resumes.
+ q.taskMap.set(made.map { (key: $0.task.key, meta: $0.meta) })
+ for m in made {
+ inFlight[id, default: [:]][m.index] = m.task.key
+ q.liveTasks[m.task.key] = (id, m.task)
+ if let delayMs = m.delayMs {
+ partBeginAt[id, default: [:]][m.index] = q.now() + Double(delayMs)
+ } else {
+ partBeginAt[id]?[m.index] = nil
}
- inFlight[id, default: [:]][b.index] = task.key
- q.liveTasks[task.key] = (id, task)
- if let delayMs = b.delayMs {
- partBeginAt[id, default: [:]][b.index] = q.now() + Double(delayMs)
+ m.task.resume()
+ }
+ if let unopened {
+ // The session could not open the part file. As for a failed part
+ // file build: a short blob can never succeed; otherwise try again
+ // later. The tasks made before it keep running. The ordinals saved
+ // for the parts left are not reused.
+ let part = e.parts[unopened.index]
+ let blobSize = FileIO.size(q.store.fileURL(id, blob)) ?? 0
+ if blobSize < part.end {
+ q.settle(id, .fileError("cannot read part \(unopened.index): \(unopened.error.localizedDescription)",
+ partIndex: unopened.index))
} else {
- partBeginAt[id]?[b.index] = nil
+ refillLater(e, part: part, delayMs: unopened.delayMs)
}
+ return false
}
}
if case .later(let part, let delayMs) = failure {
diff --git a/ios/ChunkedEngine.swift b/ios/ChunkedEngine.swift
index 22cecf6b..195aa31e 100644
--- a/ios/ChunkedEngine.swift
+++ b/ios/ChunkedEngine.swift
@@ -6,16 +6,32 @@ import Foundation
// place. ChunkedCoordinator owns the session side.
enum ChunkedEngine {
- /// The part indexes to enqueue now: every pending (not accepted) part not
- /// already in flight. Every part gets its task at once, so the tasks are
- /// created while the caller is in mutate(), usually in the foreground. iOS
- /// starts a task created in the background at its own discretion. It never
+ /// The part indexes that need a task: every pending (not accepted) part
+ /// not already in flight. The coordinator starts creating every one of
+ /// them at once, a slice at a time (see slice), so the tasks are created
+ /// right after the caller's mutate(), usually in the foreground. iOS starts
+ /// a task created in the background at its own discretion. It never
/// returns an index in `inFlight`: the one-task-per-part invariant. A part
/// waiting out a backoff is in flight (its delayed task holds the slot).
static func indexesToEnqueue(pending: [Int], inFlight: Set) -> [Int] {
pending.filter { !inFlight.contains($0) }
}
+ /// The first parts of `indexes`, in order, until their bytes reach
+ /// `budget`. Always at least one part. One slice is what one block on the
+ /// coordinator queue builds, so another call waits for at most one slice
+ /// per entry that is building.
+ static func slice(_ indexes: [Int], budget: Int64, size: (Int) -> Int64) -> [Int] {
+ var bytes: Int64 = 0
+ var out: [Int] = []
+ for i in indexes {
+ if !out.isEmpty && bytes >= budget { break }
+ out.append(i)
+ bytes += size(i)
+ }
+ return out
+ }
+
// MARK: - Task identity
// A task carries its owner through the daemon in taskDescription: a prefix
diff --git a/ios/Package.swift b/ios/Package.swift
index a1c8da2e..5824b135 100644
--- a/ios/Package.swift
+++ b/ios/Package.swift
@@ -38,6 +38,7 @@ let package = Package(
"QueueStore.swift",
"RequestIndex.swift",
"RetryClassifier.swift",
+ "Signpost.swift",
"TaskMap.swift",
"Transport.swift",
"UploadOutcome.swift",
diff --git a/ios/QueueCoordinator+Enqueue.swift b/ios/QueueCoordinator+Enqueue.swift
index a5027f3b..ba293b5d 100644
--- a/ios/QueueCoordinator+Enqueue.swift
+++ b/ios/QueueCoordinator+Enqueue.swift
@@ -8,6 +8,10 @@ extension QueueCoordinator {
/// Returns the id, and whether to issue it after the resolve.
func enqueueLocked(_ raw: [String: Any]) throws -> (id: String, issue: Bool) {
+ try Signpost.interval("enqueue", raw["id"] as? String ?? "") { try enqueueRaw(raw) }
+ }
+
+ private func enqueueRaw(_ raw: [String: Any]) throws -> (id: String, issue: Bool) {
let p: ParsedEnqueue
do {
p = try EnqueueParser.parse(raw)
diff --git a/ios/QueueCoordinator+Reconcile.swift b/ios/QueueCoordinator+Reconcile.swift
index 51647822..b9d3252b 100644
--- a/ios/QueueCoordinator+Reconcile.swift
+++ b/ios/QueueCoordinator+Reconcile.swift
@@ -8,21 +8,30 @@ extension QueueCoordinator {
/// daemon may still replay before it re-issues.
static let graceMs = 10_000
- /// `completion` runs on the queue once every entry has its tasks again and
- /// every grace wait has ended. The caller holds the background completion
- /// handlers until then, so the system cannot suspend the app before the
- /// refill enqueues new tasks, or in the middle of a grace wait.
+ /// `completion` runs on the queue once every entry has its tasks again:
+ /// every grace wait has ended and every queued part slice has run. The
+ /// caller holds the background completion handlers until then, so the
+ /// system cannot suspend the app before the refill enqueues new tasks, or
+ /// in the middle of a grace wait.
func reconcileAll(completion: @escaping () -> Void) {
queue.async {
self.transport.allTasks { tasks in
self.queue.async {
self.reconcile(tasks)
- self.whenGraceEnds(completion)
+ self.whenIdleLocked(completion)
}
}
}
}
+ /// Runs `block` on the queue once no grace wait is open and no part slice
+ /// is queued. urlSessionDidFinishEvents holds the session's completion
+ /// handler until then: the completions it just delivered may have queued
+ /// part slices.
+ func whenIdle(_ block: @escaping () -> Void) {
+ queue.async { self.whenIdleLocked(block) }
+ }
+
/// One open grace wait: the TaskMap keys whose completion may still
/// replay, and what to do when the wait ends.
struct Grace {
@@ -31,15 +40,36 @@ extension QueueCoordinator {
let resolve: () -> Void
}
- /// Runs `block` now when no grace wait is open, or when the last one ends.
- func whenGraceEnds(_ block: @escaping () -> Void) {
- if graces.isEmpty {
+ /// Runs `block` now when no grace wait is open and no part slice is
+ /// queued. Otherwise it runs when the last of them ends.
+ func whenIdleLocked(_ block: @escaping () -> Void) {
+ if isIdle {
block()
} else {
afterGrace.append(block)
}
}
+ /// No grace wait is open, and no part slice is queued or the background
+ /// wake is short on time. A grace wait is never cut short.
+ private var isIdle: Bool {
+ graces.isEmpty && (chunked.queuedSlices.isEmpty || wakeIsShort)
+ }
+
+ private var wakeIsShort: Bool {
+ guard !activity.isActive, let remaining = wakeTimeRemaining?() else { return false }
+ return remaining < Self.wakeReserveSeconds
+ }
+
+ /// Runs the held blocks when nothing holds them any more. Called when a
+ /// grace wait ends and before and after each part slice.
+ func releaseIfIdle() {
+ guard !afterGrace.isEmpty, isIdle else { return }
+ let blocks = afterGrace
+ afterGrace = []
+ blocks.forEach { $0() }
+ }
+
/// Opens a grace wait for `keys`. It ends when the completion of every key
/// was handled, or after `graceMs`, whichever comes first. Then `resolve`
/// runs. Does nothing when a wait with this name is open.
@@ -66,10 +96,7 @@ extension QueueCoordinator {
guard let grace = graces[name], grace.token == token else { return }
graces[name] = nil
grace.resolve()
- guard graces.isEmpty else { return }
- let blocks = afterGrace
- afterGrace = []
- blocks.forEach { $0() }
+ releaseIfIdle()
}
func reconcile(_ tasks: [UploadTask]) {
@@ -123,6 +150,13 @@ extension QueueCoordinator {
// A crash between the settings save and the entry saves of a pause or
// resume leaves rows that do not match the gates. Match them now.
applyPauseGates()
+ // A crash between a TaskMap write and resume() leaves a task suspended.
+ // Resume every adopted task still held, after the settles and pauses
+ // above, so a task they stopped sends nothing. resume() on a running
+ // task does nothing.
+ for task in tasks where task.isLive && liveTasks[task.key] != nil {
+ task.resume()
+ }
// A launch into the foreground: replace the adopted tasks that were
// created in the background, and any this reconcile created before the
// app became active.
diff --git a/ios/QueueCoordinator+Simple.swift b/ios/QueueCoordinator+Simple.swift
index e9ab3a6f..032a119a 100644
--- a/ios/QueueCoordinator+Simple.swift
+++ b/ios/QueueCoordinator+Simple.swift
@@ -19,6 +19,12 @@ extension QueueCoordinator {
/// cancelled before it sent a body byte (replaceBackgroundTasks). It keeps
/// the ordinal and request id whether or not the attempt was waiting.
func issue(_ id: String, delayMs: Int? = nil, advanceAttempt: Bool = true, replacing: Bool = false) {
+ Signpost.interval("issue", id) {
+ issueLocked(id, delayMs: delayMs, advanceAttempt: advanceAttempt, replacing: replacing)
+ }
+ }
+
+ private func issueLocked(_ id: String, delayMs: Int?, advanceAttempt: Bool, replacing: Bool) {
guard var e = index.entry(id), !e.legacy, e.state == .queued,
!settings.isPaused(e.key) else { return }
let t = now()
@@ -88,8 +94,7 @@ extension QueueCoordinator {
buildRequest(e, url: url, requestId: requestId), fromFile: body,
wifiOnly: settings.wifiOnly(e.wifiOnly),
description: ChunkedEngine.taskDescription(id: id, attempt: e.attempts, generation: e.generation),
- beginAt: beginAt.map { Date(timeIntervalSince1970: $0 / 1000) },
- beforeResume: { key in self.taskMap.set(meta, forKey: key) })
+ beginAt: beginAt.map { Date(timeIntervalSince1970: $0 / 1000) })
} catch {
// The session could not open the body. Gone since the check above: it
// can never send. Still there: it is unreadable for now (data
@@ -105,6 +110,8 @@ extension QueueCoordinator {
deferIssue(waiting, delayMs: delayMs)
return
}
+ taskMap.set(meta, forKey: task.key)
+ task.resume()
liveTasks[task.key] = (id, task)
attemptStarts[id] = beginAt ?? t
throttle.reset(id)
diff --git a/ios/QueueCoordinator.swift b/ios/QueueCoordinator.swift
index b1bd7491..b1391b59 100644
--- a/ios/QueueCoordinator.swift
+++ b/ios/QueueCoordinator.swift
@@ -95,10 +95,17 @@ final class QueueCoordinator {
/// Bumped by scheduleReplace: only the latest scheduled replace runs.
var replaceGeneration = 0
/// Open grace waits (a completion that may still replay), by name. While
- /// any is open, `afterGrace` holds the background completion handler
- /// release.
+ /// any is open, or a part slice is queued, `afterGrace` holds the
+ /// background completion handler release (whenIdle).
var graces: [String: Grace] = [:]
var afterGrace: [() -> Void] = []
+ /// Seconds left in a background wake (UIApplication's
+ /// backgroundTimeRemaining). When it drops under wakeReserveSeconds, the
+ /// held blocks stop waiting for queued part slices. nil: no limit (tests,
+ /// and the SPM target, which has no UIKit). Read on the queue, only while
+ /// a held block waits for slices.
+ var wakeTimeRemaining: (() -> TimeInterval)?
+ static let wakeReserveSeconds: TimeInterval = 5
/// Settle outcomes (never a user cancel) whose journal write failed, by
/// eventId. They were emitted live; a timer retries the write while the
/// entry still names them.
diff --git a/ios/QueueStore.swift b/ios/QueueStore.swift
index 048911e6..b082db63 100644
--- a/ios/QueueStore.swift
+++ b/ios/QueueStore.swift
@@ -454,11 +454,17 @@ final class QueueStore {
var remaining = end - start
while remaining > 0 {
let chunk = Int(min(remaining, 1 << 20))
- guard let data = try reader.read(upToCount: chunk), !data.isEmpty else {
- throw FileIO.IOError(message: "short read building part \(index)")
+ // One pool per 1 MiB read, so each read buffer is freed before the
+ // next. This block runs inside queue.sync, on the caller's thread,
+ // so a queue-level autoreleaseFrequency would not cover it.
+ let copied = try autoreleasepool { () throws -> Int in
+ guard let data = try reader.read(upToCount: chunk), !data.isEmpty else {
+ throw FileIO.IOError(message: "short read building part \(index)")
+ }
+ try writer.write(contentsOf: data)
+ return data.count
}
- try writer.write(contentsOf: data)
- remaining -= Int64(data.count)
+ remaining -= Int64(copied)
}
try FileIO.rename(tmp, onto: dest)
return dest
diff --git a/ios/RNBackgroundUpload.swift b/ios/RNBackgroundUpload.swift
index c42cf8a0..02f8376d 100644
--- a/ios/RNBackgroundUpload.swift
+++ b/ios/RNBackgroundUpload.swift
@@ -51,11 +51,12 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate {
// AppDelegate stores the system completion handler here per session id.
private static let bgHandlerLock = NSLock()
private static var bgCompletionHandlers: [String: () -> Void] = [:]
- // While a relaunch reconcile runs (deferrals > 0), urlSessionDidFinishEvents
- // must not hand the system its handler: the system could suspend the app
- // before the refill enqueues the next tasks, leaving zero daemon tasks and
- // no future wake. A session that finishes its events in that window parks
- // its id here; the release drains it.
+ // While a relaunch reconcile runs or part slices are queued (deferrals > 0),
+ // the system must not get its handler: it could suspend the app before the
+ // refill enqueues the next tasks, leaving parts with no daemon task and no
+ // future wake. urlSessionDidFinishEvents parks its session id here and
+ // holds a deferral until the coordinator is idle; the last release drains
+ // the parked ids.
private static var bgHandlerDeferrals = 0
private static var bgSessionsAwaitingDrain: Set = []
@@ -69,6 +70,13 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate {
coordinator = QueueCoordinator(
store: .shared, journal: .shared, taskMap: .shared, transport: transport, sink: sink)
super.init()
+ // UIApplication is main-thread only. The coordinator queue may wait on
+ // main here: main never waits on the coordinator queue.
+ coordinator.wakeTimeRemaining = {
+ Thread.isMainThread
+ ? UIApplication.shared.backgroundTimeRemaining
+ : DispatchQueue.main.sync { UIApplication.shared.backgroundTimeRemaining }
+ }
transport.createSessions(delegate: self)
observeAppState()
// Claim the completion-handler deferral BEFORE the reconcile is queued.
@@ -289,16 +297,14 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate {
public func urlSessionDidFinishEvents(forBackgroundURLSession session: URLSession) {
guard let identifier = session.configuration.identifier else { return }
+ // The completions just delivered may have queued part slices, and a
+ // relaunch reconcile may be in flight. Park the id and hold one more
+ // deferral until the coordinator is idle. The last release drains the id.
RNBackgroundUpload.bgHandlerLock.lock()
- guard RNBackgroundUpload.bgHandlerDeferrals <= 0 else {
- // A relaunch reconcile is in flight. Park the id; the release drains it.
- RNBackgroundUpload.bgSessionsAwaitingDrain.insert(identifier)
- RNBackgroundUpload.bgHandlerLock.unlock()
- return
- }
- let handler = RNBackgroundUpload.bgCompletionHandlers.removeValue(forKey: identifier)
+ RNBackgroundUpload.bgHandlerDeferrals += 1
+ RNBackgroundUpload.bgSessionsAwaitingDrain.insert(identifier)
RNBackgroundUpload.bgHandlerLock.unlock()
- if let handler { DispatchQueue.main.async { handler() } }
+ coordinator.whenIdle { RNBackgroundUpload.releaseBackgroundCompletionHandlers() }
}
// MARK: - App state
@@ -362,7 +368,7 @@ private final class SessionTransport: Transport {
}
func upload(_ request: URLRequest, fromFile file: URL, wifiOnly: Bool, description: String,
- beginAt: Date?, beforeResume: (String) -> Void) throws -> UploadTask {
+ beginAt: Date?) throws -> UploadTask {
let session = self.session(wifiOnly: wifiOnly)
// A background session uploads from a file only. When it cannot read the
// file it raises an Objective-C exception, not an error. Swift cannot
@@ -378,10 +384,8 @@ private final class SessionTransport: Transport {
let task = created!
task.taskDescription = description
if let beginAt { task.earliestBeginDate = beginAt }
- let handle = SessionTask(session: session, task: task)
- beforeResume(handle.key)
- task.resume()
- return handle
+ // Not resumed: the caller writes the TaskMap entry first.
+ return SessionTask(session: session, task: task)
}
func allTasks(_ completion: @escaping ([UploadTask]) -> Void) {
@@ -439,5 +443,6 @@ private final class SessionTask: UploadTask {
var beginAt: Date? { task.earliestBeginDate }
var bytesSent: Int64 { task.countOfBytesSent }
var hasResponse: Bool { task.response != nil }
+ func resume() { task.resume() }
func cancel() { task.cancel() }
}
diff --git a/ios/Signpost.swift b/ios/Signpost.swift
new file mode 100644
index 00000000..38c0d556
--- /dev/null
+++ b/ios/Signpost.swift
@@ -0,0 +1,19 @@
+import Foundation
+import os
+
+/// os_signpost intervals around the work on the coordinator queue: each
+/// enqueue, each issue, and each slice of part tasks. Instruments shows them
+/// under Points of Interest, with the entry id and the part count. They
+/// cost almost nothing when no tool records them.
+enum Signpost {
+ static let subsystem = "ai.openspace.rnbgupload"
+ private static let poster = OSSignposter(subsystem: subsystem, category: .pointsOfInterest)
+
+ /// Runs `body` inside the interval `name`. `detail` says what it works on.
+ /// It is logged as public, so entry ids must stay free of user content.
+ static func interval(_ name: StaticString, _ detail: String, _ body: () throws -> T) rethrows -> T {
+ let state = poster.beginInterval(name, id: poster.makeSignpostID(), "\(detail, privacy: .public)")
+ defer { poster.endInterval(name, state) }
+ return try body()
+ }
+}
diff --git a/ios/TaskMap.swift b/ios/TaskMap.swift
index 91e264d6..47f1f80b 100644
--- a/ios/TaskMap.swift
+++ b/ios/TaskMap.swift
@@ -94,6 +94,19 @@ final class TaskMap {
}
}
+ /// set for many tasks, in one write of the file.
+ func set(_ entries: [(key: String, meta: Meta)]) {
+ guard !entries.isEmpty else { return }
+ queue.sync {
+ for (key, meta) in entries { cache[key] = meta }
+ write()
+ }
+ }
+
+ /// How many times this instance wrote the file. For tests.
+ var writeCount: Int { queue.sync { writes } }
+ private var writes = 0
+
func meta(forKey key: String) -> Meta? {
queue.sync { cache[key] }
}
@@ -146,6 +159,7 @@ final class TaskMap {
// Caller holds `queue`.
private func write() {
+ writes += 1
guard let data = try? JSONEncoder().encode(cache) else { return }
do {
try FileIO.writeAtomically(data, to: fileURL)
diff --git a/ios/Tests/CoordinatorChunkedTests.swift b/ios/Tests/CoordinatorChunkedTests.swift
index 13bcd591..bcce1172 100644
--- a/ios/Tests/CoordinatorChunkedTests.swift
+++ b/ios/Tests/CoordinatorChunkedTests.swift
@@ -505,4 +505,483 @@ final class CoordinatorChunkedTests: XCTestCase {
let recreated = fresh.transport.created.compactMap { ChunkedEngine.parsePartDescription($0.taskDescription)?.part }
XCTAssertEqual(recreated.sorted(), [0, 1], "every pending part gets its task again")
}
+ // MARK: - Part slices
+
+ /// The part indexes of `tasks`, sorted.
+ private func parts(_ tasks: [FakeTask]) -> [Int] {
+ tasks.compactMap { ChunkedEngine.parsePartDescription($0.taskDescription)?.part }.sorted()
+ }
+
+ private func partTasks(_ incarnation: String? = nil) -> [FakeTask] {
+ h.transport.created.filter {
+ guard let p = ChunkedEngine.parsePartDescription($0.taskDescription) else { return false }
+ return incarnation == nil || p.incarnation == incarnation
+ }
+ }
+
+ private func partFiles(_ id: String) -> [String] {
+ ((try? FileManager.default.contentsOfDirectory(atPath: h.store.dir(id).path)) ?? [])
+ .filter { $0.hasPrefix("part-") }.sorted()
+ }
+
+ // 6 parts of 10 bytes and a 20-byte budget: 3 slices of 2 parts.
+ private func enqueueSixPartsInThreeSlices(id: String = "cap", extra: [String: Any] = [:]) throws {
+ h.sliceBytes = 20
+ _ = try h.enqueue(h.chunkedRaw(id: id, size: 60, parts: 6, extra: extra)).get()
+ }
+
+ func testAnEnqueueAfterAChunkedEnqueueResolvesBeforeEveryPartHasATask() throws {
+ h.sliceBytes = 20
+ var partTasksAtResolve: Int?
+ // As Diana does: the metadata mutate() right after the file mutate() resolves.
+ h.coordinator.enqueue(h.chunkedRaw(id: "cap", size: 60, parts: 6), resolve: { [unowned self] _ in
+ self.h.coordinator.enqueue(self.h.dataRaw(id: "meta"), resolve: { [unowned self] _ in
+ partTasksAtResolve = self.partTasks().count
+ }, reject: { _, _ in XCTFail("meta rejected") })
+ }, reject: { _, _ in XCTFail("cap rejected") })
+ h.drainAll()
+ XCTAssertEqual(partTasksAtResolve, 2, "the next enqueue waits for one slice, not the whole plan")
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "then every part gets its task")
+ XCTAssertEqual(h.entry("meta")?.state, .running)
+ }
+
+ func testSlicesBuildInOrderAndEveryPartGetsExactlyOneTask() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1])
+ XCTAssertEqual(h.heldSlices.count, 1, "one continuation per entry")
+ h.runHeldSlice()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3])
+ h.runHeldSlice()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5])
+ XCTAssertTrue(h.heldSlices.isEmpty, "nothing left to build")
+ XCTAssertEqual(h.transport.created.count, 6)
+ XCTAssertTrue(h.transport.created.allSatisfy(\.resumed))
+ let attempts = h.transport.created.compactMap { h.map.meta(forKey: $0.key)?.attempt }
+ XCTAssertEqual(attempts.sorted(), [1, 2, 3, 4, 5, 6], "each task has its own ordinal")
+ XCTAssertEqual(h.store.load("cap")?.attempts, 6)
+ }
+
+ func testAPartThatCompletesBetweenSlicesGetsNoSecondTask() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.complete(try XCTUnwrap(partTask(0)), status: 500) // a delayed retry holds the slot
+ h.complete(try XCTUnwrap(partTask(1))) // accepted: its refill builds the next slice
+ XCTAssertEqual(parts(h.transport.live), [0, 2, 3])
+ XCTAssertEqual(h.heldSlices.count, 1, "the refill did not queue a second continuation")
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 2, 3, 4, 5], "one task per pending part")
+ XCTAssertEqual(parts(partTasks()), [0, 0, 1, 2, 3, 4, 5], "only part 0's retry is a second task")
+ }
+
+ func testCancelBetweenSlicesStopsTheBuild() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.cancel("cap")
+ h.runHeldSlices()
+ XCTAssertEqual(h.transport.created.count, 2, "no task after the cancel")
+ XCTAssertTrue(h.transport.live.isEmpty)
+ XCTAssertEqual(h.sink.settled.last?["kind"] as? String, "cancelled")
+ XCTAssertFalse(partFiles("cap").contains { !$0.hasPrefix("part-0.") && !$0.hasPrefix("part-1.") },
+ "no part file after the cancel: \(partFiles("cap"))")
+ }
+
+ func testPauseBetweenSlicesStopsTheBuildAndResumeFinishesIt() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.pause()
+ h.runHeldSlices()
+ XCTAssertEqual(h.transport.created.count, 2, "a paused entry builds no part")
+ XCTAssertTrue(h.transport.live.isEmpty)
+ h.resume()
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5])
+ }
+
+ func testReplaceBetweenSlicesBuildsOnlyTheNewPlan() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices(extra: ["retry": ["terminalHttp": ["exempt": []]]])
+ let old = try XCTUnwrap(h.entry("cap")?.incarnation)
+ h.complete(try XCTUnwrap(partTask(0)), status: 404)
+ XCTAssertEqual(h.entry("cap")?.state, .error)
+ _ = try h.enqueue(h.chunkedRaw(id: "cap", size: 60, parts: 6, urlPrefix: "https://s3.test/v2-")).get()
+ let new = try XCTUnwrap(h.entry("cap")?.incarnation)
+ XCTAssertNotEqual(new, old)
+ h.runHeldSlices()
+ XCTAssertEqual(parts(partTasks(old)), [0, 1], "the old plan stopped at its first slice")
+ XCTAssertTrue(partTasks(old).allSatisfy { !$0.isLive })
+ XCTAssertEqual(parts(partTasks(new)), [0, 1, 2, 3, 4, 5], "the new plan: one task per part")
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5])
+ }
+
+ func testExpiryBetweenSlicesSettlesAndBuildsNothingMore() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.clock += 15 * 24 * 3_600_000
+ h.runHeldSlices()
+ XCTAssertEqual(h.entry("cap")?.state, .error)
+ XCTAssertEqual((h.sink.settled.last?["error"] as? [String: Any])?["errorKind"] as? String, "expired")
+ XCTAssertEqual(h.transport.created.count, 2)
+ XCTAssertTrue(h.transport.live.isEmpty)
+ }
+
+ /// Slice 1 has tasks and TaskMap keys; the process dies with the rest
+ /// queued. Returns a new process over the same disk, before boot.
+ private func crashAfterTheFirstSlice() throws -> (fresh: Harness, survivors: [FakeTask]) {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ let survivors = h.transport.live.map {
+ FakeTask(key: $0.key, description: $0.taskDescription, request: $0.request, beginAt: $0.beginAt)
+ }
+ XCTAssertEqual(parts(survivors), [0, 1])
+ let fresh = Harness(root: h.root)
+ fresh.sliceBytes = 20
+ fresh.relaunch(daemonTasks: survivors)
+ return (fresh, survivors)
+ }
+
+ func testACrashBetweenSlicesIsRepairedByReconcileAndRefill() throws {
+ let (fresh, survivors) = try crashAfterTheFirstSlice()
+ fresh.boot()
+ fresh.drainAll()
+ XCTAssertEqual(parts(fresh.transport.created), [2, 3, 4, 5], "the parts with no task get one")
+ XCTAssertTrue(survivors.allSatisfy { !$0.cancelled }, "the live tasks are adopted")
+ XCTAssertTrue(survivors.allSatisfy(\.resumed), "a task left suspended by the crash runs")
+ let attempts = (survivors + fresh.transport.created).compactMap { fresh.map.meta(forKey: $0.key)?.attempt }
+ XCTAssertEqual(Set(attempts).count, 6, "no ordinal repeats")
+ }
+
+ func testACrashBeforeTheTaskMapWriteIsRepairedFromTheDescriptions() throws {
+ let (fresh, survivors) = try crashAfterTheFirstSlice()
+ // The process died after creating slice 1's tasks, before its TaskMap
+ // write: the daemon holds them, the map does not name them.
+ let map = TaskMap(fileURL: fresh.map.fileURL)
+ survivors.forEach { map.removeKey($0.key) }
+ fresh.relaunch(daemonTasks: survivors)
+ fresh.boot()
+ fresh.drainAll()
+ XCTAssertTrue(survivors.allSatisfy { !$0.cancelled && $0.resumed }, "adopted by description and resumed")
+ XCTAssertEqual(parts(fresh.transport.created), [2, 3, 4, 5], "parts 0 and 1 get no second task")
+ }
+
+ func testAWakeShortOnTimeReleasesTheHandlerBeforeTheLastSlice() throws {
+ let (fresh, _) = try crashAfterTheFirstSlice()
+ fresh.holdSlices = true
+ var remaining: TimeInterval = 30
+ fresh.coordinator.wakeTimeRemaining = { remaining }
+ var partsAtRelease: [Int]?
+ fresh.coordinator.reconcileAll { partsAtRelease = self.parts(fresh.transport.created) }
+ fresh.drain()
+ fresh.drain()
+ XCTAssertNil(partsAtRelease, "30 s left: the handler waits for the slices")
+ XCTAssertEqual(parts(fresh.transport.created), [2, 3])
+ remaining = 3
+ fresh.runHeldSlice()
+ XCTAssertEqual(partsAtRelease, [2, 3], "released as the next slice starts")
+ XCTAssertEqual(parts(fresh.transport.created), [2, 3, 4, 5], "the slice still builds")
+ }
+
+ func testAnActiveAppIgnoresTheWakeTime() throws {
+ h.holdSlices = true
+ h.coordinator.wakeTimeRemaining = { 0 }
+ h.becomeActive()
+ try enqueueSixPartsInThreeSlices()
+ var idle = false
+ h.coordinator.whenIdle { idle = true }
+ h.drain()
+ XCTAssertFalse(idle)
+ h.runHeldSlices()
+ XCTAssertTrue(idle)
+ }
+
+ func testALaterSliceClearsTheWaitOfEarlierDelayedParts() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.complete(try XCTUnwrap(partTask(0)), status: 503)
+ h.complete(try XCTUnwrap(partTask(1)), status: 503)
+ XCTAssertNotNil(h.row("cap")?["nextAttemptAt"], "every part in flight waits")
+ h.runHeldSlice()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3])
+ XCTAssertNil(h.row("cap")?["nextAttemptAt"], "parts 2 and 3 send now")
+ }
+
+ /// The app becomes active, and the replace delay passes.
+ private func activate() {
+ h.becomeActive()
+ h.advance(Double(QueueCoordinator.replaceDelayMs))
+ }
+
+ private func isForeground(_ task: FakeTask) -> Bool { h.map.meta(forKey: task.key)?.foreground == true }
+
+ func testActivationReplacesBackgroundPartTasksOneSliceAtATime() throws {
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ let old = h.transport.live
+ XCTAssertEqual(parts(old), [0, 1, 2, 3, 4, 5])
+ XCTAssertFalse(old.contains(where: isForeground))
+ h.holdSlices = true
+ activate()
+ XCTAssertEqual(parts(old.filter(\.cancelled)), [0, 1])
+ XCTAssertEqual(parts(old.filter(\.isLive)), [2, 3, 4, 5], "a part keeps its old task until its slice")
+ XCTAssertEqual(h.heldSlices.count, 1)
+ h.runHeldSlice()
+ XCTAssertEqual(parts(old.filter(\.isLive)), [4, 5])
+ h.runHeldSlices()
+ XCTAssertTrue(old.allSatisfy(\.cancelled))
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "one task per part")
+ XCTAssertTrue(h.transport.live.allSatisfy(isForeground))
+ XCTAssertEqual(h.transport.created.count, 12)
+ }
+
+ func testActivationWhileBuildSlicesAreQueued() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ let old = h.transport.live
+ activate()
+ XCTAssertTrue(old.allSatisfy(\.cancelled), "slice 1's background tasks are replaced")
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "one task per part")
+ XCTAssertTrue(h.transport.live.allSatisfy(isForeground), "the later slices run while active")
+ XCTAssertEqual(h.transport.created.count, 8)
+ }
+
+ func testAReplaceSliceWithNothingLeftToReplaceStillContinuesTheBuild() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ h.runHeldSlice() // parts 2 and 3; the block for 4 and 5 waits
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3])
+ activate() // replaces 0 and 1 now; 2 and 3 wait for the queued block
+ // Before that block runs, 2 and 3 fail and their retries hold the slots.
+ h.complete(try XCTUnwrap(partTask(2)), status: 503)
+ h.complete(try XCTUnwrap(partTask(3)), status: 503)
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "parts 4 and 5 get their tasks")
+ }
+
+ func testAnOldTaskThatStartsWhileWaitingForItsSliceIsKept() throws {
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ let old = h.transport.live
+ h.holdSlices = true
+ activate() // replaces 0 and 1; 2 to 5 wait for their slices
+ let sending = try XCTUnwrap(old.first { parts([$0]) == [2] })
+ sending.bytesSent = 4 // the daemon started it meanwhile
+ h.runHeldSlices()
+ XCTAssertFalse(sending.cancelled, "a task that sends is never replaced")
+ XCTAssertEqual(parts(old.filter(\.isLive)), [2])
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "one task per part")
+ XCTAssertEqual(h.transport.created.count, 11)
+ }
+
+ func testASameIdMutateDuringAReplaceChainLeavesTheChain() throws {
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ h.holdSlices = true
+ activate()
+ let created = h.transport.created.count
+ // The same body: a live resume, nothing changes.
+ _ = try h.enqueue(h.chunkedRaw(id: "cap", size: 60, parts: 6, source: h.root.appendingPathComponent("x"))).get()
+ // A different body while running rejects.
+ guard case .failure(let e) = h.enqueue(h.chunkedRaw(id: "cap", size: 60, parts: 6,
+ urlPrefix: "https://other.test/")) else {
+ return XCTFail("a different body was accepted")
+ }
+ XCTAssertEqual(e.code, "E_RUNNING")
+ XCTAssertEqual(h.transport.created.count, created)
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "one task per part")
+ XCTAssertEqual(h.transport.created.count, 12, "the chain replaced every part once")
+ }
+
+ func testASliceThatFailsPartwayIsFinishedByTheBackoffRefill() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ var failed = false
+ h.transport.failUpload = { file in
+ guard !failed, file.lastPathComponent.hasPrefix("part-3.") else { return nil }
+ failed = true
+ return NSError(domain: NSURLErrorDomain, code: NSURLErrorCannotOpenFile)
+ }
+ h.runHeldSlice()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2], "part 2 keeps its task")
+ XCTAssertTrue(h.heldSlices.isEmpty, "no slice is queued; the backoff timer refills")
+ h.advance(4 * 3_600_000)
+ h.runHeldSlices()
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5], "one task per part")
+ XCTAssertEqual(h.transport.created.count, 6)
+ XCTAssertTrue(h.sink.settled.isEmpty)
+ }
+
+ func testASmallRequestWaitsForOneSlicePerChunkedUpload() throws {
+ h.sliceBytes = 20
+ h.holdSlices = true
+ for id in ["a", "b", "c"] { _ = try h.enqueue(h.chunkedRaw(id: id, size: 60, parts: 6)).get() }
+ XCTAssertEqual(partTasks().count, 6, "slice 1 of each")
+ XCTAssertEqual(h.heldSlices.count, 3)
+ // The three next slices go on the queue, then the small request.
+ h.requeueHeldSlices()
+ var partTasksAtResolve: Int?
+ h.coordinator.enqueue(h.dataRaw(id: "meta"), resolve: { [unowned self] _ in
+ partTasksAtResolve = self.partTasks().count
+ }, reject: { _, _ in XCTFail("meta rejected") })
+ h.drain()
+ XCTAssertEqual(partTasksAtResolve, 12, "it waited for one slice of each upload, not for whole plans")
+ h.runHeldSlices()
+ XCTAssertEqual(partTasks().count, 18)
+ for id in ["a", "b", "c"] {
+ let mine = h.transport.live.filter { ChunkedEngine.parsePartDescription($0.taskDescription)?.id == id }
+ XCTAssertEqual(parts(mine), [0, 1, 2, 3, 4, 5])
+ }
+ }
+
+ func testAShortWakeNeverCutsAGraceWaitShort() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ // Part 1's task finished while the app was dead: its key stays, the
+ // daemon no longer holds it, and its completion may still replay.
+ let survivor = try XCTUnwrap(partTask(0))
+ let fresh = Harness(root: h.root)
+ fresh.sliceBytes = 20
+ fresh.holdSlices = true
+ fresh.relaunch(daemonTasks: [FakeTask(key: survivor.key, description: survivor.taskDescription,
+ request: survivor.request)])
+ fresh.coordinator.wakeTimeRemaining = { 1 }
+ var released = false
+ fresh.coordinator.reconcileAll { released = true }
+ fresh.drain()
+ fresh.drain()
+ fresh.runHeldSlices()
+ XCTAssertFalse(released, "the grace wait holds the handler, short wake or not")
+ fresh.advance(Double(QueueCoordinator.graceMs))
+ XCTAssertTrue(released, "released when the grace ends; queued slices do not hold a short wake")
+ }
+
+ func testReplaceStopsWhenTheAppResignsActive() throws {
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ let old = h.transport.live
+ h.holdSlices = true
+ activate()
+ XCTAssertEqual(parts(old.filter(\.cancelled)), [0, 1])
+ h.resignActive()
+ h.runHeldSlices()
+ XCTAssertEqual(parts(old.filter(\.isLive)), [2, 3, 4, 5], "the old tasks keep their place")
+ XCTAssertEqual(h.transport.created.count, 8, "no background task made in their place")
+ XCTAssertEqual(parts(h.transport.live), [0, 1, 2, 3, 4, 5])
+ }
+
+ func testAnAdoptedTaskWhoseEntrySettlesAtLaunchIsNotResumed() throws {
+ let (fresh, survivors) = try crashAfterTheFirstSlice()
+ fresh.clock = h.clock + 15 * 24 * 3_600_000
+ fresh.boot()
+ XCTAssertEqual(fresh.entry("cap")?.state, .error, "expired")
+ XCTAssertTrue(survivors.allSatisfy(\.cancelled))
+ XCTAssertFalse(survivors.contains(where: \.resumed), "a stopped task sends nothing")
+ }
+
+ func testAnExpiredEntryDueForReplacementOnActivationSettlesExpired() throws {
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ let old = h.transport.live
+ h.clock += 15 * 24 * 3_600_000
+ activate()
+ XCTAssertEqual(h.entry("cap")?.state, .error)
+ XCTAssertEqual((h.sink.settled.last?["error"] as? [String: Any])?["errorKind"] as? String, "expired")
+ XCTAssertTrue(old.allSatisfy(\.cancelled))
+ XCTAssertEqual(h.transport.created.count, 6, "no new part task")
+ }
+
+ func testTheBackgroundCompletionHandlerWaitsForQueuedSlices() throws {
+ let (fresh, _) = try crashAfterTheFirstSlice()
+ fresh.holdSlices = true
+ var partsAtRelease: [Int]?
+ fresh.coordinator.reconcileAll { partsAtRelease = self.parts(fresh.transport.created) }
+ fresh.drain()
+ fresh.drain()
+ XCTAssertNil(partsAtRelease, "a slice is still queued")
+ XCTAssertEqual(fresh.heldSlices.count, 1)
+ fresh.runHeldSlices()
+ XCTAssertEqual(partsAtRelease, [2, 3, 4, 5], "released after the last slice")
+ }
+
+ func testWhenIdleWaitsForQueuedSlices() throws {
+ h.holdSlices = true
+ try enqueueSixPartsInThreeSlices()
+ var idle = false
+ h.coordinator.whenIdle { idle = true }
+ h.drain()
+ XCTAssertFalse(idle)
+ h.runHeldSlices()
+ XCTAssertTrue(idle)
+ }
+
+ func testTheTaskMapIsWrittenOncePerSlice() throws {
+ let before = h.map.writeCount
+ try enqueueSixPartsInThreeSlices()
+ h.drainAll()
+ XCTAssertEqual(h.transport.live.count, 6)
+ XCTAssertEqual(h.map.writeCount - before, 3, "one write per slice of 2 tasks")
+ }
+
+ func testEveryTaskMapEntryIsWrittenBeforeItsTaskResumes() throws {
+ var missing: [String] = []
+ h.transport.onResume = { [unowned self] task in
+ if self.h.map.meta(forKey: task.key) == nil { missing.append(task.key) }
+ }
+ try enqueueSixPartsInThreeSlices()
+ _ = try h.enqueue(h.dataRaw(id: "note")).get()
+ h.drainAll()
+ XCTAssertEqual(h.transport.created.count, 7)
+ XCTAssertTrue(h.transport.created.allSatisfy(\.resumed))
+ XCTAssertEqual(missing, [])
+ }
+}
+
+/// `RNBGU_PERF=1 swift test --filter ChunkedPerfTests`. Writes about 1.2 GB
+/// to the temp directory, so it does not run by default.
+final class ChunkedPerfTests: XCTestCase {
+ private static let mib = 1 << 20
+
+ /// The time from the resolve of a 200 MiB chunked enqueue (10 parts of 20
+ /// MiB) to the resolve of the enqueue that follows it at once. Real part
+ /// files, fake sessions: it leaves out nsurlsessiond.
+ private func nextEnqueueMs(sliceBytes: Int64?) throws -> Double {
+ let h = Harness()
+ defer { try? FileManager.default.removeItem(at: h.root) }
+ h.boot()
+ h.sliceBytes = sliceBytes
+ let size = 200 * Self.mib
+ let src = h.root.appendingPathComponent("capture.bin")
+ FileManager.default.createFile(atPath: src.path, contents: nil)
+ let writer = try FileHandle(forWritingTo: src)
+ let block = Data(repeating: 7, count: Self.mib)
+ for _ in 0..<200 { try writer.write(contentsOf: block) }
+ try writer.close()
+
+ var first: UInt64 = 0
+ var next: UInt64 = 0
+ let done = DispatchSemaphore(value: 0)
+ h.coordinator.enqueue(h.chunkedRaw(id: "cap", size: size, parts: 10, source: src), resolve: { _ in
+ first = DispatchTime.now().uptimeNanoseconds
+ h.coordinator.enqueue(h.dataRaw(id: "meta"), resolve: { _ in
+ next = DispatchTime.now().uptimeNanoseconds
+ done.signal()
+ }, reject: { _, _ in XCTFail("meta rejected") })
+ }, reject: { _, _ in XCTFail("cap rejected") })
+ done.wait()
+ h.drainAll()
+ XCTAssertEqual(h.transport.live.count, 11)
+ return Double(next - first) / 1_000_000
+ }
+
+ func testTheNextEnqueueWaitsForOneSliceOfA200MiBPlan() throws {
+ try XCTSkipUnless(ProcessInfo.processInfo.environment["RNBGU_PERF"] == "1", "set RNBGU_PERF=1 to run")
+ func median(_ v: [Double]) -> Double { v.sorted()[v.count / 2] }
+ let sliced = try (0..<3).map { _ in try nextEnqueueMs(sliceBytes: nil) }
+ // Int64.max puts every part in one slice: the build before slices.
+ let whole = try (0..<3).map { _ in try nextEnqueueMs(sliceBytes: .max) }
+ print(String(format: "RNBGU_PERF nextMs: sliced median %.1f ms %@; one block median %.1f ms %@",
+ median(sliced), sliced.description, median(whole), whole.description))
+ XCTAssertLessThan(median(sliced), 500)
+ }
}
diff --git a/ios/Tests/CoordinatorRelaunchTests.swift b/ios/Tests/CoordinatorRelaunchTests.swift
index 2f765198..e0e7fecf 100644
--- a/ios/Tests/CoordinatorRelaunchTests.swift
+++ b/ios/Tests/CoordinatorRelaunchTests.swift
@@ -32,6 +32,7 @@ final class CoordinatorRelaunchTests: XCTestCase {
fresh.relaunch(daemonTasks: [survivor])
fresh.boot()
XCTAssertTrue(fresh.transport.created.isEmpty, "adopted, not re-issued")
+ XCTAssertTrue(survivor.resumed, "a task left suspended by a crash runs")
XCTAssertEqual(fresh.row("a")?["state"] as? String, "running")
fresh.complete(survivor)
XCTAssertEqual(fresh.entry("a")?.state, .completed)
diff --git a/ios/Tests/SupportComponentTests.swift b/ios/Tests/SupportComponentTests.swift
index b9499777..3d3762e4 100644
--- a/ios/Tests/SupportComponentTests.swift
+++ b/ios/Tests/SupportComponentTests.swift
@@ -200,6 +200,21 @@ final class TaskMapTests: XCTestCase {
XCTAssertEqual(reread.meta(forKey: "s:2"), .init(id: "b", purpose: .superseded), "a key with no meta gets one")
}
+ func testSetManyEntriesInOneWrite() {
+ let url = makeTempDir().appendingPathComponent("map.json")
+ let map = TaskMap(fileURL: url)
+ let a = TaskMap.Meta(id: "a", partIndex: 0, incarnation: "i", attempt: 1, purpose: .attempt)
+ let b = TaskMap.Meta(id: "a", partIndex: 1, incarnation: "i", attempt: 2, purpose: .attempt)
+ let before = map.writeCount
+ map.set([(key: "s:1", meta: a), (key: "s:2", meta: b)])
+ XCTAssertEqual(map.writeCount - before, 1)
+ let reread = TaskMap(fileURL: url)
+ XCTAssertEqual(reread.meta(forKey: "s:1"), a)
+ XCTAssertEqual(reread.meta(forKey: "s:2"), b)
+ map.set([])
+ XCTAssertEqual(map.writeCount - before, 1, "nothing to write: no write")
+ }
+
func testV9EntryDecodes() throws {
let url = makeTempDir().appendingPathComponent("map.json")
try Data(#"{"s:9":{"id":"old","acceptStatus":[409]},"s:10":{"id":"new","purpose":"future"}}"#.utf8).write(to: url)
@@ -232,6 +247,18 @@ final class ChunkedEngineTests: XCTestCase {
XCTAssertEqual(ChunkedEngine.indexesToEnqueue(pending: [3, 4], inFlight: [3]), [4], "never an in-flight index")
}
+ func testSliceTakesPartsUntilTheBudgetAndAtLeastOne() {
+ let mib: Int64 = 1 << 20
+ let size: (Int) -> Int64 = { $0 == 4 ? 8 * mib : 20 * mib }
+ XCTAssertEqual(ChunkedEngine.slice([0, 1, 2, 3, 4], budget: 32 * mib, size: size), [0, 1],
+ "20 MiB parts: two reach 32 MiB")
+ XCTAssertEqual(ChunkedEngine.slice([3, 4], budget: 32 * mib, size: size), [3, 4])
+ XCTAssertEqual(ChunkedEngine.slice([2, 3], budget: 20 * mib, size: size), [2], "exactly the budget")
+ XCTAssertEqual(ChunkedEngine.slice([0, 1, 2], budget: 1, size: size), [0], "at least one part")
+ XCTAssertEqual(ChunkedEngine.slice([0, 1, 2], budget: .max, size: size), [0, 1, 2])
+ XCTAssertEqual(ChunkedEngine.slice([], budget: 32 * mib, size: size), [])
+ }
+
func testDescriptionsSurviveHostileIds() {
let id = #"cap:1/"x"\n"#
let part = ChunkedEngine.taskDescription(id: id, part: 7, incarnation: "inc")
diff --git a/ios/Tests/TestSupport.swift b/ios/Tests/TestSupport.swift
index 1d1c7559..cabbf11a 100644
--- a/ios/Tests/TestSupport.swift
+++ b/ios/Tests/TestSupport.swift
@@ -63,6 +63,8 @@ final class FakeTask: UploadTask {
var bytesSent: Int64 = 0
var hasResponse = false
private(set) var cancelled = false
+ private(set) var resumed = false
+ var onResume: ((FakeTask) -> Void)?
let request: URLRequest
let file: URL?
let beginAt: Date?
@@ -83,6 +85,11 @@ final class FakeTask: UploadTask {
isLive = false
}
+ func resume() {
+ resumed = true
+ onResume?(self)
+ }
+
func header(_ name: String) -> String? { request.value(forHTTPHeaderField: name) }
}
@@ -100,14 +107,16 @@ final class FakeTransport: Transport {
/// When set, upload calls it and throws what it returns: the session could
/// not open the file. It may delete the file first, to race the check.
var failUpload: ((URL) -> Error?)?
+ /// Called as each created task resumes.
+ var onResume: ((FakeTask) -> Void)?
func upload(_ request: URLRequest, fromFile file: URL, wifiOnly: Bool, description: String,
- beginAt: Date?, beforeResume: (String) -> Void) throws -> UploadTask {
+ beginAt: Date?) throws -> UploadTask {
if let error = failUpload?(file) { throw error }
let task = FakeTask(key: "\(wifiOnly ? "wifi" : "any"):\(Self.next)", description: description,
request: request, file: file, beginAt: beginAt, wifiOnly: wifiOnly)
Self.next += 1
- beforeResume(task.key)
+ task.onResume = { [weak self] in self?.onResume?($0) }
created.append(task)
return task
}
@@ -156,6 +165,18 @@ final class Harness {
var timers: [(delayMs: Int, block: () -> Void)] = []
private(set) var coordinator: QueueCoordinator!
let queue = DispatchQueue(label: "test.queue")
+ /// The part-slice budget in bytes. Set it small to build a plan in many
+ /// slices. Kept across relaunch().
+ var sliceBytes: Int64? {
+ didSet { applySliceHooks() }
+ }
+ /// When true, each queued part slice is held here instead of going on the
+ /// queue. Dropping them models a crash between slices. Kept across
+ /// relaunch().
+ var holdSlices = false {
+ didSet { applySliceHooks() }
+ }
+ private(set) var heldSlices: [() -> Void] = []
init(root: URL = makeTempDir()) {
self.root = root
@@ -173,6 +194,36 @@ final class Harness {
store: store, journal: journal, taskMap: TaskMap(fileURL: taskMap.fileURL),
transport: transport, sink: sink, queue: queue, now: { [unowned self] in self.clock },
random: { 0.5 }, schedule: { [unowned self] ms, block in self.timers.append((ms, block)) })
+ heldSlices = []
+ applySliceHooks()
+ }
+
+ private func applySliceHooks() {
+ guard let coordinator else { return }
+ if let sliceBytes { coordinator.chunked.sliceBytes = sliceBytes }
+ coordinator.chunked.dispatchSlice = holdSlices ? { [unowned self] in self.heldSlices.append($0) } : nil
+ }
+
+ /// Runs the oldest held slice on the queue. Returns false when none is held.
+ @discardableResult
+ func runHeldSlice() -> Bool {
+ guard !heldSlices.isEmpty else { return false }
+ let block = heldSlices.removeFirst()
+ queue.sync { block() }
+ return true
+ }
+
+ /// Runs held slices until no slice is held.
+ func runHeldSlices() {
+ while runHeldSlice() {}
+ }
+
+ /// Puts the held slices on the queue, in order, without running them.
+ /// The slices they queue are held again.
+ func requeueHeldSlices() {
+ let blocks = heldSlices
+ heldSlices = []
+ blocks.forEach { queue.async(execute: $0) }
}
var map: TaskMap { coordinator.taskMap }
@@ -186,6 +237,11 @@ final class Harness {
func drain() { queue.sync {} }
+ /// Drains until no part slice is queued: every part build has finished.
+ func drainAll() {
+ repeat { drain() } while queue.sync { !coordinator.chunked.queuedSlices.isEmpty }
+ }
+
/// Fires every timer that is due within `ms` (advancing the clock).
func advance(_ ms: Double) {
clock += ms
diff --git a/ios/Transport.swift b/ios/Transport.swift
index 7ede044a..2c2e9a41 100644
--- a/ios/Transport.swift
+++ b/ios/Transport.swift
@@ -20,16 +20,19 @@ protocol UploadTask: AnyObject {
var bytesSent: Int64 { get }
/// The task has a response: the server answered, whatever bytesSent says.
var hasResponse: Bool { get }
+ /// Starts a task that `Transport.upload` created. Call it only after the
+ /// task's TaskMap entry is written.
+ func resume()
func cancel()
}
protocol Transport: AnyObject {
- /// Creates an upload task, sets its description and begin date, calls
- /// `beforeResume` with its key (the caller writes the TaskMap there), then
- /// resumes it. Throws when the session cannot open `file`: no task exists
- /// then, and `beforeResume` was not called.
+ /// Creates an upload task and sets its description and begin date. It
+ /// does not resume the task: the caller writes the TaskMap entry for its
+ /// key, then calls resume(). So a batch of tasks needs one TaskMap write.
+ /// Throws when the session cannot open `file`: no task exists then.
func upload(_ request: URLRequest, fromFile file: URL, wifiOnly: Bool, description: String,
- beginAt: Date?, beforeResume: (String) -> Void) throws -> UploadTask
+ beginAt: Date?) throws -> UploadTask
/// Every task of both sessions. The completion may run on any queue.
func allTasks(_ completion: @escaping ([UploadTask]) -> Void)