fix(realtime): serialize channel subscription

Author: grdsdev

Sources/Realtime/RealtimeChannelV2.swift

@@ -37,6 +37,7 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
     var clientChanges: [PostgresJoinConfig] = []
     var joinRef: String?
     var pushes: [String: PushV2] = [:]
+    var subscribeTask: Task<Void, any Error>?
   }
 
   @MainActor
@@ -92,7 +93,22 @@ public final class RealtimeChannelV2: Sendable, RealtimeChannelProtocol {
   }
 
   /// Subscribes to the channel.
+  @MainActor
   public func subscribeWithError() async throws {
+    if let subscribeTask = mutableState.subscribeTask {
+      try await subscribeTask.value
+      return
+    }
+
+    mutableState.subscribeTask = Task {
+      defer { self.mutableState.subscribeTask = nil }
+      try await self.performSubscribeWithRetry()
+    }
+
+    return try await mutableState.subscribeTask!.value
+  }
+
+  private func performSubscribeWithRetry() async throws {
     logger?.debug(
       "Starting subscription to channel '\(topic)' (attempt 1/\(socket.options.maxRetryAttempts))"
     )

Sources/Realtime/RealtimeClientV2.swift

@@ -315,6 +315,7 @@ public final class RealtimeClientV2: Sendable, RealtimeClientProtocol {
       let realtimeTopic = "realtime:\(topic)"
 
       if let channel = $0.channels[realtimeTopic] {
+        self.options.logger?.debug("Reusing existing channel for topic: \(realtimeTopic)")
         return channel
       }
 

Tests/RealtimeTests/RealtimeChannelTests.swift

@@ -5,6 +5,7 @@
 //  Created by Guilherme Souza on 09/09/24.
 //
 
+import ConcurrencyExtras
 import InlineSnapshotTesting
 import TestHelpers
 import XCTest
@@ -196,6 +197,58 @@ final class RealtimeChannelTests: XCTestCase {
     // The subscription is still in progress when we clean up
   }
 
+  func testConcurrentSubscribeRunsSingleOperation() async throws {
+    let socket = ConcurrentSubscribeRealtimeClient()
+    let channel = RealtimeChannelV2(
+      topic: "test-topic",
+      config: RealtimeChannelConfig(
+        broadcast: BroadcastJoinConfig(),
+        presence: PresenceJoinConfig(),
+        isPrivate: false
+      ),
+      socket: socket,
+      logger: nil
+    )
+
+    async let firstSubscribe = channel.subscribeWithError()
+    async let secondSubscribe = channel.subscribeWithError()
+
+    try await waitForJoin(
+      on: socket,
+      expectedCount: 1
+    )
+    XCTAssertEqual(socket.joinPushCount, 1)
+
+    await channel.onMessage(
+      RealtimeMessageV2(
+        joinRef: nil,
+        ref: nil,
+        topic: channel.topic,
+        event: ChannelEvent.system,
+        payload: ["status": "ok"]
+      )
+    )
+
+    try await firstSubscribe
+    try await secondSubscribe
+
+    XCTAssertEqual(socket.joinPushCount, 1)
+  }
+
+  private func waitForJoin(
+    on socket: ConcurrentSubscribeRealtimeClient,
+    expectedCount: Int
+  ) async throws {
+    for _ in 0..<50 {
+      if socket.joinPushCount == expectedCount {
+        return
+      }
+      try await Task.sleep(nanoseconds: 10_000_000)
+    }
+
+    XCTFail("Timed out waiting for join push")
+  }
+
   func testHttpSendThrowsWhenAccessTokenIsMissing() async {
     let httpClient = HTTPClientMock()
     let (client, _) = FakeWebSocket.fakes()
@@ -439,7 +492,9 @@ final class RealtimeChannelTests: XCTestCase {
       XCTFail("Expected httpSend to throw an error on 503 status")
     } catch {
       // Should fall back to localized status text
-      XCTAssertTrue(error.localizedDescription.contains("503") || error.localizedDescription.contains("unavailable"))
+      XCTAssertTrue(
+        error.localizedDescription.contains("503")
+          || error.localizedDescription.contains("unavailable"))
     }
   }
 }
@@ -455,3 +510,50 @@ private struct BroadcastPayload: Decodable {
     let `private`: Bool
   }
 }
+
+private final class ConcurrentSubscribeRealtimeClient: RealtimeClientProtocol, @unchecked Sendable {
+  private let _pushedMessages = LockIsolated<[RealtimeMessageV2]>([])
+  private let _status = LockIsolated<RealtimeClientStatus>(.connected)
+
+  let options: RealtimeClientOptions
+  let http: any HTTPClientType
+  let broadcastURL = URL(string: "https://localhost:54321/realtime/v1/api/broadcast")!
+
+  init() {
+    self.options = RealtimeClientOptions(
+      headers: ["apikey": "test-key"],
+      timeoutInterval: 5.0
+    )
+    self.http = HTTPClientMock()
+  }
+
+  var status: RealtimeClientStatus {
+    _status.value
+  }
+
+  var pushedMessages: [RealtimeMessageV2] {
+    _pushedMessages.value
+  }
+
+  var joinPushCount: Int {
+    pushedMessages.filter { $0.event == ChannelEvent.join }.count
+  }
+
+  func connect() async {
+    _status.setValue(.connected)
+  }
+
+  func push(_ message: RealtimeMessageV2) {
+    _pushedMessages.withValue { $0.append(message) }
+  }
+
+  func _getAccessToken() async -> String? {
+    nil
+  }
+
+  func makeRef() -> String {
+    UUID().uuidString
+  }
+
+  func _remove(_: any RealtimeChannelProtocol) {}
+}