feat(realtime): subscribe retry improvements (#747)

* feat(realtime): add retry limit to subscribe

* feat(realtime): add exponential backoff with jitter and cancellation support

* use Encodable type for broadcast

* fix tests

* test: revamp realtime tests

* remove redundant test case

* fix(realtime): fix subscription retry logic and test failures

- Add missing return statement in subscribeWithError() to prevent infinite retries
- Add heartbeat response handling in testBehavior test
- Fix test expectations for retry attempts
- Ensure proper cleanup on subscription failure

This fixes the issue where successful subscriptions would continue retrying
and tests would fail due to incorrect retry attempt counting.

* add configurable option for maxRetryAttempts

Author: grdsdev

Sources/Helpers/Task+withTimeout.swift

@@ -10,7 +10,7 @@ import Foundation
 @discardableResult
 package func withTimeout<R: Sendable>(
   interval: TimeInterval,
-  @_inheritActorContext operation: @escaping @Sendable () async throws -> R
+  @_inheritActorContext operation: @escaping @Sendable () async -> R
 ) async throws -> R {
   try await withThrowingTaskGroup(of: R.self) { group in
     defer {
@@ -20,7 +20,7 @@ package func withTimeout<R: Sendable>(
     let deadline = Date(timeIntervalSinceNow: interval)
 
     group.addTask {
-      try await operation()
+      await operation()
     }
 
     group.addTask {

Sources/Realtime/RealtimeChannelV2.swift

@@ -91,9 +91,101 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
     callbackManager.reset()
   }
 
-  /// Subscribes to the channel
+  /// Subscribes to the channel.
+  public func subscribeWithError() async throws {
+    logger?.debug("Starting subscription to channel '\(topic)' (attempt 1/\(socket.options.maxRetryAttempts))")
+
+    status = .subscribing
+
+    defer {
+      // If the subscription fails, we need to set the status to unsubscribed
+      // to avoid the channel being stuck in a subscribing state.
+      if status != .subscribed {
+        status = .unsubscribed
+      }
+    }
+
+    var attempts = 0
+
+    while attempts < socket.options.maxRetryAttempts {
+      attempts += 1
+
+      do {
+        logger?.debug(
+          "Attempting to subscribe to channel '\(topic)' (attempt \(attempts)/\(socket.options.maxRetryAttempts))"
+        )
+
+        try await withTimeout(interval: socket.options.timeoutInterval) { [self] in
+          await _subscribe()
+        }
+
+        logger?.debug("Successfully subscribed to channel '\(topic)'")
+        return
+
+      } catch is TimeoutError {
+        logger?.debug(
+          "Subscribe timed out for channel '\(topic)' (attempt \(attempts)/\(socket.options.maxRetryAttempts))"
+        )
+
+        if attempts < socket.options.maxRetryAttempts {
+          // Add exponential backoff with jitter
+          let delay = calculateRetryDelay(for: attempts)
+          logger?.debug(
+            "Retrying subscription to channel '\(topic)' in \(String(format: "%.2f", delay)) seconds..."
+          )
+
+          do {
+            try await _clock.sleep(for: delay)
+          } catch {
+            // If sleep is cancelled, break out of retry loop
+            logger?.debug("Subscription retry cancelled for channel '\(topic)'")
+            throw CancellationError()
+          }
+        } else {
+          logger?.error(
+            "Failed to subscribe to channel '\(topic)' after \(socket.options.maxRetryAttempts) attempts due to timeout"
+          )
+        }
+      } catch is CancellationError {
+        logger?.debug("Subscription retry cancelled for channel '\(topic)'")
+        throw CancellationError()
+      } catch {
+        preconditionFailure(
+          "The only possible error here is TimeoutError or CancellationError, this should never happen."
+        )
+      }
+    }
+
+    logger?.error("Subscription to channel '\(topic)' failed after \(attempts) attempts")
+    throw RealtimeError.maxRetryAttemptsReached
+  }
+
+  /// Subscribes to the channel.
+  @available(*, deprecated, message: "Use `subscribeWithError` instead")
   @MainActor
   public func subscribe() async {
+    try? await subscribeWithError()
+  }
+
+  /// Calculates retry delay with exponential backoff and jitter
+  private func calculateRetryDelay(for attempt: Int) -> TimeInterval {
+    let baseDelay: TimeInterval = 1.0
+    let maxDelay: TimeInterval = 30.0
+    let backoffMultiplier: Double = 2.0
+
+    let exponentialDelay = baseDelay * pow(backoffMultiplier, Double(attempt - 1))
+    let cappedDelay = min(exponentialDelay, maxDelay)
+
+    // Add jitter (±25% random variation) to prevent thundering herd
+    let jitterRange = cappedDelay * 0.25
+    let jitter = Double.random(in: -jitterRange...jitterRange)
+
+    return max(0.1, cappedDelay + jitter)
+  }
+
+  /// Subscribes to the channel
+  @MainActor
+  private func _subscribe() async {
     if socket.status != .connected {
       if socket.options.connectOnSubscribe != true {
         reportIssue(
@@ -104,7 +196,6 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
       await socket.connect()
     }
 
-    status = .subscribing
     logger?.debug("Subscribing to channel \(topic)")
 
     config.presence.enabled = callbackManager.callbacks.contains(where: { $0.isPresence })
@@ -133,18 +224,7 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
       payload: try! JSONObject(payload)
     )
 
-    do {
-      try await withTimeout(interval: socket.options.timeoutInterval) { [self] in
-        _ = await statusChange.first { @Sendable in $0 == .subscribed }
-      }
-    } catch {
-      if error is TimeoutError {
-        logger?.debug("Subscribe timed out.")
-        await subscribe()
-      } else {
-        logger?.error("Subscribe failed: \(error)")
-      }
-    }
+    _ = await statusChange.first { @Sendable in $0 == .subscribed }
   }
 
   public func unsubscribe() async {
@@ -183,13 +263,6 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
   @MainActor
   public func broadcast(event: String, message: JSONObject) async {
     if status != .subscribed {
-      struct Message: Encodable {
-        let topic: String
-        let event: String
-        let payload: JSONObject
-        let `private`: Bool
-      }
-
       var headers: HTTPFields = [.contentType: "application/json"]
       if let apiKey = socket.options.apikey {
         headers[.apiKey] = apiKey
@@ -198,23 +271,34 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
         headers[.authorization] = "Bearer \(accessToken)"
       }
 
+      struct BroadcastMessagePayload: Encodable {
+        let messages: [Message]
+
+        struct Message: Encodable {
+          let topic: String
+          let event: String
+          let payload: JSONObject
+          let `private`: Bool
+        }
+      }
+
       let task = Task { [headers] in
         _ = try? await socket.http.send(
           HTTPRequest(
             url: socket.broadcastURL,
             method: .post,
             headers: headers,
             body: JSONEncoder().encode(
-              [
-                "messages": [
-                  Message(
+              BroadcastMessagePayload(
+                messages: [
+                  BroadcastMessagePayload.Message(
                     topic: topic,
                     event: event,
                     payload: message,
                     private: config.isPrivate
                   )
                 ]
-              ]
+              )
             )
           )
         )

Sources/Realtime/RealtimeError.swift

@@ -14,3 +14,10 @@ struct RealtimeError: LocalizedError {
     self.errorDescription = errorDescription
   }
 }
+
+extension RealtimeError {
+  /// The maximum retry attempts reached.
+  static var maxRetryAttemptsReached: Self {
+    Self("Maximum retry attempts reached.")
+  }
+}

Sources/Realtime/Types.swift

@@ -20,6 +20,7 @@ public struct RealtimeClientOptions: Sendable {
   var timeoutInterval: TimeInterval
   var disconnectOnSessionLoss: Bool
   var connectOnSubscribe: Bool
+  var maxRetryAttempts: Int
 
   /// Sets the log level for Realtime
   var logLevel: LogLevel?
@@ -32,6 +33,7 @@ public struct RealtimeClientOptions: Sendable {
   public static let defaultTimeoutInterval: TimeInterval = 10
   public static let defaultDisconnectOnSessionLoss = true
   public static let defaultConnectOnSubscribe: Bool = true
+  public static let defaultMaxRetryAttempts: Int = 5
 
   public init(
     headers: [String: String] = [:],
@@ -40,6 +42,7 @@ public struct RealtimeClientOptions: Sendable {
     timeoutInterval: TimeInterval = Self.defaultTimeoutInterval,
     disconnectOnSessionLoss: Bool = Self.defaultDisconnectOnSessionLoss,
     connectOnSubscribe: Bool = Self.defaultConnectOnSubscribe,
+    maxRetryAttempts: Int = Self.defaultMaxRetryAttempts,
     logLevel: LogLevel? = nil,
     fetch: (@Sendable (_ request: URLRequest) async throws -> (Data, URLResponse))? = nil,
     accessToken: (@Sendable () async throws -> String?)? = nil,
@@ -51,6 +54,7 @@ public struct RealtimeClientOptions: Sendable {
     self.timeoutInterval = timeoutInterval
     self.disconnectOnSessionLoss = disconnectOnSessionLoss
     self.connectOnSubscribe = connectOnSubscribe
+    self.maxRetryAttempts = maxRetryAttempts
     self.logLevel = logLevel
     self.fetch = fetch
     self.accessToken = accessToken

Tests/IntegrationTests/RealtimeIntegrationTests.swift

@@ -70,7 +70,11 @@ struct TestLogger: SupabaseLogger {
 
       await Task.yield()
 
-      await channel.subscribe()
+      do {
+        try await channel.subscribeWithError()
+      } catch {
+        XCTFail("Expected .subscribed but got error: \(error)")
+      }
 
       struct Message: Codable {
         var value: Int
@@ -141,7 +145,11 @@ struct TestLogger: SupabaseLogger {
 
       await Task.yield()
 
-      await channel.subscribe()
+      do {
+        try await channel.subscribeWithError()
+      } catch {
+        XCTFail("Expected .subscribed but got error: \(error)")
+      }
 
       struct UserState: Codable, Equatable {
         let email: String
@@ -201,7 +209,11 @@ struct TestLogger: SupabaseLogger {
       }
 
       await Task.yield()
-      await channel.subscribe()
+      do {
+        try await channel.subscribeWithError()
+      } catch {
+        XCTFail("Expected .subscribed but got error: \(error)")
+      }
 
       struct Entry: Codable, Equatable {
         let key: String

Tests/RealtimeTests/RealtimeChannelTests.swift

@@ -161,7 +161,7 @@ final class RealtimeChannelTests: XCTestCase {
     XCTAssertTrue(channel.callbackManager.callbacks.contains(where: { $0.isPresence }))
 
     // Start subscription process
-    Task {
+    let subscribeTask = Task {
       await channel.subscribe()
     }
 
@@ -191,5 +191,8 @@ final class RealtimeChannelTests: XCTestCase {
     presenceSubscription.cancel()
     await channel.unsubscribe()
     socket.disconnect()
+    
+    // Note: We don't assert the subscribe status here because the test doesn't wait for completion
+    // The subscription is still in progress when we clean up
   }
 }