Fix race condition in WebSocketClient connection handling

[sampepose]

🔗 Issue Links

Provide all JIRA tickets and/or GitHub issues related to this PR, if applicable.

🎯 Goal

The current implementation uses shared mutable state (connected and timeout flags) that can be accessed concurrently from different threads, causing a data race when the WebSocket connects while the main task is checking connection status.

This can be seen from the logs:

WARNING: ThreadSanitizer: data race (pid=75749)
  Write of size 1 at 0x00010aad29f0 by thread T1:
    #0 closure #2 () -> () in StreamVideo.StreamVideo.(connectWebSocketClient in _57FBE95933FF925DEA7AF207F70F5588)() async throws -> () <null> (MyTestApp.debug.dylib:arm64+0x1554088)
    #1 partial apply forwarder for closure #2 () -> () in StreamVideo.StreamVideo.(connectWebSocketClient in _57FBE95933FF925DEA7AF207F70F5588)() async throws -> () <null> (MyTestApp.debug.dylib:arm64+0x15623e4)
    #2 StreamVideo.WebSocketClient.(handle in _EA224C36B0C3BE21B3EC405BB58853DD)(healthcheck: StreamVideo.WrappedEvent, info: StreamVideo.HealthCheckInfo) -> () <null> (MyTestApp.debug.dylib:arm64+0x18cff14)
    #3 StreamVideo.WebSocketClient.webSocketDidReceiveMessage(Foundation.Data) -> () <null> (MyTestApp.debug.dylib:arm64+0x18ceba8)
    #4 protocol witness for StreamVideo.WebSocketEngineDelegate.webSocketDidReceiveMessage(Foundation.Data) -> () in conformance StreamVideo.WebSocketClient : StreamVideo.WebSocketEngineDelegate in StreamVideo <null> (MyTestApp.debug.dylib:arm64+0x18d183c)
    #5 closure #2 @Sendable () -> () in closure #1 @Sendable (Swift.Result<(extension in Foundation):__C.NSURLSessionWebSocketTask.Message, Swift.Error>) -> () in StreamVideo.URLSessionWebSocketEngine.(doRead in _E102C83AE68006C0360AB6BB3CADE04A)() -> () <null> (MyTestApp.debug.dylib:arm64+0x18bfc14)
    #6 partial apply forwarder for closure #2 @Sendable () -> () in closure #1 @Sendable (Swift.Result<(extension in Foundation):__C.NSURLSessionWebSocketTask.Message, Swift.Error>) -> () in StreamVideo.URLSessionWebSocketEngine.(doRead in _E102C83AE68006C0360AB6BB3CADE04A)() -> () <null> (MyTestApp.debug.dylib:arm64+0x18c52a8)
    #7 reabstraction thunk helper from @escaping @callee_guaranteed @Sendable () -> () to @escaping @callee_unowned @convention(block) @Sendable () -> () <null> (MyTestApp.debug.dylib:arm64+0x10b740c)
    #8 __tsan::invoke_and_release_block(void*) <null> (libclang_rt.tsan_iossim_dynamic.dylib:arm64+0x7bccc)
    #9 _dispatch_client_callout <null> (libdispatch.dylib:arm64+0x5738)

  Previous read of size 1 at 0x00010aad29f0 by thread T5:
    #0 (3) suspend resume partial function for StreamVideo.StreamVideo.(connectWebSocketClient in _57FBE95933FF925DEA7AF207F70F5588)() async throws -> () <null> (MyTestApp.debug.dylib:arm64+0x15535cc)
    #1 swift::runJobInEstablishedExecutorContext(swift::Job*) <null> (libswift_Concurrency.dylib:arm64+0x38108)

  Location is heap block of size 17 at 0x00010aad29e0 allocated by thread T1:
    #0 malloc <null> (libclang_rt.tsan_iossim_dynamic.dylib:arm64+0x5397c)
    #1 _malloc_type_malloc_outlined <null> (libsystem_malloc.dylib:arm64+0xf3ec)
    #2 swift::runJobInEstablishedExecutorContext(swift::Job*) <null> (libswift_Concurrency.dylib:arm64+0x38108)

  Thread T1 (tid=35619471, running) is a GCD worker thread

  Thread T5 (tid=35619547, running) is a GCD worker thread

SUMMARY: ThreadSanitizer: data race (/Users/sampepose/Library/Developer/CoreSimulator/Devices/B6999A6E-7380-48B5-85B9-A4E835706191/data/Containers/Bundle/Application/C68BB6C7-D8BF-402F-927A-1187AA422791/MyTestApp.app/MyTestApp.debug.dylib:arm64+0x1554088) in closure #2 () -> () in StreamVideo.StreamVideo.(connectWebSocketClient in _57FBE95933FF925DEA7AF207F70F5588)() async throws -> ()+0x118

📝 Summary

I've refactored the connection handling to use Swift's modern concurrency features to eliminate the data race while maintaining the same timeout and connection logic behavior.

🛠 Implementation

• Replaced the polling loop with a task group that properly coordinates timeouts and connections
• Implemented a safe continuation pattern to prevent potential double-resume issues
• Used a shared handler reference that can be nullified to ensure the continuation is resumed exactly once
• Added proper weak self references to avoid potential retain cycles

🧪 Manual Testing Notes

1. Join a call successfully
2. Attempt to join a call, but time out (must wait the 30 seconds)

See no crashes or ThreadSanitizer complaints in either case.

☑️ Contributor Checklist

• I have signed the Stream CLA (required)
• This change follows zero ⚠️ policy (required)
• This change should receive manual QA
• Changelog is updated with client-facing changes
• New code is covered by unit tests
• Comparison screenshots added for visual changes
• Affected documentation updated (tutorial, CMS)

[sampepose]

Hey, I didn't see a CONTRIBUTORS file...hope it's ok to send this bug fix in for consideration!

[sampepose]

@martinmitrevski @ipavlidakis is there a process to get this reviewed?

[ipavlidakis]

Hi @sampepose,

Thank you for this PR. I would agree with you that this part is racey and requires improvement. However, may I suggest something around the following:

log.debug("Listening for WS connection")
let subject = PassthroughSubject<Void, Error>()
webSocketClient?.onConnected = { subject.send(()) }
try await subject.nextValue(timeout: 30)
log.debug("WS connected")

This one has the benefit of appearing simpler, being more readable, reusing logic that already exists in the SDK and avoids having many small rigid Task.sleep.

Best regards,
Ilias

[sampepose]

Good idea @ipavlidakis! I made those changes.