Wait out the iOS retry backoff when the system refuses a delayed task

CHANGELOG.md

@@ -21,6 +21,18 @@ caller's file into its queue, so this deleted the only copy. The file now
 stays, `enqueue` still rejects `E_STORAGE`, and a retry with the same id
 uses it, as Android already did.
 
+iOS now waits out a retry's backoff when the system refuses the retry
+before it starts. A retry is a background task with a begin date, so when
+`nsurlsessiond` refused the app's connection, the task failed at once, the
+next retry was created at once, and that one failed too: hundreds of
+attempts a second, until the app used all its CPU and gigabytes of memory.
+A delayed task that ends before its begin date with no response and no
+bytes sent never reached the server. It no longer counts as an attempt or
+emits an event; the rest of the delay is waited out in the app and the
+attempt is sent again under the same id. A chunked part does the same. As a
+backstop, a retry of a simple request never starts sooner than the base
+backoff after the previous attempt started.
+
 ## 11.1.0
 
 Added `takeDiagnostics()`. On Android it returns the module's recent

ios/ChunkedCoordinator.swift

@@ -126,6 +126,7 @@ final class ChunkedCoordinator {
   func partCompleted(id: String, part: Int, incarnation: String?, key: String, meta: TaskMap.Meta?,
                      completion c: TaskCompletion) {
     let owned = inFlight[id]?[part] == key
+    let beginAt = owned ? partBeginAt[id]?[part] : nil
     if owned {
       inFlight[id]?[part] = nil
       partSent[id]?[part] = nil
@@ -146,6 +147,16 @@ final class ChunkedCoordinator {
     }
     let cancelled = RetryClassifier.isCancellation(c.error)
     if cancelled, meta?.purpose == .pause || meta?.purpose == .superseded { return }
+    // A delayed part task that ended before its begin date, with no response
+    // and no body byte sent, never reached the daemon (see
+    // QueueCoordinator.taskCompletedLocked). No event and no rejection: keep
+    // its begin date and refill once it has passed. The new task takes a new
+    // ordinal, as every part task does.
+    if let beginAt, e.state == .running, q.now() < beginAt, c.statusCode == nil, c.bytesSent == 0 {
+      partBeginAt[id, default: [:]][part] = beginAt
+      q.schedule(Int((beginAt - q.now()).rounded(.up))) { [weak self] in self?.refill(id) }
+      return
+    }
     let accepted = c.error == nil
       && c.statusCode.map { UploadOutcome.isAccepted($0, body: c.body, accept: e.accept) } == true
     // A system cancel is not an attempt: no event. It retries below.

ios/QueueCoordinator+Outcomes.swift

@@ -78,6 +78,7 @@ extension QueueCoordinator {
     }
     disarmExpiry(id)
     throttle.reset(id)
+    attemptStarts[id] = nil
     return true
   }
 
@@ -130,6 +131,7 @@ extension QueueCoordinator {
     }
     disarmExpiry(id)
     throttle.reset(id)
+    attemptStarts[id] = nil
   }
 
   /// cancel() on a settled or legacy entry: the row, its bytes and its
@@ -158,6 +160,7 @@ extension QueueCoordinator {
     index.remove(id)
     disarmExpiry(id)
     throttle.reset(id)
+    attemptStarts[id] = nil
     if let aside { store.discard(aside) }
   }
 

ios/QueueCoordinator+Simple.swift

@@ -106,6 +106,7 @@ extension QueueCoordinator {
       return
     }
     liveTasks[task.key] = (id, task)
+    attemptStarts[id] = beginAt ?? t
     throttle.reset(id)
   }
 
@@ -124,6 +125,20 @@ extension QueueCoordinator {
     }
   }
 
+  /// A delayed task that never reached the daemon (see taskCompletedLocked).
+  /// Waits out the rest of its delay in this process, then issues the same
+  /// attempt again, unless something else moved the entry first.
+  private func waitOutNeverStarted(_ e: QueueEntry, beginAt: Double) {
+    let generation = e.generation
+    let attempts = e.attempts
+    schedule(Int((beginAt - now()).rounded(.up))) { [weak self] in
+      guard let self, let current = self.index.entry(e.id), current.generation == generation,
+            current.attempts == attempts, current.nextAttemptAt == beginAt,
+            !self.liveTasks.values.contains(where: { $0.id == e.id }) else { return }
+      self.issue(e.id, advanceAttempt: false)
+    }
+  }
+
   // MARK: - Delegate hooks
 
   /// didCompleteWithError. Synchronous, so the journal write for a terminal
@@ -161,6 +176,20 @@ extension QueueCoordinator {
     // is live. The server has the body then, so the entry completes, even
     // when the replacement failed and the entry moved to a later attempt.
     if meta?.purpose == .superseded && (!accepted || !e.isLive) { return }
+    // A delayed task that ended before its begin date, with no response and
+    // no body byte sent, never reached the daemon: nsurlsessiond refused the
+    // connection (a simulator build with no entitlements does this). It is
+    // not an attempt: no event, and it keeps its ordinal and request id.
+    // Retrying now would make another delayed task that fails at once, in a
+    // tight loop, so wait out the rest of the delay here. All three tests,
+    // not the time alone: a clock that moved during the wait must not turn a
+    // real failure into this. Not the error code: -1 comes with real
+    // failures too.
+    if e.state == .queued, let beginAt = e.nextAttemptAt, now() < beginAt, c.statusCode == nil,
+       c.bytesSent == 0 {
+      waitOutNeverStarted(e, beginAt: beginAt)
+      return
+    }
     // A cancel the library did not ask for (the system, a force-quit) is not
     // an attempt: no event. It retries below.
     if !cancelled {
@@ -277,16 +306,45 @@ extension QueueCoordinator {
 
   /// Backoff from the attempt count. A wait that would pass expiresAt
   /// settles 'expired' now instead of scheduling.
+  ///
+  /// A floor under the backoff, as a safety net: the next attempt never
+  /// begins within baseMs of the previous one's start, however that one
+  /// ended. One that ended before its own start (the daemon dropped a
+  /// delayed task at once, in a way taskCompletedLocked did not catch) gets
+  /// no task for baseMs, so such failures cannot spin. That wait is in this
+  /// process, and bounded, so a clock that moved cannot stall the retry.
+  /// The floor is in memory only: a relaunch starts without it.
   func scheduleRetry(_ e: QueueEntry) {
-    let delay = RetryClassifier.backoffMs(attempt: max(e.attempts, 1), policy: policy(e), random: random)
+    let base = policy(e).baseMs
+    var delay = RetryClassifier.backoffMs(attempt: max(e.attempts, 1), policy: policy(e), random: random)
+    let start = attemptStarts[e.id]
+    let endedBeforeStart = start.map { now() < $0 } ?? false
+    if let start {
+      delay = max(delay, endedBeforeStart ? Int(base) : Int((start + base - now()).rounded(.up)))
+    }
     if now() + Double(delay) >= e.expiresAt {
       settle(e.id, .expired)
       return
     }
     var n = e
     n.state = .queued
+    guard endedBeforeStart else {
+      index.upsert(n)
+      issue(n.id, delayMs: delay)
+      return
+    }
+    // The entry waits in memory with no task, as in deferIssue.
+    n.nextAttemptAt = nil
     index.upsert(n)
-    issue(n.id, delayMs: delay)
+    let beginAt = now() + Double(delay)
+    let generation = e.generation
+    let attempts = e.attempts
+    schedule(Int(base)) { [weak self] in
+      guard let self, let current = self.index.entry(e.id), current.generation == generation,
+            current.attempts == attempts, !self.liveTasks.values.contains(where: { $0.id == e.id }) else { return }
+      let remaining = Int((beginAt - self.now()).rounded(.up))
+      self.issue(e.id, delayMs: remaining > 0 ? remaining : nil)
+    }
   }
 
   /// awaiting-auth: no task, one `state` event. updateHeaders() resumes it.

ios/QueueCoordinator.swift

@@ -89,6 +89,9 @@ final class QueueCoordinator {
   /// Every task this process created or adopted, by TaskMap key.
   var liveTasks: [String: (id: String, task: UploadTask)] = [:]
   var expiryTokens: [String: UUID] = [:]
+  /// When each simple entry's latest task begins (its begin date, or when it
+  /// was made), by id. In memory only: the floor in scheduleRetry.
+  var attemptStarts: [String: Double] = [:]
   /// Bumped by scheduleReplace: only the latest scheduled replace runs.
   var replaceGeneration = 0
   /// Open grace waits (a completion that may still replay), by name. While

ios/RNBackgroundUpload.swift

@@ -283,7 +283,7 @@ public class RNBackgroundUpload: NSObject, URLSessionDataDelegate {
       key: key, description: task.taskDescription,
       url: task.originalRequest?.url?.absoluteString, statusCode: http?.statusCode,
       headers: headers, body: http == nil ? nil : body, bodyTruncated: truncated,
-      error: error as NSError?))
+      error: error as NSError?, bytesSent: task.countOfBytesSent))
   }
 
   public func urlSessionDidFinishEvents(forBackgroundURLSession session: URLSession) {

ios/Tests/CoordinatorChunkedTests.swift

@@ -222,6 +222,23 @@ final class CoordinatorChunkedTests: XCTestCase {
     XCTAssertEqual(h.entry("cap")?.state, .running)
   }
 
+  func testARefusedDelayedPartTaskIsNoAttemptAndWaitsOutItsBackoff() throws {
+    _ = try h.enqueue(h.chunkedRaw(id: "cap", size: 30, parts: 3)).get()
+    h.complete(try XCTUnwrap(partTask(0)), status: 503)
+    let delayed = try XCTUnwrap(partTask(0))
+    XCTAssertNotNil(delayed.beginAt)
+    let attempts = h.sink.attempts.count
+    // nsurlsessiond refused the connection: -1 at once, no response, no byte.
+    h.complete(delayed, error: NSError(domain: NSURLErrorDomain, code: NSURLErrorUnknown))
+    XCTAssertEqual(h.sink.attempts.count, attempts, "no attempt event")
+    XCTAssertEqual(h.entry("cap")?.parts[0].rejections, 1, "no rejection")
+    XCTAssertNil(partTask(0), "no task before the begin date")
+    h.advance(1_000)
+    let again = try XCTUnwrap(partTask(0))
+    XCTAssertNil(again.beginAt)
+    XCTAssertEqual(h.entry("cap")?.parts[0].rejections, 1)
+  }
+
   func testRowShowsNextAttemptAtWhenEveryPartWaits() throws {
     _ = try h.enqueue(h.chunkedRaw(id: "cap", size: 30, parts: 3)).get()
     h.complete(try XCTUnwrap(partTask(0)), status: 503)

ios/Tests/CoordinatorSimpleTests.swift

@@ -347,6 +347,8 @@ final class CoordinatorSimpleTests: XCTestCase {
     let retry = onlyTask()
     XCTAssertEqual(retry.beginAt, Date(timeIntervalSince1970: (h.clock + 1_000) / 1000))
     XCTAssertEqual(e.attempts, 2)
+    // The daemon begins the retry once its delay has passed.
+    h.advance(1_000)
     h.complete(retry, status: 503)
     XCTAssertEqual(h.entry("a")?.nextAttemptAt, h.clock + 2_000, "backoff doubles")
     h.complete(onlyTask(), status: 200)
@@ -394,6 +396,95 @@ final class CoordinatorSimpleTests: XCTestCase {
     XCTAssertNotNil(onlyTask().beginAt)
   }
 
+  // MARK: - Delayed tasks that never reach the daemon
+
+  /// NSURLErrorDomain -1, as nsurlsessiond reports a connection it refused.
+  private let refused = NSError(domain: NSURLErrorDomain, code: NSURLErrorUnknown)
+
+  /// What a simulator build with no entitlements sees: nsurlsessiond refuses
+  /// the app's connection, so every task ends at once with -1, no response
+  /// and no byte sent. Fails when one refusal makes another task at once,
+  /// over and over.
+  private func refuseEveryTask(file: StaticString = #filePath, line: UInt = #line) {
+    var count = 0
+    while let task = h.transport.live.first {
+      h.complete(task, error: refused)
+      count += 1
+      if count > 10 { return XCTFail("refused tasks retry in a tight loop", file: file, line: line) }
+    }
+  }
+
+  func testRefusedTasksRetryAtTheBackoffRateNotInATightLoop() throws {
+    _ = try h.enqueue(h.dataRaw(id: "a")).get()
+    for _ in 0..<100 { // 10 s in steps of 100 ms
+      refuseEveryTask()
+      h.advance(100)
+    }
+    // Attempts at 0, 1, 3 and 7 s (backoff 1, 2 and 4 s). The fifth waits for 15 s.
+    XCTAssertEqual(h.sink.attempts.count, 4)
+    XCTAssertEqual(h.entry("a")?.attempts, 5)
+    XCTAssertEqual(h.transport.created.count, 8, "each retry: the refused delayed task, then its re-issue")
+  }
+
+  func testARefusedDelayedTaskIsNoAttemptAndKeepsItsOrdinal() throws {
+    _ = try h.enqueue(h.dataRaw(id: "a")).get()
+    h.complete(onlyTask(), status: 503)
+    let delayed = onlyTask()
+    let beginAt = h.clock + 1_000
+    h.complete(delayed, error: refused)
+    XCTAssertEqual(h.sink.attempts.count, 1, "no attempt event")
+    XCTAssertEqual(h.entry("a")?.attempts, 2)
+    XCTAssertEqual(h.store.load("a")?.attempts, 2)
+    XCTAssertEqual(h.entry("a")?.state, .queued)
+    XCTAssertEqual(h.entry("a")?.nextAttemptAt, beginAt, "the wait stands")
+    h.advance(999)
+    XCTAssertTrue(h.transport.live.isEmpty, "no task before the begin date")
+    h.advance(1)
+    let again = onlyTask()
+    XCTAssertNil(again.beginAt)
+    XCTAssertEqual(again.header("X-Request-Id"), delayed.header("X-Request-Id"))
+    XCTAssertEqual(h.entry("a")?.attempts, 2)
+    XCTAssertEqual(h.entry("a")?.state, .running)
+  }
+
+  func testADelayedTaskThatFailsAfterItsBeginDateCounts() throws {
+    _ = try h.enqueue(h.dataRaw(id: "a")).get()
+    h.complete(onlyTask(), status: 503)
+    h.advance(1_000)
+    h.complete(onlyTask(), error: refused)
+    XCTAssertEqual(h.sink.attempts.count, 2)
+    XCTAssertEqual(h.entry("a")?.attempts, 3)
+    XCTAssertEqual(h.entry("a")?.nextAttemptAt, h.clock + 2_000, "backoff doubles")
+    XCTAssertEqual(onlyTask().beginAt, Date(timeIntervalSince1970: (h.clock + 2_000) / 1000))
+  }
+
+  /// A task that sent bytes or got a response reached the daemon, even if
+  /// it ended before its begin date (a clock that moved). It counts. The
+  /// floor holds its successor's task back for baseMs; the begin date keeps
+  /// the backoff.
+  func testADelayedTaskThatRanCountsEvenBeforeItsBeginDate() throws {
+    for ran in ["bytes", "response"] {
+      h = Harness()
+      h.boot()
+      _ = try h.enqueue(h.dataRaw(id: "a")).get()
+      h.complete(onlyTask(), status: 503)
+      let delayed = onlyTask()
+      let start = h.clock
+      if ran == "bytes" {
+        delayed.bytesSent = 3
+        h.complete(delayed, error: refused)
+      } else {
+        h.complete(delayed, status: 503)
+      }
+      XCTAssertEqual(h.sink.attempts.count, 2, ran)
+      XCTAssertEqual(h.entry("a")?.state, .queued, ran)
+      XCTAssertTrue(h.transport.live.isEmpty, "\(ran): the floor")
+      h.advance(1_000)
+      XCTAssertEqual(h.entry("a")?.attempts, 3, ran)
+      XCTAssertEqual(onlyTask().beginAt, Date(timeIntervalSince1970: (start + 2_000) / 1000), ran)
+    }
+  }
+
   func testBackoffPastExpiresAtSettlesExpired() throws {
     _ = try h.enqueue(h.dataRaw(id: "a", extra: ["expiresAt": h.clock + 500])).get()
     h.complete(onlyTask(), status: 503)
@@ -540,9 +631,11 @@ final class CoordinatorSimpleTests: XCTestCase {
     XCTAssertEqual(h.entry("a")?.transientHitsByRule, ["400|Bad Request": 1])
     XCTAssertEqual(h.sink.attempts.last?["outcome"] as? String, "error")
     XCTAssertEqual(h.sink.attempts.last?["httpCode"] as? Int, 400)
+    h.advance(1_000)
     h.complete(onlyTask(), status: 400, body: bodyStopped)
     XCTAssertEqual(h.entry("a")?.state, .queued)
     XCTAssertEqual(h.entry("a")?.transientHitsByRule, ["400|Bad Request": 2])
+    h.advance(2_000)
     h.complete(onlyTask(), status: 400, body: bodyStopped)
     XCTAssertEqual(h.entry("a")?.state, .error, "the third matching response is past the cap")
     let error = try XCTUnwrap(h.sink.settled.last?["error"] as? [String: Any])
@@ -582,7 +675,9 @@ final class CoordinatorSimpleTests: XCTestCase {
     _ = try h.enqueue(h.dataRaw(id: "a", extra: bodyStoppedRules)).get()
     h.complete(onlyTask(), status: 400, body: bodyStopped)
     XCTAssertEqual(h.entry("a")?.state, .queued, "the descriptor's own rule applies with no defaults")
+    h.advance(1_000)
     h.complete(onlyTask(), status: 400, body: bodyStopped)
+    h.advance(2_000)
     h.complete(onlyTask(), status: 400, body: bodyStopped)
     XCTAssertEqual(h.entry("a")?.state, .error)
     XCTAssertEqual(h.entry("a")?.transientHitsByRule, ["400|Bad Request": 2])
@@ -773,6 +868,7 @@ final class CoordinatorSimpleTests: XCTestCase {
     XCTAssertEqual(h.entry("a")?.attempts, 2, "no HTTP attempt ran")
     XCTAssertEqual(h.store.load("a")?.attempts, 2)
     XCTAssertEqual(onlyTask().header("X-Request-Id"), waiting.header("X-Request-Id"))
+    h.advance(1_000)
     h.complete(onlyTask(), status: 503)
     XCTAssertEqual(h.entry("a")?.nextAttemptAt, h.clock + 2_000, "the exponent follows real attempts")
   }

ios/Tests/TestSupport.swift

@@ -267,7 +267,7 @@ final class Harness {
     coordinator.taskCompleted(TaskCompletion(
       key: task.key, description: task.taskDescription, url: task.request.url?.absoluteString,
       statusCode: error == nil ? status : nil, headers: headers, body: error == nil ? body : nil,
-      bodyTruncated: false, error: error))
+      bodyTruncated: false, error: error, bytesSent: task.bytesSent))
   }
 
   /// Delivers the NSURLErrorCancelled callback of a task the test cancelled.

ios/Transport.swift

@@ -61,6 +61,8 @@ struct TaskCompletion {
   let body: String?
   let bodyTruncated: Bool
   let error: NSError?
+  /// countOfBytesSent when the task ended.
+  let bytesSent: Int64
 }
 
 /// Who a task belongs to. From taskDescription first, TaskMap second.