Combine framework retry after delay?

Question

I see how to use .retry directly, to resubscribe after an error, like this:

    URLSession.shared.dataTaskPublisher(for:url)
        .retry(3)

But that seems awfully simple-minded. What if I think that this error might go away if I wait awhile? I could insert a .delay operator, but then the delay operates even if there is no error. And there doesn't seem to be a way to apply an operator conditionally (i.e. only when there's an error).

I see how I could work around this by writing a RetryWithDelay operator from scratch, and indeed such an operator has been written by third parties. But is there a way to say "delay if there's an error", purely using the operators we're given?

My thought was that I could use .catch, because its function runs only if there is an error. But the function needs to return a publisher, and what publisher would we use? If we return somePublisher.delay(...) followed by .retry, we'd be applying .retry to the wrong publisher, wouldn't we?

Answer 1

I remembered that the RxSwiftExt library had a really nice implementation of a custom retry + delay operator with many options (linear and exponential delay, plus an option to provide a custom closure) and I tried to recreate it in Combine. The original implementation is here.

/**
 Provides the retry behavior that will be used - the number of retries and the delay between two subsequent retries.
 - `.immediate`: It will immediatelly retry for the specified retry count
 - `.delayed`: It will retry for the specified retry count, adding a fixed delay between each retry
 - `.exponentialDelayed`: It will retry for the specified retry count.
 The delay will be incremented by the provided multiplier after each iteration
 (`multiplier = 0.5` corresponds to 50% increase in time between each retry)
 - `.custom`: It will retry for the specified retry count. The delay will be calculated by the provided custom closure.
 The closure's argument is the current retry
 */
enum RetryBehavior<S> where S: Scheduler {
  case immediate(retries: UInt)
  case delayed(retries: UInt, time: TimeInterval)
  case exponentialDelayed(retries: UInt, initial: TimeInterval, multiplier: Double)
  case custom(retries: UInt, delayCalculator: (UInt) -> TimeInterval)
}

fileprivate extension RetryBehavior {
  
  func calculateConditions(_ currentRetry: UInt) -> (maxRetries: UInt, delay: S.SchedulerTimeType.Stride) {
    
    switch self {
    case let .immediate(retries):
      // If immediate, returns 0.0 for delay
      return (maxRetries: retries, delay: .zero)
    case let .delayed(retries, time):
      // Returns the fixed delay specified by the user
      return (maxRetries: retries, delay: .seconds(time))
    case let .exponentialDelayed(retries, initial, multiplier):
      // If it is the first retry the initial delay is used, otherwise it is calculated
      let delay = currentRetry == 1 ? initial : initial * pow(1 + multiplier, Double(currentRetry - 1))
      return (maxRetries: retries, delay: .seconds(delay))
    case let .custom(retries, delayCalculator):
      // Calculates the delay with the custom calculator
      return (maxRetries: retries, delay: .seconds(delayCalculator(currentRetry)))
    }
    
  }
  
}

public typealias RetryPredicate = (Error) -> Bool

extension Publisher {
  /**
   Retries the failed upstream publisher using the given retry behavior.
   - parameter behavior: The retry behavior that will be used in case of an error.
   - parameter shouldRetry: An optional custom closure which uses the downstream error to determine
   if the publisher should retry.
   - parameter tolerance: The allowed tolerance in firing delayed events.
   - parameter scheduler: The scheduler that will be used for delaying the retry.
   - parameter options: Options relevant to the scheduler’s behavior.
   - returns: A publisher that attempts to recreate its subscription to a failed upstream publisher.
   */
  func retry<S>(
    _ behavior: RetryBehavior<S>,
    shouldRetry: RetryPredicate? = nil,
    tolerance: S.SchedulerTimeType.Stride? = nil,
    scheduler: S,
    options: S.SchedulerOptions? = nil
  ) -> AnyPublisher<Output, Failure> where S: Scheduler {
    return retry(
      1,
      behavior: behavior,
      shouldRetry: shouldRetry,
      tolerance: tolerance,
      scheduler: scheduler,
      options: options
    )
  }
  
  private func retry<S>(
    _ currentAttempt: UInt,
    behavior: RetryBehavior<S>,
    shouldRetry: RetryPredicate? = nil,
    tolerance: S.SchedulerTimeType.Stride? = nil,
    scheduler: S,
    options: S.SchedulerOptions? = nil
  ) -> AnyPublisher<Output, Failure> where S: Scheduler {
    
    // This shouldn't happen, in case it does we finish immediately
    guard currentAttempt > 0 else { return Empty<Output, Failure>().eraseToAnyPublisher() }
    
    // Calculate the retry conditions
    let conditions = behavior.calculateConditions(currentAttempt)
    
    return self.catch { error -> AnyPublisher<Output, Failure> in
      
      // If we exceed the maximum retries we return the error
      guard currentAttempt <= conditions.maxRetries else {
        return Fail(error: error).eraseToAnyPublisher()
      }
      
      if let shouldRetry = shouldRetry, shouldRetry(error) == false {
        // If the shouldRetry predicate returns false we also return the error
        return Fail(error: error).eraseToAnyPublisher()
      }
      
      guard conditions.delay != .zero else {
        // If there is no delay, we retry immediately
        return self.retry(
          currentAttempt + 1,
          behavior: behavior,
          shouldRetry: shouldRetry,
          tolerance: tolerance,
          scheduler: scheduler,
          options: options
        )
        .eraseToAnyPublisher()
      }
      
      // We retry after the specified delay
      return Just(()).delay(for: conditions.delay, tolerance: tolerance, scheduler: scheduler, options: options).flatMap {
        return self.retry(
          currentAttempt + 1,
          behavior: behavior,
          shouldRetry: shouldRetry,
          tolerance: tolerance,
          scheduler: scheduler,
          options: options
        )
        .eraseToAnyPublisher()
      }
      .eraseToAnyPublisher()
    }
    .eraseToAnyPublisher()
  }
  
}

Answer 2

I found a few quirks with the implementations in the accepted answer.

• Firstly the first two attempts will be fired off without a delay since the first delay will only take effect after the second attempt.

• Secondly if any one of the retry attempts succeed, the output value will also delayed which seems unnecessary.

• Thirdly the extension is not flexible enough to allow the user to decide which scheduler it would like the retry attempts to be dispatched to.

After some tinkering back and forth I ended up with a solution like this:

public extension Publisher {
    /**
     Creates a new publisher which will upon failure retry the upstream publisher a provided number of times, with the provided delay between retry attempts.
     If the upstream publisher succeeds the first time this is bypassed and proceeds as normal.

     - Parameters:
        - retries: The number of times to retry the upstream publisher.
        - delay: Delay in seconds between retry attempts.
        - scheduler: The scheduler to dispatch the delayed events.

     - Returns: A new publisher which will retry the upstream publisher with a delay upon failure.

     ~~~
     let url = URL(string: "https://api.myService.com")!

     URLSession.shared.dataTaskPublisher(for: url)
         .retryWithDelay(retries: 4, delay: 5, scheduler: DispatchQueue.global())
         .sink { completion in
             switch completion {
             case .finished:
                 print("Success 😊")
             case .failure(let error):
                 print("The last and final failure after retry attempts: \(error)")
             }
         } receiveValue: { output in
             print("Received value: \(output)")
         }
         .store(in: &cancellables)
     ~~~
     */
    func retryWithDelay<S>(
        retries: Int,
        delay: S.SchedulerTimeType.Stride,
        scheduler: S
    ) -> AnyPublisher<Output, Failure> where S: Scheduler {
        self
            .delayIfFailure(for: delay, scheduler: scheduler)
            .retry(retries)
            .eraseToAnyPublisher()
    }

    private func delayIfFailure<S>(
        for delay: S.SchedulerTimeType.Stride,
        scheduler: S
    ) -> AnyPublisher<Output, Failure> where S: Scheduler {
        self.catch { error in
            Future { completion in
                scheduler.schedule(after: scheduler.now.advanced(by: delay)) {
                    completion(.failure(error))
                }
            }
        }
        .eraseToAnyPublisher()
    }
}

Answer 3

Using .catch is indeed the answer. We simply make a reference to the data task publisher and use that reference as the head of both pipelines — the outer pipeline that does the initial networking, and the inner pipeline produced by the .catch function.

Let's start by creating the data task publisher and stop:

let pub = URLSession.shared.dataTaskPublisher(for: url).share()

Now I can form the head of the pipeline:

let head = pub.catch {_ in pub.delay(for: 3, scheduler: DispatchQueue.main)}
    .retry(3)

That should do it! head is now a pipeline that inserts a delay operator only just in case there is an error. We can then proceed to form the rest of the pipeline, based on head.

Observe that we do indeed change publishers; if there is a failure and the catch function runs, the pub which is the upstream of the .delay becomes the publisher, replacing the pub we started out with. However, they are the same object (because I said share), so this is a distinction without a difference.

Answer 4

It was a topic of conversation on the Using Combine project repo a while back - the whole thread: https://github.com/heckj/swiftui-notes/issues/164.

The long and short was we made an example that I think does what you want, although it does use catch:

let resultPublisher = upstreamPublisher.catch { error -> AnyPublisher<String, Error> in
    return Publishers.Delay(upstream: upstreamPublisher,
                            interval: 3,
                            tolerance: 1,
                            scheduler: DispatchQueue.global())
    // moving retry into this block reduces the number of duplicate requests
    // In effect, there's the original request, and the `retry(2)` here will operate
    // two additional retries on the otherwise one-shot publisher that is initiated with
    // the `Publishers.Delay()` just above. Just starting this publisher with delay makes
    // an additional request, so the total number of requests ends up being 4 (assuming all
    // fail). However, no delay is introduced in this sequence if the original request
    // is successful.
    .retry(2)
    .eraseToAnyPublisher()
}

This is referencing the a retry pattern I have in the book/online, which is basically what you describe (but wasn't what you asked about).

The person I was corresponding with on the issue provided a variant in that thread as an extension that might be interesting as well:

extension Publisher {
  func retryWithDelay<T, E>()
    -> Publishers.Catch<Self, AnyPublisher<T, E>> where T == Self.Output, E == Self.Failure
  {
    return self.catch { error -> AnyPublisher<T, E> in
      return Publishers.Delay(
        upstream: self,
        interval: 3,
        tolerance: 1,
        scheduler: DispatchQueue.global()).retry(2).eraseToAnyPublisher()
    }
  }
}