[Realtime] RealtimeV2 broadcastStreams skip every other "Broadcast"

[mkaulfers]

Bug report

• I confirm this is a bug with Supabase, not with my own application.
• I confirm I have searched the Docs, GitHub Discussions, and Discord.

Describe the bug

There is an issue, with RealtimeV2 where performing a broadcast, is only received by listeners (broadcastStream) every other broadcast. For example n % 2 != 0 broadcasts. here is an example of a broadcaster, and a listener that I have setup in my project that reproduces this issue. It exists in an @Environment(DataManager.self) object, so it's never reinitialized while the app is running and I've confirmed that the task does indeed always exist.

struct ScanPayload: Codable, Equatable {
    let error: Bool
    let message: String
}

extension DataManager {
    func performBroadcast(with scanResult: ScanPayload, to userId: String) async throws {
        let scanChannel = await client.realtimeV2.channel("scan_broadcast_user_\(userId)")
        await scanChannel.subscribe()
        try await scanChannel.broadcast(event: "scan_result_\(userId)", message: scanResult)
    }
    
    func listenForScanResult(completion: @escaping (ScanPayload) -> Void) async {
        guard let profile else {
            return
        }
        
        let channel = await client.realtimeV2.channel("scan_broadcast_user_\(profile.userId.uuidString)")
        let broadcastStream = await channel.broadcastStream(event: "scan_result_\(profile.userId.uuidString)")
        await channel.subscribe()
        
        Task {
            for await message in broadcastStream {
                do {
                    let scanPayload = try message["payload"]?.decode(as: ScanPayload.self)
                    completion(scanPayload!)
                } catch {
                    let errorPayload = ScanPayload(error: true, message: "An error occurred while processing the scan.")
                    completion(errorPayload)
                }
            }
        }
    }

To Reproduce

Steps to reproduce the behavior, please provide code snippets or a repository:

• Create a function to performa. broadcast, to a specific channel, with a custom event and custom message.
• Create a function to listen for that broadcast, using the specific channel, decoding with a custom object.
• Setup functions with exact setup in example above; Tasks, async, etc..
• Perform a broadcast and observe that it will receive the 1, 3, 5, 6, etc.. broadcasts, but will not receive 2, 4, 6, etc.. broadcasts.

Expected behavior

• Expected to receive all broadcasts, when channels, and events are exactly the same.

System information

• OS: iOS
• Version of supabase-swift: 2.8.1

Additional context

What my code above is attempting to do, is scan a QR on one device, from another, and it should notify the scanned device that it was either successful or failed.

[grdsdev]

Hi @mkaulfers I made a few tests and I was able to implement the broadcast without losing any event.

Here is my test implementation:

import Supabase
import Foundation

let supabase = SupabaseClient(
  supabaseURL: URL(string: "https://PROJECT_KEY.supabase.co")!,
  supabaseKey: "ANON_KEY"
)

let channel = await supabase.channel("scan") {
  $0.broadcast.receiveOwnBroadcasts = true
}
await channel.subscribe()

struct Payload: Codable {
  let value: Int
}

Task {
  for await event in await channel.broadcastStream(event: "scan_result") {
    guard let payload = try? event["payload"]?.decode(as: Payload.self) else {
      print("Raw event", event)
      continue
    }

    print("Event received", payload)
  }
}

await Task.yield()

for i in 0..<Int.max {
  let message = Payload(value: i)
  try await channel.broadcast(event: "scan_result", message: message)
  print("Event sent", message)
  try await Task.sleep(for: .seconds(1))
}

---

Here are a few things I noticed in your code:

1. You recreate the channel every time you call the performBroadcast or listenForScanResult
   I'd recommend that you create the channel once and use it in the methods.

2. To receive your broadcast, you must set $0.broadcast.receiveOwnBroadcasts = true when creating the channel.

Please, review my sample and your code, and let me know what the results are.

Thanks.

[mkaulfers]

Okay so I played around with this quite a bit more and have some additional context. I Implemented the await Task.yield() which seems to have fixed it 90% and now I much more reliably get my responses. Is the expectation with broadcasts, to subscribe to the channel immediately, and always re-use it? I changed my code, such that I'm always re-using, rather than on-demand. That seemed to have a significant impact to the reliability as well. However, I am curious if we should be able to do it on-demand instead. I'm concerned about having many unique channels open whenever we're at scale.

[grdsdev]

I'm concerned about having many unique channels open whenever we're at scale.

If you're concerned about having too many open channels, you can use a single channel named scan_broadcast and then filter the user by the event name you subscribe to, scan_result_\(profile.userId.uuidString).