From 6523157eec70159f5b1b3d5725fdf7009a0a5350 Mon Sep 17 00:00:00 2001 From: Dylan Murphy Date: Mon, 24 Aug 2026 16:20:19 -0400 Subject: [PATCH] iOS chunked-upload engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per docs/design/chunked-uploads.md, the counterpart to the Android slice with identical JS-visible semantics. startChunkedUpload renames the source into a library-owned directory (Application Support/RNFileUploaderChunked//blob) and persists a durable Codable manifest before any task is enqueued; re-calls reconcile — same parts resume (stored headers/expiresAt replaced, accepted parts skipped), a different parts array recreates per the design's rule. The manifest carries a persisted `stalled` flag instead of Android's it-just-isn't-scheduled, because iOS reconciles every upload on relaunch and must not auto-resume one that journaled a terminal; it also stores no absolute source path (the app container moves between launches — the blob location is derived from the id). The coordinator keeps a sliding window of 3 part tasks enqueued with the daemon per upload (the design's liveness runway for a dead app), never two for one part index: inFlight maps each part to the task key that owns it, every transition runs on one serial queue, only refill creates tasks, and a reconcile token discards stale daemon snapshots so overlapping reconciles can't double-enqueue. Background sessions only upload from files, so each enqueued part gets a temp file of exactly its byte range (FileHandle range copy, tmp+rename), deleted when the part finishes and rebuilt on retry — transient disk stays window x partSize. Part completions evaluate the accept rules against the manifest, persist the flag, emit byte-weighted aggregate progress, and refill; transient failures (network, 5xx, system cancels) back off exponentially with expiresAt as the only cap; a non-accepted response out of its 3 per-part retries journals an 'error' with the part detail and stalls. Relaunch reconciliation runs whenever the sessions are recreated (module init and the AppDelegate background-wake hook both land in `shared`'s init): enumerate the daemon's tasks, match them to manifests via taskDescription with the TaskMap entry as durable fallback, rebuild the window, sweep orphaned temp files. The existing setBackgroundSessionCompletionHandler flow is untouched. Bytes are deleted only when a 'completed' event is acked; cancelUpload journals one user 'cancelled' and keeps manifest+bytes; removeUpload releases both with no event (its cancellations are suppressed for simple uploads too, matching Android). Accept rules also replace the dead acceptStatus parsing for simple uploads. httpMaximumConnectionsPerHost stays 1 — the library-wide cap of 4 belongs to the hardening slice. Review fixes: - Simple-upload idempotency had a check-then-act race: the existence check enumerates session tasks asynchronously, so two concurrent same-id calls could both pass it and enqueue duplicates. check-and-create is now serialized per id — the first caller claims the id synchronously before dispatching enumeration; concurrent callers park their resolve and are answered with the id once the task lands. - The manifest now carries an incarnation token, persisted and encoded into each part task's identity (taskDescription and TaskMap). A late delegate callback whose token doesn't match the stored manifest's is dropped, so a removed-then-recreated upload can never have an old incarnation's part completion written into the new manifest; relaunch reconciliation cancels rather than adopts mismatched tasks. - Recreate semantics implemented to match Android's ChunkedManifest.reconcile: a different parts array is accepted only while the upload is not running (stalled or expired) — keep the blob, replace parts/headers/accept/expiresAt, reset every part to unsent, require the new ranges to tile [0, blob size), reject while running; path stays ignored whenever a manifest exists. The incarnation rotates on recreate. - Part temp files are no longer trusted by size alone on relaunch: the filename encodes the incarnation and byte range that produced the file, so a stale copy from a prior plan is rebuilt instead of reused, and stale siblings are swept when a part is next materialized. - A blob shorter than a part's range now stalls with a terminal 'file' error on the first refill instead of surfacing only when the window reached the short part (Android is aligned to the same rule in its slice). - startChunkedUpload documents its byte-safety ordering: every rejection-type validation precedes the source move, and the one post-move failure (the manifest save) leaves the blob adoptable by a retry with the same id. Final review fixes: - Relaunch refill no longer races the background completion handler: replayed part completions run unowned and never refill, so reconcileAll's async chain was the only refill — and urlSessionDidFinishEvents could hand the system its completion handler (and the app its suspension) before a single new part task was enqueued, likeliest exactly when every enqueued part finished while the app was dead (zero daemon tasks, no future wake, silent stall). The coordinator now claims the pending completion handlers when reconciliation begins and releases them only after every upload's post-reconcile refill has resumed its tasks; a didFinishEvents in between parks its session id and is drained at release. - The per-part non-transient HTTP retry budget is persisted in the manifest instead of in-memory (where every process death reset it, letting a deterministic 4xx re-upload its part once per wake until expiresAt with no terminal ever journaled), and an unowned non-transient HTTP failure with no live replacement task now counts toward it, so the budget can trip across wakes. Resume and recreate still grant a fresh budget — both rebuild parts from the incoming call. - clearState prunes expiryArmed (a removed-then-recreated id could not arm its possibly-earlier deadline until the stale timer fired) and the coordinator's lastProgressAt entry (unbounded growth). - TaskMap.Meta decodes the legacy acceptStatus:[Int] field and maps it to {status} accept rules, mirroring Android's Upload.normalized(), so a task persisted by an earlier build keeps its accept rules when it completes under this one. - First-call (create) startChunkedUpload validates that the parts tile [0, blob size) exactly, same as recreate and as Android's validatedForCreate, rejecting after the source move but before the manifest is saved or anything is enqueued — the moved blob stays adoptable by a corrected retry. Co-Authored-By: Claude Fable 5 --- ios/ChunkedCoordinator.swift | 721 +++++++++++++++++++++++++++++++++++ ios/ChunkedEngine.swift | 81 ++++ ios/ChunkedManifest.swift | 404 ++++++++++++++++++++ ios/EventJournal.swift | 9 +- ios/RNBackgroundUpload.swift | 299 +++++++++++++-- ios/RNFileUploader.mm | 18 +- ios/TaskMap.swift | 59 ++- ios/UploadOutcome.swift | 40 ++ 8 files changed, 1584 insertions(+), 47 deletions(-) create mode 100644 ios/ChunkedCoordinator.swift create mode 100644 ios/ChunkedEngine.swift create mode 100644 ios/ChunkedManifest.swift create mode 100644 ios/UploadOutcome.swift diff --git a/ios/ChunkedCoordinator.swift b/ios/ChunkedCoordinator.swift new file mode 100644 index 00000000..20121ed5 --- /dev/null +++ b/ios/ChunkedCoordinator.swift @@ -0,0 +1,721 @@ +import Foundation + +/// Runs chunked uploads against the background sessions. It keeps the sliding +/// window of part tasks enqueued with the daemon. It evaluates the outcome of +/// each part. After a relaunch, it reconciles the durable [ChunkedManifest] +/// with the tasks that the daemon still holds. +/// +/// Every state transition occurs on one serial queue. The queue enforces the +/// invariants that the design marks binding: at most [ChunkedEngine.window] +/// part tasks are enqueued per upload, and never two for the same part index. +/// `inFlight` maps each enqueued part to the task key that owns it. Only +/// refill, on this queue, creates tasks. +final class ChunkedCoordinator { + + // The singleton that owns the background sessions. It outlives this object. + // Both live for the whole process. + private unowned let uploader: RNBackgroundUpload + + private let queue = DispatchQueue(label: "ai.openspace.rnbgupload.chunked") + + // partIndex -> the TaskMap key of the one task that may be in flight for it. + // The key lets us tell a superseded task's late completion (possible around + // a relaunch reconcile) apart from the live task's completion. + private var inFlight: [String: [Int: String]] = [:] + // The uploads whose in-flight set we rebuild from the daemon now. Refill is + // blocked until the rebuild lands. Thus a stale snapshot can never + // double-enqueue. The token makes overlapping reconciles safe: only the + // latest reconcile may apply its snapshot. An earlier snapshot could miss + // tasks enqueued after it was taken. To apply it would re-enqueue their + // part indexes. + private var reconcileToken: [String: UUID] = [:] + private var cooldownUntil: [String: [Int: Double]] = [:] // epoch ms + private var transientAttempts: [String: [Int: Int]] = [:] + private var expiryArmed: Set = [] + // The in-flight bytes per part index. They feed the byte-weighted + // aggregate progress. + private var partSent: [String: [Int: Int64]] = [:] + // A cache of the stored manifests, refreshed on every load. The progress + // path reads it. Thus didSendBodyData never touches the disk. + private var manifests: [String: ChunkedManifest] = [:] + + private static let progressThrottle: TimeInterval = 0.5 // seconds, per upload + private let progressLock = NSLock() + private var lastProgressAt: [String: TimeInterval] = [:] + + init(uploader: RNBackgroundUpload) { + self.uploader = uploader + } + + private func nowMs() -> Double { Date().timeIntervalSince1970 * 1000 } + + // MARK: - Entry points (module methods) + + /// Starts, or resumes, a chunked upload. The durable manifest makes the call + /// idempotent. A first call takes ownership of the source file (an O(1) + /// rename into the library's directory) and saves the manifest BEFORE any + /// task is enqueued. A new call with the same id reconciles instead. The + /// same parts resume: the stored headers are replaced, and accepted parts + /// are skipped. Different parts recreate the upload, per the design's rule + /// (see ChunkedManifest.reconciled). Crash recovery, resume after a stop, + /// and resume with fresh auth are all this same call. + /// + /// Every rejection-type validation runs BEFORE the source is consumed. The + /// parse throws first, and a reconcile never touches the source (`path` is + /// ignored once a manifest exists). One rejection is possible after the + /// move: the manifest save can fail. That leaves the blob adoptable. A + /// retry with the same id finds the blob at the blob path and proceeds (see + /// takeOwnership). + func startUpload(_ options: [String: Any], + resolve: @escaping (String) -> Void, + reject: @escaping (String) -> Void) { + queue.async { + do { + let incoming = try ChunkedManifest.parse(options, createdAt: self.nowMs()) + let id = incoming.id + let manifest: ChunkedManifest + if let existing = ChunkedStore.load(id) { + // "Running" per the design's recreate rule: not stalled (no + // journaled terminal error or cancel that awaits this resume) and + // not past its deadline. Everything else rejects a different parts + // array. That includes part tasks live with the daemon, and + // finished-but-unacked. + let running = !existing.stalled && !existing.isExpired(self.nowMs()) + manifest = try existing.reconciled( + with: incoming, running: running, blobSize: ChunkedStore.blobSize(id)) + if manifest.incarnation != existing.incarnation { + // This is a recreate. The in-flight byte counts belong to the + // replaced parts. reconcileLocked below cancels the old + // incarnation's tasks, and does not adopt them. enqueuePart + // sweeps its temp files. + self.partSent[id] = nil + } + } else { + guard let path = options["path"] as? String else { + throw ChunkedManifest.ParseError(message: "Missing 'path'") + } + try self.takeOwnership(path: path, id: id) + // The same rule as recreate, and as Android's validatedForCreate: + // the parts must tile [0, blob size) exactly. A partial or + // overlapping cover would silently upload wrong bytes. This throws + // BEFORE the manifest is saved and before anything is enqueued. + // Thus the moved blob stays adoptable by a corrected retry with the + // same id (see takeOwnership). + let blobSize = ChunkedStore.blobSize(id) + guard ChunkedManifest.tilesExactly(incoming.parts, size: blobSize) else { + throw ChunkedManifest.ParseError( + message: "chunked upload '\(id)' parts must tile exactly [0, \(blobSize))") + } + manifest = incoming + } + try ChunkedStore.save(manifest) + self.manifests[id] = manifest + // A fresh call gets a fresh retry budget. The persisted per-part + // rejection counts reset in the parts rebuild above (reconciled or + // parse). + self.transientAttempts[id] = nil + self.cooldownUntil[id] = nil + self.reconcileLocked(id, resumedByStart: true) + resolve(id) + } catch { + reject(error.localizedDescription) + } + } + } + + /// Rebuilds every stored upload's in-flight set from the daemon, and then + /// refills. Called when the sessions are created or recreated: an app + /// relaunch, a JS reload, or the background-wake path through + /// `RNBackgroundUpload.shared`. + func reconcileAll() { + // Claim the system's background completion handlers BEFORE the reconcile + // is queued. After a relaunch, the replayed didCompleteWithError callbacks + // run unowned and never refill. Thus this chain is the only refill. + // Nothing else stops urlSessionDidFinishEvents from handing the system + // its handler, and the app its suspension, before one new part task is + // enqueued. The risk is largest exactly when every enqueued part finished + // while the app was dead: zero daemon tasks left, no future wake, and a + // silent stall. The claim provably precedes any drain: a relaunch reaches + // this point inside the init of `shared`, and the AppDelegate hook + // finishes that init before it stores the handler. + RNBackgroundUpload.deferBackgroundCompletionHandlers() + queue.async { + let group = DispatchGroup() + for manifest in ChunkedStore.all() { + self.manifests[manifest.id] = manifest + group.enter() + self.reconcileLocked(manifest.id, resumedByStart: false) { group.leave() } + } + group.notify(queue: self.queue) { + // Every upload's post-reconcile refill has resumed its tasks. The + // handlers can drain now. + RNBackgroundUpload.releaseBackgroundCompletionHandlers() + } + } + } + + /// Cancels a chunked upload. It journals one 'cancelled' (user) terminal + /// and stalls the upload. The manifest and the bytes are kept. Thus the + /// next startUpload resumes. Completion receives nil when the id has no + /// manifest (not a chunked upload). It receives false when nothing runs + /// (the upload is already terminal). + func cancel(_ id: String, completion: @escaping (Bool?) -> Void) { + queue.async { + guard let manifest = self.latest(id) else { completion(nil); return } + if manifest.stalled || manifest.allAccepted { completion(false); return } + var entry = JournaledEvent( + eventId: UUID().uuidString, id: id, type: "cancelled", timestamp: self.nowMs()) + entry.cancelReason = "user" + self.stall(id, entry: entry) + completion(true) + } + } + + /// An explicit release. It cancels the in-flight part tasks, with no + /// terminal event: the consumer lets go, and awaits no outcome. It deletes + /// the manifest, the moved bytes, and all part temp files. + func remove(_ id: String, completion: @escaping () -> Void) { + queue.async { + if self.latest(id) != nil { self.cancelTasks(for: id) } + ChunkedStore.remove(id) + self.clearState(id) + completion() + } + } + + /// The one moment when the library may delete a chunked upload's bytes: the + /// consumer acknowledged its 'completed' terminal event. + func releaseCompleted(_ ids: [String], completion: @escaping () -> Void) { + queue.async { + for id in ids { + ChunkedStore.remove(id) + self.clearState(id) + } + completion() + } + } + + /// The chunked rows for getAllUploads: one aggregate row per manifest. The + /// part tasks are transport detail. bytesSent counts accepted parts only. + /// That is the durable number. + func snapshots(completion: @escaping ([[String: Any]]) -> Void) { + queue.async { + let rows = ChunkedStore.all().map { manifest -> [String: Any] in + let state: String + if manifest.allAccepted { + state = "completed" + } else if manifest.stalled { + state = "error" + } else if !(self.inFlight[manifest.id] ?? [:]).isEmpty { + state = "running" + } else { + state = "pending" + } + return ["id": manifest.id, + "state": state, + "bytesSent": manifest.acceptedBytes, + "totalBytes": manifest.totalBytes] + } + completion(rows) + } + } + + // MARK: - Delegate hooks (called by RNBackgroundUpload) + + func partProgress(id: String, part: Int, incarnation: String?, sent: Int64) { + let now = Date().timeIntervalSince1970 + progressLock.lock() + if let last = lastProgressAt[id], now - last < Self.progressThrottle { + progressLock.unlock() + return + } + lastProgressAt[id] = now + progressLock.unlock() + queue.async { + // A removed or replaced incarnation's task must not feed the aggregate. + guard let manifest = self.manifests[id], manifest.incarnation == incarnation else { return } + self.partSent[id, default: [:]][part] = sent + self.emitAggregateProgress(id, manifest) + } + } + + /// One part task finished (a foreground or background-wake delegate + /// callback). We evaluate the accept rules, update the manifest, delete the + /// temp file, and refill the window. It is synchronous on purpose: the + /// journal write for a terminal outcome must land before the delegate + /// callback returns. The simple-upload path obeys the same rule. + func handlePartCompletion(id: String, part: Int, incarnation: String?, taskKey: String, + statusCode: Int?, headers: [String: String], + body: String?, error: NSError?) { + queue.sync { + TaskMap.removeKey(taskKey) + let owned = inFlight[id]?[part] == taskKey + if owned { + inFlight[id]?[part] = nil + partSent[id]?[part] = nil + } + guard var manifest = latest(id) else { + // The upload was removed (removeUpload, or a completed ack) while + // this task was in flight. There is nothing left to report. + if owned { ChunkedStore.removePartFile(id, part) } + return + } + // A late callback from a removed-then-recreated or replaced + // incarnation. Its response is about byte ranges and URLs that this + // manifest no longer describes. Thus nothing about it, the accept flag + // included, may be written into the current manifest. Its temp file has + // the old token in its name. The sweep removes it when the current plan + // next materializes this index. + guard incarnation == manifest.incarnation else { + if owned { refill(id) } + return + } + // A part index that the manifest does not know (corrupt task metadata) + // must not crash the delegate. Drop the task's outcome and let refill + // plan again. + guard manifest.parts.indices.contains(part) else { + if owned { refill(id) } + return + } + + // Accept evaluation comes first. The server holds these bytes now, + // regardless of a concurrent stall or a superseded task in the same + // incarnation. If we lose the flag, we re-send a part that the server + // already has. + if error == nil, let statusCode, + UploadOutcome.isAccepted(statusCode, body: body, accept: manifest.accept) { + manifest = updateManifest(id) { $0.withPartAccepted(part) } + ?? manifest.withPartAccepted(part) + transientAttempts[id]?[part] = nil + cooldownUntil[id]?[part] = nil + if owned { ChunkedStore.removePartFile(id, part) } + // A stalled upload keeps the flag but reports nothing more. The + // journaled terminal stands until the next startUpload resume. That + // resume finds all parts accepted and completes without a re-send. + guard !manifest.stalled else { return } + if manifest.allAccepted { + finalizeCompleted(id, manifest, reemit: false) + } else { + emitAggregateProgress(id, manifest) + if owned { refill(id) } + } + return + } + + // A superseded task's failure carries no policy weight. The live task + // for this part drives the retries. But an UNOWNED task with no live + // replacement is a relaunch replay that runs before reconcile rebuilds + // ownership. If we drop its deterministic HTTP rejection, the part gets + // a fresh retry budget on every system wake. So count it, and let it + // trip the budget. The in-flight reconcile does the re-enqueueing. + guard owned else { + if inFlight[id]?[part] == nil, !manifest.stalled, !manifest.parts[part].accepted, + error == nil, let code = statusCode, !ChunkedEngine.isTransientHttp(code) { + recordRejection(id, part: part, manifest: manifest, code: code, + headers: headers, body: body, scheduleRetryInBudget: false) + } + return + } + ChunkedStore.removePartFile(id, part) // the retry builds the file again + // This is a duplicate of a part that a superseded task already + // delivered. The part is settled, whatever this task's outcome was. Its + // failure must not burn retries. + if manifest.parts[part].accepted { + if !manifest.stalled { refill(id) } + return + } + // A terminal is already journaled (a cancel, or a sibling part's + // stall). Swallow the fallout. + guard !manifest.stalled else { return } + + if let error, error.domain == NSURLErrorDomain, error.code == NSURLErrorCancelled { + // A user cancel journals and stalls in cancel() before the tasks are + // torn down. Thus a cancel here, with no stall, comes from the + // system. Retry it like a transient failure. + scheduleTransientRetry(id, part: part) + return + } + + if manifest.isExpired(nowMs()) { + stall(id, entry: expiredEntry(id)) + return + } + + if let error { + if RNBackgroundUpload.errorKind(for: error) == "file", + !FileManager.default.fileExists(atPath: ChunkedStore.blobURL(id).path) { + stall(id, entry: errorEntry( + id: id, error: "chunked source blob missing", errorKind: "file", partIndex: part)) + } else { + // This includes a lost temp part file. The retry rebuilds it from + // the blob. + scheduleTransientRetry(id, part: part) + } + return + } + + let code = statusCode ?? 0 + if ChunkedEngine.isTransientHttp(code) { + scheduleTransientRetry(id, part: part) + return + } + recordRejection(id, part: part, manifest: manifest, code: code, + headers: headers, body: body, scheduleRetryInBudget: true) + } + } + + /// The identity of a chunked part task, or nil for a simple upload's task. + /// taskDescription is primary. The persisted TaskMap entry, written before + /// the task first resumed, is the durable fallback. `incarnation` is the + /// manifest token that the task was created under. It is nil only for + /// corrupt metadata, and the consumers treat nil as a mismatch. + static func partRef(_ session: URLSession, _ task: URLSessionTask) + -> (id: String, part: Int, incarnation: String?)? { + if let ref = ChunkedEngine.parseTaskDescription(task.taskDescription) { return ref } + if let meta = TaskMap.meta(forKey: TaskMap.key(session, task)), let part = meta.partIndex { + return (meta.id, part, meta.incarnation) + } + return nil + } + + // MARK: - Window (all on `queue`) + + /// Rebuilds inFlight for one upload from the daemon's live tasks, and then + /// refills. A task in the .completed or .canceling state is NOT live: its + /// delegate callback, replayed after a relaunch, settles it. A part with no + /// live task simply enqueues again. Accept evaluation absorbs a + /// completed-but-unreported duplicate. We never guess. + /// `completion` fires, on `queue`, when this reconcile has settled: the + /// refill ran, or a newer reconcile superseded this one. reconcileAll gates + /// the background completion handlers on it. + private func reconcileLocked(_ id: String, resumedByStart: Bool, + completion: (() -> Void)? = nil) { + let token = UUID() + reconcileToken[id] = token + enumerateAllTasks { tasks in + self.queue.async { + defer { completion?() } + guard self.reconcileToken[id] == token else { return } // superseded + let manifest = self.latest(id) + var live: [Int: String] = [:] + for (session, task) in tasks { + guard let ref = Self.partRef(session, task), ref.id == id, + task.state == .running || task.state == .suspended else { continue } + if ref.incarnation != manifest?.incarnation || live[ref.part] != nil { + // Never adopt a task from a replaced incarnation. Its bytes and + // URL belong to the old plan, and the token check in + // handlePartCompletion drops its late completion. Never adopt a + // second live task for one part index: concurrent PUTs of one + // partNum are verified unsafe on the server side. + task.cancel() + } else { + live[ref.part] = TaskMap.key(session, task) + } + } + self.inFlight[id] = live + self.reconcileToken[id] = nil + if let manifest { + // Temp files for accepted parts with no live task are orphans. + for index in manifest.parts.indices + where manifest.parts[index].accepted && live[index] == nil { + ChunkedStore.removePartFile(id, index) + } + } + self.refill(id, resumedByStart: resumedByStart) + } + } + } + + /// Fills the window back up to [ChunkedEngine.window] enqueued part tasks. + /// Called after every part completion (the background-wake refill that the + /// design's liveness rationale requires), after a retry cooldown, and at + /// the end of every reconcile. + private func refill(_ id: String, resumedByStart: Bool = false) { + guard reconcileToken[id] == nil, let manifest = latest(id) else { return } + // Stalled wins, even over all-accepted. The journaled terminal stands + // until an explicit startUpload resume. The resume clears the stall, + // lands here again, and completes without a re-send. + guard !manifest.stalled else { return } + if manifest.allAccepted { + finalizeCompleted(id, manifest, reemit: resumedByStart) + return + } + let now = nowMs() + if manifest.isExpired(now) { + stall(id, entry: expiredEntry(id)) + return + } + armExpiryCheck(id, expiresAt: manifest.expiresAt) + // A blob shorter than a part's range can never finish. Report a terminal + // 'file' now, not a surprise when the window reaches the short part + // later. A retry cannot help, because the bytes are not there. Thus this + // stalls, and awaits removeUpload or a recreate whose tiling rule fits + // the real size. + let blobSize = ChunkedStore.blobSize(id) + if let short = manifest.parts.indices.first(where: { manifest.parts[$0].end > blobSize }) { + stall(id, entry: errorEntry( + id: id, + error: "source blob is \(blobSize) bytes; part \(short) needs " + + "[\(manifest.parts[short].start), \(manifest.parts[short].end))", + errorKind: "file", partIndex: short)) + return + } + let flight = Set((inFlight[id] ?? [:]).keys) + let cooling = Set((cooldownUntil[id] ?? [:]).filter { $0.value > now }.keys) + for index in ChunkedEngine.indexesToEnqueue( + pending: manifest.pendingIndexes(), inFlight: flight, cooling: cooling) { + if !enqueuePart(id, index, manifest) { return } // stalled inside + } + } + + private func enqueuePart(_ id: String, _ index: Int, _ manifest: ChunkedManifest) -> Bool { + let part = manifest.parts[index] + guard let url = URL(string: part.url) else { + stall(id, entry: errorEntry( + id: id, error: "part \(index) url is not a valid URL", errorKind: "unknown", + partIndex: index)) + return false + } + // A background session can upload only from a file. Thus each enqueued + // part gets a temp file that holds exactly its byte range. The transient + // disk usage stays at window × partSize, not a second full copy of the + // source. + let partFile: URL + do { + partFile = try ChunkedStore.writePartFile( + id: id, index: index, start: part.start, end: part.end, + incarnation: manifest.incarnation) + } catch { + stall(id, entry: errorEntry( + id: id, error: "cannot materialize part \(index): \(error.localizedDescription)", + errorKind: "file", partIndex: index)) + return false + } + var request = URLRequest(url: url) + request.httpMethod = "PUT" + // Unchanged, per the protocol-as-data rule. The library adds nothing. + for (key, value) in part.headers { + request.setValue(value, forHTTPHeaderField: key) + } + let session = uploader.session(wifiOnly: manifest.wifiOnly) + let task = session.uploadTask(with: request, fromFile: partFile) + task.taskDescription = ChunkedEngine.taskDescription( + id: id, part: index, incarnation: manifest.incarnation) + let key = TaskMap.key(session, task) + TaskMap.set(TaskMap.Meta(id: id, accept: nil, partIndex: index, + incarnation: manifest.incarnation), forKey: key) + inFlight[id, default: [:]][index] = key + task.resume() + return true + } + + // MARK: - Terminal transitions (all on `queue`) + + /// Journals the terminal, marks the upload stalled, and cancels its + /// in-flight tasks. The stall is durable: relaunch reconciliation must not + /// resume the upload; only startUpload may. The manifest and the bytes are + /// kept. Every non-completed terminal leaves the consumer its recovery + /// options. + private func stall(_ id: String, entry: JournaledEvent) { + _ = updateManifest(id) { manifest in + var next = manifest + next.stalled = true + return next + } + cancelTasks(for: id) + partSent[id] = nil + cooldownUntil[id] = nil + RNBackgroundUpload.journalAndEmit(entry) + } + + private func finalizeCompleted(_ id: String, _ manifest: ChunkedManifest, reemit: Bool) { + for index in manifest.parts.indices { ChunkedStore.removePartFile(id, index) } + partSent[id] = nil + // A resume of a finished-but-unacked upload must not mint a second + // terminal event. Emit the journaled event again. Thus a live listener + // still hears it, with the eventId that the consumer will ack. + if let existing = EventJournal.unacknowledgedEntries() + .first(where: { $0.id == id && $0.type == "completed" }) { + if reemit { RNBackgroundUpload.emitEvent(existing) } + return + } + // There are no response fields, because no single response represents N + // accepted parts. The blob is deleted only when this event is ACKED (see + // ackEvents). + RNBackgroundUpload.journalAndEmit( + JournaledEvent(eventId: UUID().uuidString, id: id, type: "completed", timestamp: nowMs())) + } + + // MARK: - Retry scheduling (all on `queue`) + + /// Counts one non-transient HTTP rejection against the budget of `part`. + /// The count lives in the manifest, persisted best-effort like the accepted + /// flag. Thus it survives process death and can trip across wakes. A resume + /// or a recreate resets it (ChunkedManifest.reconciled rebuilds the parts + /// from the incoming call). Over budget: journal the terminal 'http' and + /// stall. In budget: schedule the backoff retry when this callback owns the + /// part. For an unowned replay, the reconcile already in flight does the + /// re-enqueueing. + private func recordRejection(_ id: String, part: Int, manifest: ChunkedManifest, + code: Int, headers: [String: String], body: String?, + scheduleRetryInBudget: Bool) { + let count = (manifest.parts[part].rejections ?? 0) + 1 + _ = updateManifest(id) { $0.withPartRejections(part, count) } + if count > ChunkedEngine.partHttpRetries { + let (capped, truncated) = EventJournal.capBody(body) + var entry = errorEntry( + id: id, error: "HTTP \(code) on part \(part)", errorKind: "http", partIndex: part) + entry.responseCode = code + entry.responseBody = capped + entry.responseBodyTruncated = truncated + entry.responseHeaders = headers + stall(id, entry: entry) + } else if scheduleRetryInBudget { + scheduleRetry(id, part: part, attempt: count) + } + } + + private func scheduleTransientRetry(_ id: String, part: Int) { + let attempt = (transientAttempts[id]?[part] ?? 0) + 1 + transientAttempts[id, default: [:]][part] = attempt + scheduleRetry(id, part: part, attempt: attempt) + } + + private func scheduleRetry(_ id: String, part: Int, attempt: Int) { + let delayMs = ChunkedEngine.backoffMs(attempt: attempt) + cooldownUntil[id, default: [:]][part] = nowMs() + Double(delayMs) + queue.asyncAfter(deadline: .now() + .milliseconds(delayMs)) { [weak self] in + guard let self else { return } + self.cooldownUntil[id]?[part] = nil + self.refill(id) + } + } + + // Expiry is evaluated on every transition. But an upload whose tasks all + // wait (for connectivity, or for backoff) would pass its deadline silently + // while the app is alive. Thus we arm one timer at the deadline. When a + // resume extended expiresAt, the stale timer's refill is a no-op that arms + // the timer again. + private func armExpiryCheck(_ id: String, expiresAt: Double) { + guard !expiryArmed.contains(id) else { return } + expiryArmed.insert(id) + let delayMs = Int(min(max(expiresAt - nowMs(), 0) + 100, 7 * 24 * 3_600_000)) + queue.asyncAfter(deadline: .now() + .milliseconds(delayMs)) { [weak self] in + guard let self else { return } + self.expiryArmed.remove(id) + self.refill(id) + } + } + + // MARK: - Helpers + + // The stored copy is the truth. A reconcile can have replaced the headers + // or expiresAt. The cache exists for the progress path, and as a fallback + // when a read fails in flight. + private func latest(_ id: String) -> ChunkedManifest? { + guard let manifest = ChunkedStore.load(id) else { + manifests[id] = nil + return nil + } + manifests[id] = manifest + return manifest + } + + private func updateManifest( + _ id: String, _ transform: (ChunkedManifest) -> ChunkedManifest + ) -> ChunkedManifest? { + // Here the save is best-effort, unlike in startUpload. A lost accepted + // flag only causes a re-send of a part, and the server absorbs the + // duplicate through the accept rules. That is better than a failed upload + // that the server in fact took. + let next = ChunkedStore.update(id, transform) ?? manifests[id].map(transform) + if let next { manifests[id] = next } + return next + } + + private func takeOwnership(path: String, id: String) throws { + let source = URL(string: path) ?? URL(fileURLWithPath: path) + let blob = ChunkedStore.blobURL(id) + let fm = FileManager.default + guard fm.fileExists(atPath: source.path) else { + // A crash between the move and the manifest save leaves the bytes at + // the blob path with no manifest. Adopt the bytes. Do not fail the + // retry. + if fm.fileExists(atPath: blob.path) { return } + throw ChunkedManifest.ParseError( + message: "chunked source file does not exist: \(source.path)") + } + try fm.createDirectory(at: ChunkedStore.uploadDir(id), withIntermediateDirectories: true) + try? fm.removeItem(at: blob) + // This is an O(1) rename on the same volume. Across volumes, FileManager + // falls back to a copy. + try fm.moveItem(at: source, to: blob) + } + + private func emitAggregateProgress(_ id: String, _ manifest: ChunkedManifest) { + let total = manifest.totalBytes + guard total > 0 else { return } + let sent = min(manifest.acceptedBytes + (partSent[id]?.values.reduce(0, +) ?? 0), total) + RNBackgroundUpload.emitProgress(id: id, progress: 100.0 * Float(sent) / Float(total)) + } + + private func clearState(_ id: String) { + inFlight[id] = nil + reconcileToken[id] = nil // discards any pending reconcile snapshot + partSent[id] = nil + cooldownUntil[id] = nil + transientAttempts[id] = nil + manifests[id] = nil + // A removed-then-recreated id must be able to arm its own expiry + // deadline, which is possibly earlier. It must not wait out the stale + // timer. + expiryArmed.remove(id) + progressLock.lock() + lastProgressAt[id] = nil // without this, one entry per id stays forever + progressLock.unlock() + } + + private func cancelTasks(for id: String) { + enumerateAllTasks { tasks in + for (session, task) in tasks where Self.partRef(session, task)?.id == id { + task.cancel() + } + } + } + + // Always examine both sessions. A resume can change wifiOnly while earlier + // part tasks continue where they started. + private func enumerateAllTasks( + _ completion: @escaping ([(URLSession, URLSessionTask)]) -> Void + ) { + let sessions = [uploader.session(wifiOnly: false), uploader.session(wifiOnly: true)] + let group = DispatchGroup() + let lock = NSLock() + var collected: [(URLSession, URLSessionTask)] = [] + for session in sessions { + group.enter() + session.getAllTasks { tasks in + lock.lock() + collected.append(contentsOf: tasks.map { (session, $0) }) + lock.unlock() + group.leave() + } + } + group.notify(queue: .global()) { completion(collected) } + } + + private func expiredEntry(_ id: String) -> JournaledEvent { + errorEntry(id: id, error: "upload expired before every part was accepted", + errorKind: "expired") + } + + private func errorEntry(id: String, error: String, errorKind: String, + partIndex: Int? = nil) -> JournaledEvent { + var entry = JournaledEvent( + eventId: UUID().uuidString, id: id, type: "error", timestamp: nowMs()) + entry.error = error + entry.errorKind = errorKind + entry.partIndex = partIndex + return entry + } +} diff --git a/ios/ChunkedEngine.swift b/ios/ChunkedEngine.swift new file mode 100644 index 00000000..caf3007c --- /dev/null +++ b/ios/ChunkedEngine.swift @@ -0,0 +1,81 @@ +import Foundation + +// The pure scheduling half of chunked execution: window arithmetic, the +// retry policy, backoff, and the part-task identity encoding. It is kept free +// of session state. Thus the highest-consequence invariants (at most WINDOW +// part tasks enqueued per upload, and never two for one part index) can be +// examined in one place. [ChunkedCoordinator] owns the session side. +enum ChunkedEngine { + + // The number of part tasks of one upload enqueued with the daemon at one + // time. It is a library constant, not an option: if soak data argues for a + // different value, this constant changes, not the API. The window is also a + // liveness decision. A background session only progresses tasks that are + // already enqueued. Thus WINDOW tasks of runway let a multi-part upload + // proceed while the app is dead. Without them, the upload pays a + // rate-limited wake per part. + static let window = 3 + + // A non-accepted, non-transient HTTP response is retried this many times + // for each part. Then it becomes a terminal error and stalls the upload. + // The number is small on purpose. A response that the server repeats (401, + // 400) will not change without a new startUpload. Only transient failures + // retry without a limit. + static let partHttpRetries = 3 + + private static let backoffBaseMs = 1_000 + private static let backoffCapMs = 60_000 + + // A 5xx means that the server fails, not that the request is wrong. Thus + // it retries like a transport failure: without a limit, within expiresAt. + static func isTransientHttp(_ code: Int) -> Bool { (500...599).contains(code) } + + /// Exponential backoff for transient failures: 1s, 2s, 4s, up to a 60s cap. + static func backoffMs(attempt: Int) -> Int { + min(backoffBaseMs << min(max(attempt - 1, 0), 6), backoffCapMs) + } + + /// The part indexes to enqueue now: pending (not accepted), not already + /// enqueued, and not cooling down after a failure, up to the window size. + /// It never returns an index in `inFlight`. That is the one-task-per-part + /// invariant. + static func indexesToEnqueue( + pending: [Int], inFlight: Set, cooling: Set, window: Int = window + ) -> [Int] { + let slots = window - inFlight.count + guard slots > 0 else { return [] } + return Array(pending.filter { !inFlight.contains($0) && !cooling.contains($0) }.prefix(slots)) + } + + // MARK: - Part-task identity + + // A chunked part task must carry (uploadId, partIndex, incarnation) + // through the daemon. taskDescription is the primary carrier. It is a + // prefix plus JSON, so a consumer id that contains a delimiter survives. + // TaskMap holds the same triple as the durable fallback, per the DTS + // guidance that TaskMap documents. The incarnation is the manifest token + // that the task was created under. A callback whose token no longer matches + // the stored manifest's token is from a removed or replaced plan. It must + // not write into the current plan. + private static let descriptionPrefix = "rnbgu-chunk:" + + private struct PartRef: Codable { + let id: String + let part: Int + var inc: String? + } + + static func taskDescription(id: String, part: Int, incarnation: String) -> String { + let data = (try? JSONEncoder().encode(PartRef(id: id, part: part, inc: incarnation))) ?? Data() + return descriptionPrefix + (String(data: data, encoding: .utf8) ?? "") + } + + static func parseTaskDescription( + _ description: String? + ) -> (id: String, part: Int, incarnation: String?)? { + guard let description, description.hasPrefix(descriptionPrefix) else { return nil } + let json = Data(description.dropFirst(descriptionPrefix.count).utf8) + guard let ref = try? JSONDecoder().decode(PartRef.self, from: json) else { return nil } + return (ref.id, ref.part, ref.inc) + } +} diff --git a/ios/ChunkedManifest.swift b/ios/ChunkedManifest.swift new file mode 100644 index 00000000..5375cd3c --- /dev/null +++ b/ios/ChunkedManifest.swift @@ -0,0 +1,404 @@ +import Foundation + +/// The durable record of one chunked upload: the parts that the consumer +/// authored, and which of them the server has accepted. [ChunkedStore] saves +/// it at startUpload, BEFORE any task is enqueued. Thus a process that the +/// system relaunches (or a startUpload after a crash, a stop, or a reauth) +/// resumes from it without a call into JS. This manifest IS the resume +/// mechanism. +/// +/// The content is the same as the Android manifest, with two platform +/// differences: +/// - There is no sourcePath field. iOS moves the app container between +/// launches, so an absolute path would go stale. The moved bytes live at a +/// location derived from the id (ChunkedStore.blobURL). +/// - `stalled` is persisted. On Android, "stalled" only means that the worker +/// is not scheduled. iOS reconciles every upload on relaunch. Thus an +/// upload that journaled a terminal outcome needs a durable marker that +/// says: await an explicit startUpload resume, and do not refill. +struct ChunkedManifest: Codable { + /// One part, exactly as the consumer authored it. The library sends the + /// file bytes [start, end) as the body of a PUT to `url`, with `headers` + /// unchanged. It never derives or edits a protocol field. + struct Part: Codable { + let url: String + var headers: [String: String] + let start: Int64 + let end: Int64 // exclusive + var accepted: Bool = false + /// The non-transient HTTP rejections counted against this part's retry + /// budget (nil means 0). It is persisted so that the budget survives + /// process death. An in-memory count resets on every system wake. That + /// would let a deterministic 4xx upload the part again until expiresAt, + /// with no terminal ever journaled. The count resets when the part is + /// rebuilt from an incoming call. Resume and recreate both do that (see + /// [reconciled]). + var rejections: Int? + + var size: Int64 { end - start } + } + + let id: String + var parts: [Part] + var accept: [UploadOutcome.AcceptRule] + /// Epoch ms. After this time, the upload stops with errorKind 'expired'. + var expiresAt: Double + var wifiOnly: Bool + let createdAt: Double + var stalled: Bool = false + /// The identity of this parts plan. It rotates on a recreate (a startUpload + /// that replaced the parts wholesale). It never rotates on a resume. Part + /// tasks carry it in their identity. Thus a late delegate callback from a + /// removed or replaced incarnation can be told apart from the live plan's + /// callbacks and dropped. Its response is about byte ranges and URLs that + /// this manifest no longer describes. + var incarnation: String + + var totalBytes: Int64 { parts.reduce(0) { $0 + $1.size } } + var acceptedBytes: Int64 { parts.filter(\.accepted).reduce(0) { $0 + $1.size } } + + /// The server's auto-publish condition. It is the only thing that + /// 'completed' may mean. + var allAccepted: Bool { parts.allSatisfy(\.accepted) } + + func isExpired(_ nowMs: Double) -> Bool { nowMs >= expiresAt } + + func pendingIndexes() -> [Int] { parts.indices.filter { !parts[$0].accepted } } + + func withPartAccepted(_ index: Int) -> ChunkedManifest { + var next = self + next.parts[index].accepted = true + return next + } + + func withPartRejections(_ index: Int, _ count: Int) -> ChunkedManifest { + var next = self + next.parts[index].rejections = count + return next + } + + struct ReconcileError: LocalizedError { + let message: String + var errorDescription: String? { message } + } + + /// A new startUpload call with an existing id is one of two things. The + /// semantics are identical to Android's `ChunkedManifest.reconcile`. + /// + /// **Resume** — the incoming parts are the SAME array (identical count, + /// ranges, and urls). The headers, the accept rules, expiresAt, and wifiOnly + /// come from the new call. This is how fresh auth reaches stalled parts, + /// and how a salvage extends the deadline. The accepted part statuses, + /// createdAt, and the incarnation survive from this manifest. A resume is + /// permitted at any time, running or not. The stall clears, because a + /// resume is the whole point of the new call. + /// + /// **Recreate** — a DIFFERENT parts array: the consumer authored the upload + /// again, under a fresh server uploadId, after the old one died. The owned + /// bytes are kept. The parts are replaced wholesale. Every part status + /// resets to unsent. The headers, accept rules, and expiresAt come from the + /// new call. The new ranges must tile exactly [0, blobSize). A partial or + /// overlapping cover would silently upload wrong bytes. A recreate is + /// accepted only while the upload is NOT running (stalled on a terminal + /// error or cancel, or expired). A different parts array while part tasks + /// are live is a consumer bug, not a recreate, because the in-flight + /// requests belong to the old parts. The incarnation rotates to the + /// incoming manifest's fresh token. Thus late callbacks from the replaced + /// parts are dropped. + func reconciled(with incoming: ChunkedManifest, running: Bool, + blobSize: Int64) throws -> ChunkedManifest { + if samePartsAs(incoming) { + // Built from `incoming`, so the per-part rejection counts reset. A + // resume arrives with fresh headers and gets a fresh retry budget. + let mergedParts = incoming.parts.enumerated().map { i, new -> Part in + var part = new + part.accepted = parts[i].accepted + return part + } + return ChunkedManifest( + id: id, parts: mergedParts, accept: incoming.accept, expiresAt: incoming.expiresAt, + wifiOnly: incoming.wifiOnly, createdAt: createdAt, stalled: false, + incarnation: incarnation) + } + guard !running else { + throw ReconcileError(message: + "chunked upload '\(id)' is running; a different parts array is only accepted once it stops") + } + guard Self.tilesExactly(incoming.parts, size: blobSize) else { + throw ReconcileError(message: + "chunked upload '\(id)' recreate parts must tile exactly [0, \(blobSize))") + } + return ChunkedManifest( + id: id, parts: incoming.parts, accept: incoming.accept, expiresAt: incoming.expiresAt, + wifiOnly: incoming.wifiOnly, createdAt: createdAt, stalled: false, + incarnation: incoming.incarnation) + } + + private func samePartsAs(_ incoming: ChunkedManifest) -> Bool { + incoming.parts.count == parts.count && parts.indices.allSatisfy { i in + incoming.parts[i].url == parts[i].url + && incoming.parts[i].start == parts[i].start + && incoming.parts[i].end == parts[i].end + } + } + + /// Tells whether `parts` cover [0, size) exactly: no gap, no overlap, and + /// nothing past the end. It is order-independent, like everything else + /// about parts. + static func tilesExactly(_ parts: [Part], size: Int64) -> Bool { + guard !parts.isEmpty else { return false } + var cursor: Int64 = 0 + for part in parts.sorted(by: { $0.start < $1.start }) { + guard part.start == cursor, part.end > part.start else { return false } + cursor = part.end + } + return cursor == size + } + + struct ParseError: LocalizedError { + let message: String + var errorDescription: String? { message } + } + + /// Turns bridged options into a manifest. It throws on each field that the + /// engine relies on. JS validates first. Thus a throw here is a bug worth + /// surfacing, not UX. + static func parse(_ options: [String: Any], createdAt: Double) throws -> ChunkedManifest { + guard let id = options["id"] as? String, !id.isEmpty else { + throw ParseError(message: "Missing 'id'") + } + guard let rawParts = options["parts"] as? [[String: Any]], !rawParts.isEmpty else { + throw ParseError(message: "'parts' must be a non-empty array") + } + guard let expiresAt = (options["expiresAt"] as? NSNumber)?.doubleValue else { + throw ParseError(message: "Missing 'expiresAt'") + } + let parts = try rawParts.enumerated().map { i, raw -> Part in + guard let url = raw["url"] as? String else { + throw ParseError(message: "Missing 'parts[\(i)].url'") + } + guard let range = raw["range"] as? [String: Any], + let start = (range["start"] as? NSNumber)?.int64Value, + let end = (range["end"] as? NSNumber)?.int64Value, + start >= 0, start < end else { + throw ParseError(message: "Invalid 'parts[\(i)].range'") + } + return Part(url: url, headers: parseHeaders(raw["headers"]), start: start, end: end) + } + return ChunkedManifest( + id: id, + parts: parts, + accept: UploadOutcome.parseAcceptRules(options["accept"]), + expiresAt: expiresAt, + wifiOnly: (options["wifiOnly"] as? Bool) ?? false, + createdAt: createdAt, + incarnation: UUID().uuidString) + } + + // The same header coercion as the simple-upload path: strings and numbers + // only. Anything else is skipped. It is not interpolated onto the wire. + private static func parseHeaders(_ raw: Any?) -> [String: String] { + guard let headers = raw as? [String: Any] else { return [:] } + var result: [String: String] = [:] + for (key, value) in headers { + if let s = value as? String { + result[key] = s + } else if let n = value as? NSNumber { + result[key] = n.stringValue + } + } + return result + } +} + +/// A file-backed store: one directory per upload id. The directory holds +/// `manifest.json`, `blob` (the moved source bytes), and the in-flight part +/// temp files. It has the same durability pattern as [EventJournal]: a +/// synchronous serial queue, atomic writes, and corrupt files read as +/// absent. +enum ChunkedStore { + struct StoreError: LocalizedError { + let message: String + var errorDescription: String? { message } + } + + private static let queue = DispatchQueue(label: "ai.openspace.rnbgupload.chunkedstore") + + private static let dirURL: URL = { + let base = FileManager.default.urls(for: .applicationSupportDirectory, in: .userDomainMask)[0] + var dir = base.appendingPathComponent("RNFileUploaderChunked", isDirectory: true) + try? FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true) + // This is device-local upload state. Keep it out of iCloud/iTunes + // backups. + var values = URLResourceValues() + values.isExcludedFromBackup = true + try? dir.setResourceValues(values) + return dir + }() + + // Upload ids come from the consumer. They can contain path separators or + // other filesystem-hostile characters. Thus the directory name is an + // encoding of the id, never the id itself. The id is read back from the + // manifest, not decoded from the name. + static func uploadDir(_ id: String) -> URL { + dirURL.appendingPathComponent( + Data(id.utf8).base64EncodedString() + .replacingOccurrences(of: "+", with: "-") + .replacingOccurrences(of: "/", with: "_") + .replacingOccurrences(of: "=", with: ""), + isDirectory: true) + } + + private static func manifestURL(_ id: String) -> URL { + uploadDir(id).appendingPathComponent("manifest.json") + } + + /// The location where startUpload moves the source file for this id. + static func blobURL(_ id: String) -> URL { + uploadDir(id).appendingPathComponent("blob") + } + + /// The temp file that holds exactly the byte range of part `index` while + /// that part is enqueued with the daemon (a background session can upload + /// only from a file). The name encodes the manifest incarnation and the + /// byte range. Thus a stale file from a previous incarnation or a different + /// plan can never be adopted by a size coincidence. Reuse checks the full + /// identity, not only the byte count. + static func partFileURL(_ id: String, _ index: Int, incarnation: String, + start: Int64, end: Int64) -> URL { + uploadDir(id).appendingPathComponent("part-\(index).\(incarnation).\(start)-\(end)") + } + + /// The size in bytes of the moved blob. It is 0 when the blob is missing. + static func blobSize(_ id: String) -> Int64 { + queue.sync { + (((try? FileManager.default.attributesOfItem(atPath: blobURL(id).path))?[.size] + as? NSNumber)?.int64Value) ?? 0 + } + } + + static func load(_ id: String) -> ChunkedManifest? { + queue.sync { read(manifestURL(id)) } + } + + /// Throws on a write failure. A manifest that did not persist must fail + /// the startUpload call. + static func save(_ manifest: ChunkedManifest) throws { + try queue.sync { + try FileManager.default.createDirectory( + at: uploadDir(manifest.id), withIntermediateDirectories: true) + let data = try JSONEncoder().encode(manifest) + try data.write(to: manifestURL(manifest.id), options: .atomic) + } + } + + /// An atomic read-modify-write. Thus a mark of one part as accepted can + /// never clobber a concurrent reconcile's fresh headers, or another part's + /// flag. It returns nil, and does not throw, when the manifest is gone or + /// the write failed. Callers that can proceed from memory do so. + static func update(_ id: String, _ transform: (ChunkedManifest) -> ChunkedManifest) -> ChunkedManifest? { + queue.sync { + guard let manifest = read(manifestURL(id)) else { return nil } + let next = transform(manifest) + guard let data = try? JSONEncoder().encode(next) else { return nil } + do { + try data.write(to: manifestURL(id), options: .atomic) + return next + } catch { + return nil + } + } + } + + /// Deletes the manifest, the moved bytes, AND all part temp files. It does + /// nothing for an unknown id (for example, a simple upload's id). + static func remove(_ id: String) { + queue.sync { try? FileManager.default.removeItem(at: uploadDir(id)) } + } + + static func all() -> [ChunkedManifest] { + queue.sync { + let dirs = (try? FileManager.default.contentsOfDirectory( + at: dirURL, includingPropertiesForKeys: nil)) ?? [] + return dirs.compactMap { read($0.appendingPathComponent("manifest.json")) } + } + } + + // Codable enforces the non-optional fields at decode time, unlike Gson. + // Thus a corrupt or field-renamed file simply reads as absent. + private static func read(_ url: URL) -> ChunkedManifest? { + guard let data = try? Data(contentsOf: url) else { return nil } + guard let m = try? JSONDecoder().decode(ChunkedManifest.self, from: data), + !m.parts.isEmpty else { return nil } + return m + } + + /// Writes bytes [start, end) of the blob into `dest`. It writes a tmp file + /// and renames it. Thus a partial write can never be mistaken for a + /// finished part file. It throws when the blob is missing or shorter than + /// `end`. For the caller that is a 'file' terminal, because a retry can + /// never succeed. + static func writePartFile(id: String, index: Int, start: Int64, end: Int64, + incarnation: String) throws -> URL { + try queue.sync { + let dest = partFileURL(id, index, incarnation: incarnation, start: start, end: end) + // Sweep the other files of this index first. A temp file left by a + // replaced incarnation or plan must not stay and leak disk. It cannot + // be reused, because the identity is in the name, but it can pile up. + removePartFilesLocked(id, index, keeping: dest) + // An existing file with exactly this identity and size is a finished + // copy from a previous enqueue of this part. Reuse it. Size alone is + // not trusted. The name carries the incarnation and the range that + // produced the file. + if let size = try? FileManager.default.attributesOfItem(atPath: dest.path)[.size] as? NSNumber, + size.int64Value == end - start { + return dest + } + let blob = blobURL(id) + let blobSize = ((try FileManager.default.attributesOfItem(atPath: blob.path)[.size] + as? NSNumber)?.int64Value) ?? 0 + guard blobSize >= end else { + throw StoreError( + message: "source blob is \(blobSize) bytes; part \(index) needs [\(start), \(end))") + } + let tmp = uploadDir(id).appendingPathComponent("part-\(index).tmp") + FileManager.default.createFile(atPath: tmp.path, contents: nil) + let reader = try FileHandle(forReadingFrom: blob) + defer { try? reader.close() } + let writer = try FileHandle(forWritingTo: tmp) + defer { try? writer.close() } + try reader.seek(toOffset: UInt64(start)) + 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 StoreError(message: "short read building part \(index)") + } + try writer.write(contentsOf: data) + remaining -= Int64(data.count) + } + try? FileManager.default.removeItem(at: dest) + try FileManager.default.moveItem(at: tmp, to: dest) + return dest + } + } + + /// Removes every file for part `index`: the current incarnation's file, + /// stale files, and half-written tmp files. They all share the + /// `part-.` prefix. + static func removePartFile(_ id: String, _ index: Int) { + queue.sync { removePartFilesLocked(id, index, keeping: nil) } + } + + // Must run on `queue`. The `part-.` prefix cannot collide across + // indexes ("part-1." is not a prefix of "part-12.<...>"). + private static func removePartFilesLocked(_ id: String, _ index: Int, keeping: URL?) { + let files = (try? FileManager.default.contentsOfDirectory( + at: uploadDir(id), includingPropertiesForKeys: nil)) ?? [] + for file in files + where file.lastPathComponent.hasPrefix("part-\(index).") + && file.lastPathComponent != keeping?.lastPathComponent { + try? FileManager.default.removeItem(at: file) + } + } +} diff --git a/ios/EventJournal.swift b/ios/EventJournal.swift index 9f3f2d24..3011acd8 100644 --- a/ios/EventJournal.swift +++ b/ios/EventJournal.swift @@ -11,8 +11,9 @@ struct JournaledEvent: Codable { var responseBodyTruncated: Bool? var responseHeaders: [String: String]? var error: String? - var errorKind: String? // http | network | file | unknown + var errorKind: String? // http | network | file | expired | unknown var cancelReason: String? // user | system + var partIndex: Int? // chunked uploads only: the failing part, when known // Bridge-friendly dictionary (nil fields omitted so nothing becomes NSNull). var bridged: [String: Any] { @@ -24,6 +25,7 @@ struct JournaledEvent: Codable { if let error { m["error"] = error } if let errorKind { m["errorKind"] = errorKind } if let cancelReason { m["cancelReason"] = cancelReason } + if let partIndex { m["partIndex"] = partIndex } return m } } @@ -88,6 +90,10 @@ enum EventJournal { } static func unacknowledged() -> [[String: Any]] { + unacknowledgedEntries().map { $0.bridged } + } + + static func unacknowledgedEntries() -> [JournaledEvent] { queue.sync { let files = (try? FileManager.default.contentsOfDirectory(at: dirURL, includingPropertiesForKeys: nil)) ?? [] return files @@ -97,7 +103,6 @@ enum EventJournal { return try? JSONDecoder().decode(JournaledEvent.self, from: data) } .sorted { $0.timestamp < $1.timestamp } - .map { $0.bridged } } } diff --git a/ios/RNBackgroundUpload.swift b/ios/RNBackgroundUpload.swift index 7a11480f..ca16e166 100644 --- a/ios/RNBackgroundUpload.swift +++ b/ios/RNBackgroundUpload.swift @@ -44,6 +44,19 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { private static var responsesData: [String: NSMutableData] = [:] // sessionId:taskId -> body private static var lastProgressAt: [String: TimeInterval] = [:] // uploadId -> time private static var userCancelledIds = Set() + // The ids that removeUpload is releasing now. The cancellation of their + // tasks is an explicit release, not an outcome that the consumer awaits. + // Thus no terminal event is journaled. This matches Android, whose + // removeUpload cancels work with no user-cancel mark. + private static var removedIds = Set() + // The consumer-supplied ids whose check-and-create is in flight, mapped to + // the resolves of the concurrent same-id calls. The existence check + // enumerates the session tasks asynchronously. Without this claim, two + // concurrent calls could both see "no task" and enqueue duplicates. The id + // is claimed synchronously, under `lock`, BEFORE the enumeration is + // dispatched. The map entry drains when the first caller's create-or-find + // lands. + private static var creationsInFlight: [String: [RCTPromiseResolveBlock]] = [:] private static var backgroundSession: URLSession? private static var wifiOnlySession: URLSession? @@ -58,6 +71,19 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { // id) so the app can be relaunched to finish uploads after termination. private static let bgHandlerLock = NSLock() private static var bgCompletionHandlers: [String: () -> Void] = [:] + // Relaunch ordering: while the chunked coordinator reconciles (deferrals + // > 0), urlSessionDidFinishEvents must NOT hand the system its completion + // handler. The system could suspend the app before the post-reconcile + // refill enqueues a new part task. That would leave zero daemon tasks and + // no future wake. A session that finishes its events in that window parks + // its id here. The release drains it. + private static var bgHandlerDeferrals = 0 + private static var bgSessionsAwaitingDrain: Set = [] + + // Owns the chunked-upload window and the manifests. It is implicitly + // unwrapped only because it needs `self` (for the sessions) and is assigned + // before init returns. It is never nil after that. + private var chunked: ChunkedCoordinator! public override init() { super.init() @@ -65,6 +91,12 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { // nsurlsessiond from a previous launch are delivered to this process. _ = session(wifiOnly: false) _ = session(wifiOnly: true) + chunked = ChunkedCoordinator(uploader: self) + // Relaunch reconciliation: match the daemon's surviving tasks against + // the stored manifests, and refill each upload's window. It runs on the + // coordinator queue. Thus nothing here re-enters the initialization of + // `shared`. + chunked.reconcileAll() } // MARK: - Event delegate @@ -97,9 +129,35 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { return eventDelegate } + // Journal-before-emit, the library's one terminal-event path. The write is + // durable. The emit is best-effort, because JS can be dead. The + // simple-upload delegate handling and the chunked coordinator share it. + static func journalAndEmit(_ event: JournaledEvent) { + EventJournal.append(event) + emitEvent(event) + } + + /// Emits WITHOUT a journal write. Use it to deliver again an event that is + /// already in the journal (a resume of a finished-but-unacked upload). + static func emitEvent(_ event: JournaledEvent) { + let body = event.bridged + let delegate = currentDelegate + switch event.type { + case "completed": delegate?.emitCompleted(body) + case "cancelled": delegate?.emitCancelled(body) + default: delegate?.emitError(body) + } + } + + static func emitProgress(id: String, progress: Float) { + currentDelegate?.emitProgress(["id": id, "progress": progress]) + } + // MARK: - Sessions - private func session(wifiOnly: Bool) -> URLSession { + // Internal, not private: the chunked coordinator enqueues part tasks on the + // same two sessions. + func session(wifiOnly: Bool) -> URLSession { RNBackgroundUpload.lock.lock() defer { RNBackgroundUpload.lock.unlock() } if wifiOnly { @@ -130,16 +188,19 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { } private func taskMapKey(_ session: URLSession, _ task: URLSessionTask) -> String { - "\(session.configuration.identifier ?? ""):\(task.taskIdentifier)" + TaskMap.key(session, task) } // taskDescription is the primary id; the persisted map is the durable fallback. + // A chunked part task's description encodes (uploadId, partIndex). This + // returns the uploadId in both cases. Thus id matching works uniformly. private func uploadId(_ session: URLSession, _ task: URLSessionTask) -> String { - task.taskDescription ?? TaskMap.meta(forKey: taskMapKey(session, task))?.id ?? "unknown" + if let ref = ChunkedCoordinator.partRef(session, task) { return ref.id } + return task.taskDescription ?? TaskMap.meta(forKey: taskMapKey(session, task))?.id ?? "unknown" } - private func acceptStatus(_ session: URLSession, _ task: URLSessionTask) -> [Int] { - TaskMap.meta(forKey: taskMapKey(session, task))?.acceptStatus ?? [] + private func acceptRules(_ session: URLSession, _ task: URLSessionTask) -> [UploadOutcome.AcceptRule] { + TaskMap.meta(forKey: taskMapKey(session, task))?.accept ?? [] } private var activeSessions: [URLSession] { @@ -180,9 +241,7 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { } let wifiOnly = (options["wifiOnly"] as? Bool) ?? false - // RN bridges a JS number[] to NSArray; map explicitly rather than - // rely on an [Int] bridging cast that can yield nil and silently drop it. - let acceptStatus = (options["acceptStatus"] as? [NSNumber])?.map { $0.intValue } ?? [] + let accept = UploadOutcome.parseAcceptRules(options["accept"]) let uploadId = (options["id"] as? String) ?? UUID().uuidString let fileURL = URL(string: path) ?? URL(fileURLWithPath: path) @@ -190,10 +249,9 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { let startNew = { let task = session.uploadTask(with: request, fromFile: fileURL) task.taskDescription = uploadId - TaskMap.set(TaskMap.Meta(id: uploadId, acceptStatus: acceptStatus), + TaskMap.set(TaskMap.Meta(id: uploadId, accept: accept, partIndex: nil), forKey: self.taskMapKey(session, task)) task.resume() - resolve(uploadId) } // A consumer-supplied id makes startUpload idempotent. This is the same @@ -203,7 +261,32 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { // different wifiOnly value while the first task continues in its first // session. A generated id cannot collide, so that path does not do the // (asynchronous) task enumeration. - guard options["id"] != nil else { startNew(); return } + guard options["id"] != nil else { + startNew() + resolve(uploadId) + return + } + + // Serialize the check-and-create for each id: claim the id synchronously, + // before we dispatch the enumeration. The first caller runs the check and + // creates the task. A concurrent same-id caller parks its resolve here. + // When the task lands, we answer the parked calls with the id. There is no + // second task, and there is no polling. + RNBackgroundUpload.lock.lock() + if RNBackgroundUpload.creationsInFlight[uploadId] != nil { + RNBackgroundUpload.creationsInFlight[uploadId]?.append(resolve) + RNBackgroundUpload.lock.unlock() + return + } + RNBackgroundUpload.creationsInFlight[uploadId] = [] + RNBackgroundUpload.lock.unlock() + let settle = { + RNBackgroundUpload.lock.lock() + let waiters = RNBackgroundUpload.creationsInFlight.removeValue(forKey: uploadId) ?? [] + RNBackgroundUpload.lock.unlock() + resolve(uploadId) + for waiter in waiters { waiter(uploadId) } + } let group = DispatchGroup() let foundLock = NSLock() @@ -222,7 +305,64 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { } } group.notify(queue: .main) { - if exists { resolve(uploadId) } else { startNew() } + if !exists { startNew() } + settle() + } + } + + @objc(startChunkedUpload:resolve:reject:) + public func startChunkedUpload(_ options: [String: Any], + resolve: @escaping RCTPromiseResolveBlock, + reject: @escaping RCTPromiseRejectBlock) { + chunked.startUpload( + options, + resolve: { id in resolve(id) }, + reject: { message in reject("RN Uploader", message, nil) }) + } + + @objc(removeUpload:resolve:reject:) + public func removeUpload(_ uploadId: String, + resolve: @escaping RCTPromiseResolveBlock, + reject: @escaping RCTPromiseRejectBlock) { + // The chunked release runs first: it cancels the in-flight part tasks + // and deletes the manifest and the bytes. Then we cancel any simple task + // that wears this id. That cancel is kept out of the journal, because an + // explicit release is not an outcome that the consumer awaits. + chunked.remove(uploadId) { + RNBackgroundUpload.lock.lock() + RNBackgroundUpload.removedIds.insert(uploadId) + RNBackgroundUpload.lock.unlock() + let group = DispatchGroup() + let foundLock = NSLock() + var found = false + for session in self.activeSessions { + group.enter() + session.getAllTasks { tasks in + for task in tasks + where self.uploadId(session, task) == uploadId + && ChunkedCoordinator.partRef(session, task) == nil { + foundLock.lock() + found = true + foundLock.unlock() + task.cancel() + } + group.leave() + } + } + group.notify(queue: .main) { + foundLock.lock() + let matched = found + foundLock.unlock() + if !matched { + // Nothing was cancelled. Thus no delegate callback will consume + // the suppression. Drop it. If we keep it, a later upload that + // reuses this id has its real terminal swallowed. + RNBackgroundUpload.lock.lock() + RNBackgroundUpload.removedIds.remove(uploadId) + RNBackgroundUpload.lock.unlock() + } + resolve(nil) + } } } @@ -230,6 +370,18 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { public func cancelUpload(_ cancelUploadId: String, resolve: @escaping RCTPromiseResolveBlock, reject: @escaping RCTPromiseRejectBlock) { + // A chunked upload cancels through its coordinator: one 'cancelled' + // terminal for the whole upload, journaled before its part tasks are torn + // down. nil means that the id has no manifest. It then falls through to + // the simple-task path. + chunked.cancel(cancelUploadId) { handled in + if let handled { resolve(handled); return } + self.cancelSimpleUpload(cancelUploadId, resolve: resolve) + } + } + + private func cancelSimpleUpload(_ cancelUploadId: String, + resolve: @escaping RCTPromiseResolveBlock) { // Record intent before cancelling so the delegate reports cancelReason 'user'. RNBackgroundUpload.lock.lock() RNBackgroundUpload.userCancelledIds.insert(cancelUploadId) @@ -278,8 +430,16 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { public func ackEvents(_ eventIds: [String], resolve: @escaping RCTPromiseResolveBlock, reject: @escaping RCTPromiseRejectBlock) { + // An acked 'completed' is the ONE moment when a chunked upload's manifest + // and moved bytes may be deleted. Every other terminal keeps them for a + // resume. Find those uploads before the entries are removed. + let completedUploadIds = EventJournal.unacknowledgedEntries() + .filter { $0.type == "completed" && eventIds.contains($0.eventId) } + .map { $0.id } EventJournal.ack(eventIds) - resolve(true) + chunked.releaseCompleted(completedUploadIds) { // no-op for simple uploads + resolve(true) + } } @objc(getAllUploads:reject:) @@ -293,6 +453,9 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { group.enter() session.getAllTasks { tasks in for task in tasks { + // A chunked upload is one logical row, built from its manifest + // below. Its per-part tasks are transport detail. + if ChunkedCoordinator.partRef(session, task) != nil { continue } let id = self.uploadId(session, task) if id == "unknown" { continue } lock.lock() @@ -317,7 +480,14 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { group.leave() } } - group.notify(queue: .main) { resolve(result) } + group.notify(queue: .main) { + self.chunked.snapshots { chunkedRows in + lock.lock() + let combined = result + chunkedRows + lock.unlock() + resolve(combined) + } + } } // Called from AppDelegate.application(_:handleEventsForBackgroundURLSession:completionHandler:). @@ -329,13 +499,45 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { forIdentifier identifier: String) { // Touching `shared` recreates the background sessions when this is a fresh, // system-relaunched process, which is what lets the queued delegate events - // (and therefore this handler) actually fire. + // (and therefore this handler) actually fire. On a relaunch, it also + // claims the handler deferral (see below) BEFORE the handler is stored + // here. Thus the claim provably precedes any drain. _ = shared bgHandlerLock.lock() bgCompletionHandlers[identifier] = handler bgHandlerLock.unlock() } + /// For the chunked coordinator only. It parks every + /// urlSessionDidFinishEvents drain until the matching release. Thus the + /// system cannot suspend the app between a relaunch's replayed part + /// completions and the post-reconcile refill that enqueues the next part + /// tasks. + static func deferBackgroundCompletionHandlers() { + bgHandlerLock.lock() + bgHandlerDeferrals += 1 + bgHandlerLock.unlock() + } + + static func releaseBackgroundCompletionHandlers() { + bgHandlerLock.lock() + bgHandlerDeferrals -= 1 + var handlers: [() -> Void] = [] + if bgHandlerDeferrals <= 0 { + for identifier in bgSessionsAwaitingDrain { + if let handler = bgCompletionHandlers.removeValue(forKey: identifier) { + handlers.append(handler) + } + // A parked id with no stored handler is simply dropped. There is + // nothing to hold. If we keep it, a LATER wake's handler could drain + // before that wake's events were processed. + } + bgSessionsAwaitingDrain.removeAll() + } + bgHandlerLock.unlock() + for handler in handlers { DispatchQueue.main.async { handler() } } + } + // MARK: - URLSession delegate public func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) { @@ -356,6 +558,13 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { public func urlSession(_ session: URLSession, task: URLSessionTask, didSendBodyData bytesSent: Int64, totalBytesSent: Int64, totalBytesExpectedToSend: Int64) { + // A chunked part's bytes feed the upload's byte-weighted aggregate. A + // per-task percentage would have no meaning to the consumer. + if let ref = ChunkedCoordinator.partRef(session, task) { + chunked.partProgress(id: ref.id, part: ref.part, incarnation: ref.incarnation, + sent: totalBytesSent) + return + } // 0 rather than -1 when the length is unknown: the documented range is // 0-100, Android reports 0 for the same case, and a negative value renders // as a broken progress bar in a consumer that passes it straight through. @@ -387,6 +596,22 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { for (key, value) in http.allHeaderFields { headers["\(key)"] = "\(value)" } } + // A chunked part's outcome belongs to the coordinator of its upload: + // accept evaluation against the manifest, the window refill, and one + // journaled terminal, only when the whole upload settles. + if let ref = ChunkedCoordinator.partRef(session, task) { + RNBackgroundUpload.lock.lock() + let bodyData = RNBackgroundUpload.responsesData.removeValue(forKey: taskMapKey(session, task)) + RNBackgroundUpload.lock.unlock() + chunked.handlePartCompletion( + id: ref.id, part: ref.part, incarnation: ref.incarnation, + taskKey: taskMapKey(session, task), + statusCode: http != nil ? statusCode : nil, headers: headers, + body: bodyData.flatMap { String(data: $0 as Data, encoding: .utf8) }, + error: error as NSError?) + return + } + RNBackgroundUpload.lock.lock() let bodyData = RNBackgroundUpload.responsesData.removeValue(forKey: taskMapKey(session, task)) RNBackgroundUpload.lastProgressAt[id] = nil @@ -395,8 +620,17 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { // would otherwise linger for the life of the process and a later upload // reusing that id would report a system cancel as a user cancel. let userCancelled = RNBackgroundUpload.userCancelledIds.remove(id) != nil + let removed = RNBackgroundUpload.removedIds.remove(id) != nil RNBackgroundUpload.lock.unlock() + // removeUpload cancelled this task as an explicit release, not as an + // outcome that the consumer awaits. Journal nothing. A non-cancel + // terminal that only raced the removal still reports normally. + if removed, let nsError = error as NSError?, nsError.code == NSURLErrorCancelled { + TaskMap.removeKey(taskMapKey(session, task)) + return + } + let rawBody = bodyData.flatMap { String(data: $0 as Data, encoding: .utf8) } ?? "" let (cappedBody, truncated) = EventJournal.capBody(rawBody) let responseBody = cappedBody ?? "" @@ -412,9 +646,11 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { } if error == nil { - // "completed" only for 2xx or a per-request acceptStatus code; any other - // HTTP response is a terminal http error carrying the full response. - let accepted = (200..<300).contains(statusCode) || acceptStatus(session, task).contains(statusCode) + // "completed" only for a 2xx or a matching per-request accept rule. + // Any other HTTP response is a terminal http error that carries the + // full response. + let accepted = UploadOutcome.isAccepted( + statusCode, body: rawBody, accept: acceptRules(session, task)) if accepted { event.type = "completed" } else { @@ -434,22 +670,24 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { } } - // Journal BEFORE emitting; the emit is best-effort (JS may be dead). - EventJournal.append(event) TaskMap.removeKey(taskMapKey(session, task)) - - let body = event.bridged - let delegate = RNBackgroundUpload.currentDelegate - switch event.type { - case "completed": delegate?.emitCompleted(body) - case "cancelled": delegate?.emitCancelled(body) - default: delegate?.emitError(body) - } + // Journals BEFORE it emits. The emit is best-effort, because JS can be + // dead. + RNBackgroundUpload.journalAndEmit(event) } public func urlSessionDidFinishEvents(forBackgroundURLSession session: URLSession) { guard let identifier = session.configuration.identifier else { return } RNBackgroundUpload.bgHandlerLock.lock() + guard RNBackgroundUpload.bgHandlerDeferrals <= 0 else { + // A relaunch reconcile is in flight. If we hand the system the handler + // now, it can suspend the app before the refill enqueues new part + // tasks. The id is parked. releaseBackgroundCompletionHandlers drains + // it. + RNBackgroundUpload.bgSessionsAwaitingDrain.insert(identifier) + RNBackgroundUpload.bgHandlerLock.unlock() + return + } let handler = RNBackgroundUpload.bgCompletionHandlers.removeValue(forKey: identifier) RNBackgroundUpload.bgHandlerLock.unlock() if let handler { DispatchQueue.main.async { handler() } } @@ -457,8 +695,9 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate { // Classify a transport error to match Android's errorKind taxonomy: a missing or // unreadable source file -> 'file'; other URL-domain errors -> 'network'; anything - // else -> 'unknown'. - private static func errorKind(for error: NSError) -> String { + // else -> 'unknown'. It is internal because the chunked coordinator also + // classifies with it. + static func errorKind(for error: NSError) -> String { switch (error.domain, error.code) { case (NSURLErrorDomain, NSURLErrorFileDoesNotExist), (NSURLErrorDomain, NSURLErrorCannotOpenFile), diff --git a/ios/RNFileUploader.mm b/ios/RNFileUploader.mm index ab8c759f..3bf3c038 100644 --- a/ios/RNFileUploader.mm +++ b/ios/RNFileUploader.mm @@ -79,6 +79,13 @@ - (void)startUpload:(NSDictionary *)options [RNBackgroundUpload.shared startUpload:options resolve:resolve reject:reject]; } +- (void)startChunkedUpload:(NSDictionary *)options + resolve:(RCTPromiseResolveBlock)resolve + reject:(RCTPromiseRejectBlock)reject +{ + [RNBackgroundUpload.shared startChunkedUpload:options resolve:resolve reject:reject]; +} + - (void)cancelUpload:(NSString *)id resolve:(RCTPromiseResolveBlock)resolve reject:(RCTPromiseRejectBlock)reject @@ -86,20 +93,11 @@ - (void)cancelUpload:(NSString *)id [RNBackgroundUpload.shared cancelUpload:id resolve:resolve reject:reject]; } -// Chunked transport lands in the iOS engine change; until then the spec -// contract is met by rejecting. -- (void)startChunkedUpload:(NSDictionary *)options - resolve:(RCTPromiseResolveBlock)resolve - reject:(RCTPromiseRejectBlock)reject -{ - reject(@"E_NOT_IMPLEMENTED", @"Chunked uploads are not implemented in this build", nil); -} - - (void)removeUpload:(NSString *)id resolve:(RCTPromiseResolveBlock)resolve reject:(RCTPromiseRejectBlock)reject { - reject(@"E_NOT_IMPLEMENTED", @"removeUpload is not implemented in this build", nil); + [RNBackgroundUpload.shared removeUpload:id resolve:resolve reject:reject]; } - (void)getUnacknowledgedEvents:(RCTPromiseResolveBlock)resolve diff --git a/ios/TaskMap.swift b/ios/TaskMap.swift index 195b2a0a..2748b39d 100644 --- a/ios/TaskMap.swift +++ b/ios/TaskMap.swift @@ -1,23 +1,72 @@ import Foundation -// Durable ":" -> { id, acceptStatus } mapping. +// Durable ":" -> { id, accept, partIndex } mapping. // // Apple documents `taskDescription` only as an uninterpreted app string with no // guarantee it survives process death, and DTS guidance is to persist task // metadata externally keyed by the (stable) taskIdentifier. taskDescription -// stays the primary id; this map is the durable fallback so a task observed -// after relaunch is never orphaned under an unknown id, and so acceptStatus is -// still known when a task completes after the original startUpload options are gone. +// stays the primary id. This map is the durable fallback. Thus a task +// observed after a relaunch is never orphaned under an unknown id, and the +// accept rules are still known when a task completes after the original +// startUpload options are gone. Chunked part tasks carry `partIndex`. Their +// accept rules live in the manifest, so `accept` is nil for them. // // Synchronous serial-queue access; a single JSON file. enum TaskMap { struct Meta: Codable { let id: String - let acceptStatus: [Int] + // All are optional. Thus entries that older builds persisted still + // decode. + var accept: [UploadOutcome.AcceptRule]? + var partIndex: Int? + // Chunked part tasks only: the manifest incarnation that the task was + // created under. It mirrors the taskDescription encoding (see + // ChunkedEngine). + var incarnation: String? + + init(id: String, accept: [UploadOutcome.AcceptRule]?, partIndex: Int?, + incarnation: String? = nil) { + self.id = id + self.accept = accept + self.partIndex = partIndex + self.incarnation = incarnation + } + + private enum CodingKeys: String, CodingKey { + case id, accept, partIndex, incarnation + // Earlier builds persisted `acceptStatus: [Int]` where this build + // persists `accept` rules. The key is read, and never written. Thus a + // task that an older build enqueued keeps its accept rules when it + // completes under this build. This is the same legacy mapping as + // Android's Upload.normalized(). + case acceptStatus + } + + init(from decoder: Decoder) throws { + let c = try decoder.container(keyedBy: CodingKeys.self) + id = try c.decode(String.self, forKey: .id) + partIndex = try c.decodeIfPresent(Int.self, forKey: .partIndex) + incarnation = try c.decodeIfPresent(String.self, forKey: .incarnation) + accept = try c.decodeIfPresent([UploadOutcome.AcceptRule].self, forKey: .accept) + ?? c.decodeIfPresent([Int].self, forKey: .acceptStatus)? + .map { UploadOutcome.AcceptRule(status: $0, bodyIncludes: nil) } + } + + func encode(to encoder: Encoder) throws { + var c = encoder.container(keyedBy: CodingKeys.self) + try c.encode(id, forKey: .id) + try c.encodeIfPresent(accept, forKey: .accept) + try c.encodeIfPresent(partIndex, forKey: .partIndex) + try c.encodeIfPresent(incarnation, forKey: .incarnation) + } } private static let queue = DispatchQueue(label: "ai.openspace.rnbgupload.taskmap") + static func key(_ session: URLSession, _ task: URLSessionTask) -> String { + "\(session.configuration.identifier ?? ""):\(task.taskIdentifier)" + } + private static var fileURL: URL { let base = FileManager.default.urls(for: .applicationSupportDirectory, in: .userDomainMask)[0] try? FileManager.default.createDirectory(at: base, withIntermediateDirectories: true) diff --git a/ios/UploadOutcome.swift b/ios/UploadOutcome.swift new file mode 100644 index 00000000..a0e70f7c --- /dev/null +++ b/ios/UploadOutcome.swift @@ -0,0 +1,40 @@ +import Foundation + +// The pure classification of upload outcomes. It mirrors the Android +// UploadOutcome. This is the highest-consequence logic in the uploader, +// because it decides between success and failure. Thus it lives in a small +// unit with no session state, where it is easy to examine. +enum UploadOutcome { + + /// A non-2xx response to treat as success. `bodyIncludes` narrows the rule + /// by a response-body substring. This is necessary when one status has + /// several meanings, and only the message shows the difference (our + /// backend's 409). It is Codable: it is persisted in the chunked manifest + /// and in the TaskMap metadata. + struct AcceptRule: Codable, Equatable { + let status: Int + var bodyIncludes: String? + } + + // Tells whether an HTTP response counts as a successful completion. A 2xx + // always counts, plus any matching per-request accept rule. Anything else, + // 4xx and 5xx included, is an http error, not a completion. + static func isAccepted(_ code: Int, body: String?, accept: [AcceptRule]) -> Bool { + if (200..<300).contains(code) { return true } + return accept.contains { rule in + rule.status == code + && (rule.bodyIncludes == nil || body?.contains(rule.bodyIncludes!) == true) + } + } + + // The bridge delivers `accept` as an array of dictionaries. A malformed + // rule is dropped, not guessed at. JS validates the shape before it + // crosses. + static func parseAcceptRules(_ raw: Any?) -> [AcceptRule] { + guard let rules = raw as? [[String: Any]] else { return [] } + return rules.compactMap { rule in + guard let status = (rule["status"] as? NSNumber)?.intValue else { return nil } + return AcceptRule(status: status, bodyIncludes: rule["bodyIncludes"] as? String) + } + } +}