Proper way to subscribe to DB changes

[sproctor]

I'm trying to keep about 20 tables in sync on device and server. What I have works most of the time, except if the app is left running over night, it's giving intermittent failures when the access token needs to be refreshed. I just add the sessionStatus listener. I got some errors logged from Realtime, but the stack traces don't include the calling location.

supabase.auth.sessionStatus.collectLatest { sessionStatus ->
    if (sessionStatus is SessionStatus.Authenticated) {
        supabase.realtime.connect()
        supabase.realtime.status.collectLatest { status ->
            if (status == Realtime.Status.CONNECTED) {
                coroutineScope {
                    val channel = supabase.realtime.channel("db-changes")
                    try {
                        tables.forEach { table ->
                            channel.postgresChangeFlow<PostgresAction>(schema = "public") {
                              this.table = table
                              filter(FilterOperation("account_id", FilterOperator.EQ, accountId))
                          }
                                .onEach {
                                    syncTable(table)
                                }
                                .launchIn(this)
                        }
                        channel.subscribe(true)
                        syncAllTables()
                        awaitCancellation()
                    } finally {
                        withContext(NonCancellable) {
                            channel.unsubscribe()
                            supabase.realtime.removeChannel(channel)
                            supabase.realtime.disconnect()
                        }
                    }
                }
            }
        }
    }
}

[jan-tennert]

Even if the errors aren't that helpful, can you share them?

[sproctor]

These errors are de-obfuscated, and it's not perfect. vf I believe should be kotlinx.coroutines.

Error 1:

kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed
    at kotlinx.coroutines.channels.BufferedChannel.getReceiveException(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.resumeWaiterOnClosedChannel(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.closeLinkedList(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.completeClose(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.resumeReceiverOnClosedChannel(BufferedChannel.kt:2193)
    at kotlinx.coroutines.channels.BufferedChannel.cancelSuspendedReceiveRequests(BufferedChannel.kt:2186)
    at kotlinx.coroutines.channels.BufferedChannel.completeClose(BufferedChannel.kt:1963)
    at kotlinx.coroutines.channels.BufferedChannel.isClosed
    at vf.BufferedChannel.isClosed(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend0(BufferedChannel.kt:2217)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend(BufferedChannel.kt:2214)
    at kotlinx.coroutines.channels.BufferedChannel.completeCloseOrCancel(BufferedChannel.kt:1935)
    at kotlinx.coroutines.channels.BufferedChannel.closeOrCancelImpl(BufferedChannel.kt:1828)
    at kotlinx.coroutines.channels.BufferedChannel.close(BufferedChannel.kt:1787)
    at kotlinx.coroutines.channels.SendChannel$DefaultImpls.close$default
    at io.ktor.websocket.DefaultWebSocketSessionImpl$runIncomingProcessor$1.invokeSuspend(DefaultWebSocketSession.kt:249)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.CancellableContinuationImpl.getState$kotlinx_coroutines_core
    at kotlinx.coroutines.CancellableContinuationImpl.takeState$kotlinx_coroutines_core
    at kotlinx.coroutines.DispatchedTaskKt.resume(DebugStrings.kt:0)
    at kotlinx.coroutines.DispatchedTaskKt.resume(DebugStrings.kt:233)
    at kotlinx.coroutines.CancellableContinuationImpl.tryResume(CancellableContinuationImpl.kt:0)
    at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:0)
    at kotlinx.coroutines.DispatchedTaskKt.resumeUnconfined
    at kotlinx.coroutines.DispatchedTaskKt.dispatch
    at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:470)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$kotlinx_coroutines_core(CancellableContinuationImpl.kt:0)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$kotlinx_coroutines_core(CancellableContinuationImpl.kt:504)
    at kotlinx.coroutines.CompletionStateKt.toState
    at kotlinx.coroutines.CancellableContinuationImpl.resumeWith(CancellableContinuationImpl.kt:0)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$kotlinx_coroutines_core$default(CancellableContinuationImpl.kt:493)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeWith(CancellableContinuationImpl.kt:359)
    at kotlinx.coroutines.channels.BufferedChannel.resumeWaiterOnClosedChannel(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel$BufferedChannelIterator.tryResumeHasNextOnClosedChannel
    at kotlinx.coroutines.channels.BufferedChannel.resumeWaiterOnClosedChannel(BufferedChannel.kt:2206)
    at kotlinx.coroutines.channels.BufferedChannel.closeLinkedList(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.completeClose(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.resumeReceiverOnClosedChannel(BufferedChannel.kt:2193)
    at kotlinx.coroutines.channels.BufferedChannel.cancelSuspendedReceiveRequests(BufferedChannel.kt:2186)
    at kotlinx.coroutines.channels.BufferedChannel.completeClose(BufferedChannel.kt:1963)
    at kotlinx.coroutines.channels.BufferedChannel.isClosed
    at vf.BufferedChannel.isClosed(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend(BufferedChannel.kt:0)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend0(BufferedChannel.kt:2217)
    at kotlinx.coroutines.channels.BufferedChannel.isClosedForSend(BufferedChannel.kt:2214)
    at kotlinx.coroutines.channels.BufferedChannel.completeCloseOrCancel(BufferedChannel.kt:1935)
    at kotlinx.coroutines.channels.BufferedChannel.closeOrCancelImpl(BufferedChannel.kt:1828)
    at kotlinx.coroutines.channels.BufferedChannel.close(BufferedChannel.kt:1787)
    at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsKt.getCOROUTINE_SUSPENDED
    at io.ktor.websocket.RawWebSocketJvm$1.invokeSuspend(RawWebSocketJvm.kt:0)
    at kotlinx.coroutines.channels.SendChannel$DefaultImpls.close$default
    at io.ktor.websocket.RawWebSocketJvm$1.invokeSuspend(RawWebSocketJvm.kt:84)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:0)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
    at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:113)
    at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:0)
    at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:820)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704)

Error 2:

This one is a bit weird. I have a suspicion it might be a library bug where a CancellationException is caught and not re-thrown.

ChildCancelledException: Child of the scoped flow was cancelled

Error while trying to connect to realtime websocket. Trying again in 7s
URL: wss://....supabase.co/realtime/v1/websocket?apikey=...
xf.r: Child of the scoped flow was cancelled

[sproctor]

I've spent a couple of weeks trying to figure this one out. I think I finally have enough debug info to make some guesses.

I have a long-lived connection to synchronize data with the server. When the connection goes down, I re-subscribe and get whatever data I might have missed while it was down. The issue is happening on an Android tablet. I don't have enough data on the version running on JVM to determine if it happens there too.

Heartbeats are received until Error while listening for messages. Trying again in 7s. Channel was closed. 7 seconds later: Connected to realtime websocket! During this time, my subscriptions are shut down and restarted. After that, Realtime.status changes to DISCONNECTED and there are no more messages (of level info or higher).

Is it possible that scope is getting canceled? I'm not getting any heartbeat after that point. Ah ha! I had disconnectOnSessionLoss and disconnectOnNoSubscriptions set to true. I'm going to try not unsubscribing/resubscribing and just resync on reconnect.

[sproctor]

After making a change to never unsubscribe, I'm hitting an issue with expired tokens.

Sequence of events:
8:01:58 - access token updated (expires at 9:01)
8:07:01 - subscriptions are canceled
8:07:01 - device goes to sleep
8:07:01 - sessionStatus is initializing
9:02:56 - device wakes up
9:02:56 - subscriptions are launched
9:02:56 - "Connected to realtime websocket!"
9:02:56 - Session status is updated to token that expires at 10:03
9:02:56 - app independently calls realtime.connect() <--- Maybe this is the issue?
9:02:56 - Received message without event: RealtimeMessage(topic=realtime:db-changes, event=phx_reply, payload={"status":"error","response":{"reason":"Token has expired 59 seconds ago"}}, ref=null)
no activity here
9:03:09 - user closes the app

I was calling connect()/disconnect() on app resume and pause. I'm going to try removing those.

It looks like CHANNEL_EVENT_JOIN failed because the access token was outdated, then the access token is updated, but there's no mechanism to retry the CHANNEL_EVENT_JOIN.