From 56c51c922a0634e818d3dd3062dba4f63fbae33f Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Tue, 10 Nov 2026 11:54:47 -0500 Subject: [PATCH 1/5] Add an enqueue latency run to the example app Diana calls mutate() on a capture file and then at once on its metadata. On iOS the second mutate() can wait while the first one builds its part files. This adds a way to measure that wait on a device. - harness.chunked20: a chunked kind with Diana's parts (20 MiB, at least 8 MiB). A 2 GiB file gives 103 parts. - "Enqueue latency x7": makes a 2 GiB source one time, then 7 runs. Each run copies the source, calls mutate() on it, then at once calls mutate() on a small JSON request, both on the server's slow mode. It logs nextMs, the time the second mutate() waits, and the median and maximum. - The example README has the steps as iOS step 19. Co-Authored-By: Claude Fable 5.1 --- example/RNBGUExample/App.tsx | 11 +++ example/RNBGUExample/README.md | 20 +++- example/RNBGUExample/__tests__/App.test.tsx | 40 ++++++++ example/RNBGUExample/harness/files.ts | 14 ++- example/RNBGUExample/harness/latency.ts | 101 ++++++++++++++++++++ example/RNBGUExample/harness/uploads.ts | 62 ++++++++---- 6 files changed, 223 insertions(+), 25 deletions(-) create mode 100644 example/RNBGUExample/harness/latency.ts 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,41 @@ 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; + 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); +}); 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..e49d2cf6 --- /dev/null +++ b/example/RNBGUExample/harness/latency.ts @@ -0,0 +1,101 @@ +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). */ +const waitGone = async (ids: string[]) => { + const until = Date.now() + 60_000; + while ( + Upload.getRequests().some(r => ids.includes(r.id)) && + Date.now() < until + ) { + 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 queue is + * not paused: 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(); + 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. From c2650f7a74472ef6a6db1677d6c0cc7edcc3dd60 Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Tue, 10 Nov 2026 11:55:06 -0500 Subject: [PATCH 2/5] Build iOS part tasks in slices so the next mutate() does not wait Since 11.0.0 a chunked mutate() copied every part out of the file and created every part task in one block on the coordinator queue. A mutate() called right after it waited for all of that. For a 2 GiB capture in 20 MiB parts that is 103 part copies, 103 task map writes with fsync and 103 task creations, which can pass the 10 s enqueue timeout on an older iPhone (Sentry DIANA-11QD; out-of-memory crashes DIANA-3QVM, DIANA-3S18). - refill builds the parts in slices of about 32 MiB. Each slice is its own block on the queue, so other calls run between two slices. One continuation per entry. Each slice checks the entry again and computes "pending minus in flight" again, so a part never gets two tasks, and a cancel, pause, replace or expiry between slices stops the build. - A slice creates its tasks suspended, writes their task map entries in one write, then resumes them. Transport.upload no longer resumes. - The background completion handler waits until no slice is queued, in a relaunch reconcile and in urlSessionDidFinishEvents. - os_signpost intervals for enqueue, issue and each slice. - replace() of background-created part tasks goes through refill, so it is sliced too. - README: "starts creating" a task for every part. CHANGELOG: Unreleased entry, and the 11.0.1 line no longer says the task map was batched. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 17 +- README.md | 11 +- ios/ChunkedCoordinator.swift | 126 +++++++++---- ios/ChunkedEngine.swift | 23 ++- ios/Package.swift | 1 + ios/QueueCoordinator+Enqueue.swift | 4 + ios/QueueCoordinator+Reconcile.swift | 42 +++-- ios/QueueCoordinator+Simple.swift | 11 +- ios/QueueCoordinator.swift | 4 +- ios/RNBackgroundUpload.swift | 34 ++-- ios/Signpost.swift | 18 ++ ios/TaskMap.swift | 14 ++ ios/Tests/CoordinatorChunkedTests.swift | 238 ++++++++++++++++++++++++ ios/Tests/SupportComponentTests.swift | 27 +++ ios/Tests/TestSupport.swift | 52 +++++- ios/Transport.swift | 13 +- 16 files changed, 549 insertions(+), 86 deletions(-) create mode 100644 ios/Signpost.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index 008df890..9a3af0c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ ## Unreleased +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. Each slice writes the +task map once; before, every part task wrote the whole task map. 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. + +iOS adds `os_signpost` intervals for each enqueue, issue and part slice, so +Instruments can show how long each one holds the queue. + iOS no longer deletes a chunked upload's file when `enqueue` cannot save the entry, for example on a full disk. The library had already moved the caller's file into its queue, so this deleted the only copy. The file now @@ -41,8 +54,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 2709ebf5..717ebbb0 100644 --- a/README.md +++ b/README.md @@ -226,10 +226,17 @@ 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 one slice, +not for the whole file. 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 diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift index e4b9c2be..1e88aeec 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 one slice at most, not for the +/// whole plan. +/// /// The entry (QueueEntry) is the durable truth: parts, accepted flags, /// incarnation. Terminals go through QueueCoordinator.settle. final class ChunkedCoordinator { @@ -37,6 +42,18 @@ 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)? + init(_ coordinator: QueueCoordinator) { q = coordinator } @@ -275,24 +292,27 @@ final class ChunkedCoordinator { && inFlight[id]?[$0.part] == $0.key } 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 { 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) + // The parts have no task now: refill gives them one, a slice at a time. + // partBeginAt still holds the begin date of each one, so the new task + // keeps what is left of its backoff. + refill(id) updateWait(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). + /// 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 q.ready, let e = q.index.entry(id), e.isChunked, e.state == .running, !q.settings.isPaused(e.key) else { return } @@ -314,8 +334,27 @@ final class ChunkedCoordinator { } 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 that refills `id`, unless one is queued already. 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 } + self.queuedSlices.remove(id) + self.refill(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. @@ -334,11 +373,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 @@ -379,6 +419,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 @@ -386,34 +428,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 } - 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) + } + // 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]?[b.index] = nil + partBeginAt[id]?[m.index] = nil } + 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 { + 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..aa035bca 100644 --- a/ios/ChunkedEngine.swift +++ b/ios/ChunkedEngine.swift @@ -6,16 +6,31 @@ 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 one slice at most. + 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 ab3711e2..33682d96 100644 --- a/ios/Package.swift +++ b/ios/Package.swift @@ -37,6 +37,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 2566222a..c8718ac1 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..aaf99407 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,27 @@ 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) } } + private var isIdle: Bool { graces.isEmpty && chunked.queuedSlices.isEmpty } + + /// Runs the held blocks when nothing holds them any more. Called when a + /// grace wait ends and after each part slice. + func releaseIfIdle() { + guard isIdle, !afterGrace.isEmpty 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 +87,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]) { diff --git a/ios/QueueCoordinator+Simple.swift b/ios/QueueCoordinator+Simple.swift index 571b9dba..cfed34b9 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) throttle.reset(id) } diff --git a/ios/QueueCoordinator.swift b/ios/QueueCoordinator.swift index dcc46db8..83409f49 100644 --- a/ios/QueueCoordinator.swift +++ b/ios/QueueCoordinator.swift @@ -92,8 +92,8 @@ 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] = [] /// Settle outcomes (never a user cancel) whose journal write failed, by diff --git a/ios/RNBackgroundUpload.swift b/ios/RNBackgroundUpload.swift index f716bcd3..df5f5d25 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 = [] @@ -288,16 +289,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 @@ -361,7 +360,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 @@ -377,10 +376,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) { @@ -438,5 +435,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..4f86d5b9 --- /dev/null +++ b/ios/Signpost.swift @@ -0,0 +1,18 @@ +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. + 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 ed944253..ba2f4d94 100644 --- a/ios/Tests/CoordinatorChunkedTests.swift +++ b/ios/Tests/CoordinatorChunkedTests.swift @@ -430,4 +430,242 @@ 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") + let attempts = (survivors + fresh.transport.created).compactMap { fresh.map.meta(forKey: $0.key)?.attempt } + XCTAssertEqual(Set(attempts).count, 6, "no ordinal repeats") + } + + 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/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 296f2a15..e4c3c7fd 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,28 @@ 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() {} } var map: TaskMap { coordinator.taskMap } @@ -186,6 +229,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 3d99606f..e8459bc5 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) From 81963719a9750d7ed6ff5d81396d31dae54a1bad Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Tue, 10 Nov 2026 12:11:38 -0500 Subject: [PATCH 3/5] Resume adopted tasks and slice part replacement on iOS Fixes from the review of the part slices. - Reconcile resumes every task it adopts, part and simple. A crash between a slice's task map write and its resumes left suspended tasks that held their parts and never sent. resume() on a running task does nothing. - Each slice block runs updateWait after its work, so a row no longer shows a wait while parts of a later slice send. - A background wake stops waiting for queued slices when less than 5 s of background time is left. RNBackgroundUpload passes backgroundTimeRemaining to the coordinator as a closure; grace waits are never cut short. - replace() works one slice at a time and cancels each old task in the slice that replaces it, so a part keeps its old task until its new one exists. It settles an entry that is done, expired or has a short blob, as refill does. - Tests: adopted tasks resume (with and without task map keys, and a simple request), the wait refresh, the wake time cap, activation replace over several slices and while build slices are queued, and the expiry settle on activation. - Docs: the wake trade-off in the README, background-tagged later slices, CHANGELOG lines, the example's before/after commits, and the harness calls resume() and reports when waitGone gives up. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 13 ++- README.md | 12 ++- example/RNBGUExample/README.md | 8 +- example/RNBGUExample/__tests__/App.test.tsx | 2 + example/RNBGUExample/harness/latency.ts | 28 +++-- ios/ChunkedCoordinator.swift | 111 +++++++++++++++----- ios/QueueCoordinator+Reconcile.swift | 18 +++- ios/QueueCoordinator.swift | 7 ++ ios/RNBackgroundUpload.swift | 7 ++ ios/Signpost.swift | 1 + ios/Tests/CoordinatorChunkedTests.swift | 107 +++++++++++++++++++ ios/Tests/CoordinatorRelaunchTests.swift | 1 + 12 files changed, 271 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a3af0c7..34fb8a48 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,18 @@ 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. Each slice writes the task map once; before, every part task wrote the whole task map. 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. +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. + +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. diff --git a/README.md b/README.md index 717ebbb0..1f2caa7c 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,14 @@ 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. Every chunked upload already has +the tasks of its first slice, so a next wake comes. + # Usage The library owns a durable queue of HTTP requests: JSON bodies, multipart @@ -242,7 +250,9 @@ Interest), with the entry id and the part count. 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/README.md b/example/RNBGUExample/README.md index efa8f1d8..946596e1 100644 --- a/example/RNBGUExample/README.md +++ b/example/RNBGUExample/README.md @@ -309,8 +309,8 @@ step says otherwise. `mutate()`, then at once the metadata `mutate()`. Use an older device, such as an iPhone 11: the slow case came from one. Each run needs about 4 GiB free (the 2 GiB source copy and its part files), plus 2 GiB for the - source that the app keeps. Do not pause the queue: a paused entry builds - no parts. Press "Enqueue latency x7". The first press makes the 2 GiB + source that the app keeps. A paused entry builds no parts, so the run + calls `resume()` first. Press "Enqueue latency x7". The first press makes the 2 GiB source, which takes minutes. Then each of the 7 runs copies the source, calls `mutate()` on it with 103 parts of 20 MiB, and at once calls `mutate()` on a small JSON request. Both go to `slow`, so no part @@ -318,7 +318,9 @@ step says otherwise. `copyMs`, `fileMs` and `nextMs`. `nextMs` is how long the second `mutate()` waits. The last line gives the median and the maximum of `nextMs`. Pass: median under 500 ms and maximum under 2 s. To compare, - run it on the same device with a build of 11.1.0. To see where the time + run it on the same device twice: "before" is a build of commit 56c51c9 + (this button on top of 11.1.0's native code), "after" is a build of the + commit that adds the part slices or later. To see where the time goes, record the app in Instruments with the os_signpost or Points of Interest instrument: each `enqueue`, `issue` and `part slice` is an interval. diff --git a/example/RNBGUExample/__tests__/App.test.tsx b/example/RNBGUExample/__tests__/App.test.tsx index f86670d2..e4d49f8c 100644 --- a/example/RNBGUExample/__tests__/App.test.tsx +++ b/example/RNBGUExample/__tests__/App.test.tsx @@ -125,6 +125,7 @@ it('runs the enqueue latency measurement 7 times on /slow', async () => { 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++) { @@ -151,4 +152,5 @@ it('runs the enqueue latency measurement 7 times on /slow', async () => { 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/latency.ts b/example/RNBGUExample/harness/latency.ts index e49d2cf6..3bda828d 100644 --- a/example/RNBGUExample/harness/latency.ts +++ b/example/RNBGUExample/harness/latency.ts @@ -31,13 +31,23 @@ const source = async (): Promise => { const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)); -/** Waits until native forgets `ids` (cancel, then the automatic ack). */ +/** + * 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; - while ( - Upload.getRequests().some(r => ids.includes(r.id)) && - Date.now() < until - ) { + 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); } }; @@ -47,14 +57,16 @@ 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 queue is - * not paused: a paused entry builds no parts. Then both are cancelled, so - * the next run has the disk space. + * 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}`; diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift index 1e88aeec..63979655 100644 --- a/ios/ChunkedCoordinator.swift +++ b/ios/ChunkedCoordinator.swift @@ -53,6 +53,9 @@ final class ChunkedCoordinator { 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 @@ -74,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 @@ -93,6 +97,9 @@ final class ChunkedCoordinator { if keep { live[t.part] = t.task.key q.liveTasks[t.task.key] = (id, t.task) + // A crash between a slice's TaskMap write and its resumes leaves + // the task suspended. resume() on a running task does nothing. + t.task.resume() if let begin = t.task.beginAt.map({ $0.timeIntervalSince1970 * 1000 }), begin > q.now() { partBeginAt[id, default: [:]][t.part] = begin } @@ -284,45 +291,66 @@ 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. 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) { + guard let e = runnable(id) else { + replacing[id] = nil + return + } + 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 } - guard !stale.isEmpty else { return } - 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 + guard !now.isEmpty else { 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 } - // The parts have no task now: refill gives them one, a slice at a time. - // partBeginAt still holds the begin date of each one, so the new task - // keeps what is left of its backoff. - refill(id) - 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, 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) { + /// 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 @@ -330,8 +358,22 @@ 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) guard !indexes.isEmpty else { return } @@ -344,14 +386,27 @@ final class ChunkedCoordinator { if built && slice.count < indexes.count { refillNextSlice(id) } } - /// Queues one block that refills `id`, unless one is queued already. The - /// blocks of other calls already queued run first. + /// 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. Every entry has at least its first slice's + // tasks, so a later wake comes. + self.q.releaseIfIdle() self.queuedSlices.remove(id) - self.refill(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) } diff --git a/ios/QueueCoordinator+Reconcile.swift b/ios/QueueCoordinator+Reconcile.swift index aaf99407..98c47ada 100644 --- a/ios/QueueCoordinator+Reconcile.swift +++ b/ios/QueueCoordinator+Reconcile.swift @@ -50,12 +50,21 @@ extension QueueCoordinator { } } - private var isIdle: Bool { graces.isEmpty && chunked.queuedSlices.isEmpty } + /// 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 after each part slice. + /// grace wait ends and before and after each part slice. func releaseIfIdle() { - guard isIdle, !afterGrace.isEmpty else { return } + guard !afterGrace.isEmpty, isIdle else { return } let blocks = afterGrace afterGrace = [] blocks.forEach { $0() } @@ -118,6 +127,9 @@ extension QueueCoordinator { && entry?.attempts == attempt && !simpleLive.contains(id): simpleLive.insert(id) liveTasks[task.key] = (id, task) + // A crash between the TaskMap write and resume() leaves the task + // suspended. resume() on a running task does nothing. + task.resume() default: // No v10 owner (a v9 task), an older generation or attempt, a // duplicate, or an entry that must have no task. Drop its callback. diff --git a/ios/QueueCoordinator.swift b/ios/QueueCoordinator.swift index 83409f49..01be4bf8 100644 --- a/ios/QueueCoordinator.swift +++ b/ios/QueueCoordinator.swift @@ -96,6 +96,13 @@ final class QueueCoordinator { /// 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/RNBackgroundUpload.swift b/ios/RNBackgroundUpload.swift index df5f5d25..2af91306 100644 --- a/ios/RNBackgroundUpload.swift +++ b/ios/RNBackgroundUpload.swift @@ -70,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. diff --git a/ios/Signpost.swift b/ios/Signpost.swift index 4f86d5b9..38c0d556 100644 --- a/ios/Signpost.swift +++ b/ios/Signpost.swift @@ -10,6 +10,7 @@ enum Signpost { 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) } diff --git a/ios/Tests/CoordinatorChunkedTests.swift b/ios/Tests/CoordinatorChunkedTests.swift index ba2f4d94..710695a1 100644 --- a/ios/Tests/CoordinatorChunkedTests.swift +++ b/ios/Tests/CoordinatorChunkedTests.swift @@ -571,10 +571,117 @@ final class CoordinatorChunkedTests: XCTestCase { 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 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 diff --git a/ios/Tests/CoordinatorRelaunchTests.swift b/ios/Tests/CoordinatorRelaunchTests.swift index c9148f5d..40a75cde 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) From 998fef8d4bc51c64e7b844f49e42f0d1f546ceb2 Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Tue, 10 Nov 2026 12:24:33 -0500 Subject: [PATCH 4/5] Keep the refill chain and the foreground rule in iOS part replacement Fixes from the second review of the part slices. - A replace slice block with nothing left to replace (each part got a retry task meanwhile) now refills. Before, it queued nothing, so the later parts of a refill plan got no task until a part was accepted. - replaceSlice stops when the app is not active: it drops the rest of the replacement and refills. Before, after a resign, it cancelled background tasks only to make new background tasks. - Reconcile resumes the adopted tasks at its end, after the settles and pause gates, so a task they stopped sends nothing. - The replace doc comment and the CHANGELOG say a part keeps its old task unless its slice's build fails. - Tests for the three fixes. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 4 ++- ios/ChunkedCoordinator.swift | 20 ++++++++++---- ios/QueueCoordinator+Reconcile.swift | 10 ++++--- ios/Tests/CoordinatorChunkedTests.swift | 36 +++++++++++++++++++++++++ 4 files changed, 61 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 34fb8a48..89b49117 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,9 @@ 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. +way, and each part keeps its old task until its new one exists, unless +that slice's build fails. 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 diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift index 63979655..3f7141ec 100644 --- a/ios/ChunkedCoordinator.swift +++ b/ios/ChunkedCoordinator.swift @@ -97,9 +97,6 @@ final class ChunkedCoordinator { if keep { live[t.part] = t.task.key q.liveTasks[t.task.key] = (id, t.task) - // A crash between a slice's TaskMap write and its resumes leaves - // the task suspended. resume() on a running task does nothing. - t.task.resume() if let begin = t.task.beginAt.map({ $0.timeIntervalSince1970 * 1000 }), begin > q.now() { partBeginAt[id, default: [:]][t.part] = begin } @@ -294,7 +291,8 @@ final class ChunkedCoordinator { /// /// 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. + /// 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]) { var queued = replacing[id] ?? [] let keys = Set(queued.map(\.key)) @@ -307,6 +305,13 @@ final class ChunkedCoordinator { /// 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 @@ -318,7 +323,12 @@ final class ChunkedCoordinator { 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 - guard !now.isEmpty else { return } + // 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) diff --git a/ios/QueueCoordinator+Reconcile.swift b/ios/QueueCoordinator+Reconcile.swift index 98c47ada..b9d3252b 100644 --- a/ios/QueueCoordinator+Reconcile.swift +++ b/ios/QueueCoordinator+Reconcile.swift @@ -127,9 +127,6 @@ extension QueueCoordinator { && entry?.attempts == attempt && !simpleLive.contains(id): simpleLive.insert(id) liveTasks[task.key] = (id, task) - // A crash between the TaskMap write and resume() leaves the task - // suspended. resume() on a running task does nothing. - task.resume() default: // No v10 owner (a v9 task), an older generation or attempt, a // duplicate, or an entry that must have no task. Drop its callback. @@ -153,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/Tests/CoordinatorChunkedTests.swift b/ios/Tests/CoordinatorChunkedTests.swift index 710695a1..8ac1a8f0 100644 --- a/ios/Tests/CoordinatorChunkedTests.swift +++ b/ios/Tests/CoordinatorChunkedTests.swift @@ -670,6 +670,42 @@ final class CoordinatorChunkedTests: XCTestCase { 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 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() From 95844b113e4322f9e15c09a38a982bf7a0c95083 Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Tue, 10 Nov 2026 15:05:46 -0500 Subject: [PATCH 5/5] Keep a sending part task during iOS replacement and free copy buffers Fixes from the pre-merge review of the part slices. - replaceSlice checks again, in every slice, that each old task is live, has sent no body byte and has no response. Before, only replaceBackgroundTasks checked it, when the parts were queued, so a task the daemon started while waiting for its slice could be cancelled mid-upload, with a short window of two PUTs of one part. Such a task is now kept. - The part copy wraps each 1 MiB read in autoreleasepool, so each read buffer is freed before the next. The copy runs inside store.queue.sync on the caller's thread, so a queue autoreleaseFrequency would not cover it. - README and code comments: a mutate() waits for at most one slice per chunked upload that is building, and a next wake is usual but not certain when no slice of an upload could be built. - Tests: a sending old task is kept; a slice that fails partway and the backoff refill; three uploads slicing at once with a small request behind them; a same-id mutate() during a replace chain; a short wake while a grace wait is open. Co-Authored-By: Claude Fable 5.2 --- CHANGELOG.md | 11 ++- README.md | 11 ++- ios/ChunkedCoordinator.swift | 13 +++- ios/ChunkedEngine.swift | 3 +- ios/QueueStore.swift | 14 +++- ios/Tests/CoordinatorChunkedTests.swift | 98 +++++++++++++++++++++++++ ios/Tests/TestSupport.swift | 8 ++ 7 files changed, 141 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 89b49117..1ef3bd51 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,15 +5,18 @@ 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. Each slice writes the -task map once; before, every part task wrote the whole task map. The parts +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. Replacing stops when the app leaves the -foreground again. +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 diff --git a/README.md b/README.md index 1f2caa7c..94d87705 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,11 @@ 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. Every chunked upload already has -the tasks of its first slice, so a next wake comes. +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 @@ -237,8 +240,8 @@ requests: over HTTP/2, every queued part can be in flight at once. **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 one slice, -not for the whole file. Each part task reads its own part file, so a chunked +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 diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift index 3f7141ec..d8f0bc66 100644 --- a/ios/ChunkedCoordinator.swift +++ b/ios/ChunkedCoordinator.swift @@ -12,8 +12,8 @@ import Foundation /// /// 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 one slice at most, not for the -/// whole plan. +/// 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. @@ -316,9 +316,12 @@ final class ChunkedCoordinator { 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 } let slice = ChunkedEngine.slice(stale.map(\.part), budget: sliceBytes) { e.parts[$0].size } let now = Array(stale.prefix(slice.count)) @@ -406,8 +409,10 @@ final class ChunkedCoordinator { 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. Every entry has at least its first slice's - // tasks, so a later wake comes. + // 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 { diff --git a/ios/ChunkedEngine.swift b/ios/ChunkedEngine.swift index aa035bca..195aa31e 100644 --- a/ios/ChunkedEngine.swift +++ b/ios/ChunkedEngine.swift @@ -19,7 +19,8 @@ enum ChunkedEngine { /// 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 one slice at most. + /// 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] = [] diff --git a/ios/QueueStore.swift b/ios/QueueStore.swift index 3d3382eb..821aa797 100644 --- a/ios/QueueStore.swift +++ b/ios/QueueStore.swift @@ -232,11 +232,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/Tests/CoordinatorChunkedTests.swift b/ios/Tests/CoordinatorChunkedTests.swift index 8ac1a8f0..61be09b9 100644 --- a/ios/Tests/CoordinatorChunkedTests.swift +++ b/ios/Tests/CoordinatorChunkedTests.swift @@ -683,6 +683,104 @@ final class CoordinatorChunkedTests: XCTestCase { 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() diff --git a/ios/Tests/TestSupport.swift b/ios/Tests/TestSupport.swift index e4c3c7fd..8d656e7e 100644 --- a/ios/Tests/TestSupport.swift +++ b/ios/Tests/TestSupport.swift @@ -218,6 +218,14 @@ final class Harness { 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 } /// Runs the relaunch reconcile to completion.