Merge pull request #895 from sproctor/fix-cancellation-exceptions

Don't swallow `CancellationException`

Author: jan-tennert

Auth/src/commonMain/kotlin/io/github/jan/supabase/auth/Auth.kt

@@ -25,8 +25,10 @@ import io.github.jan.supabase.plugins.CustomSerializationPlugin
 import io.github.jan.supabase.plugins.MainPlugin
 import io.github.jan.supabase.plugins.SupabasePluginProvider
 import io.ktor.client.plugins.HttpRequestTimeoutException
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.serialization.json.JsonObject
+import kotlin.coroutines.coroutineContext
 
 /**
  * Plugin to interact with the Supabase Auth API
@@ -434,6 +436,7 @@ val SupabaseClient.auth: Auth
 private suspend fun Auth.tryToGetUser(jwt: String) = try {
     retrieveUser(jwt)
 } catch (e: Exception) {
+    coroutineContext.ensureActive()
     Auth.logger.e(e) { "Couldn't retrieve user using your custom jwt token. If you use the project secret ignore this message" }
     null
 }

Auth/src/commonMain/kotlin/io/github/jan/supabase/auth/AuthImpl.kt

@@ -42,6 +42,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -56,6 +57,7 @@ import kotlinx.serialization.json.encodeToJsonElement
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
 import kotlinx.serialization.json.put
+import kotlin.coroutines.coroutineContext
 import kotlin.time.Duration.Companion.seconds
 
 private const val SESSION_REFRESH_THRESHOLD = 0.8
@@ -466,6 +468,7 @@ internal class AuthImpl(
                 clearSession()
             }
         } catch (e: Exception) {
+            coroutineContext.ensureActive()
             Auth.logger.e(e) { "Couldn't reach Supabase. Either the address doesn't exist or the network might not be on. Retrying in ${config.retryDelay}..." }
             _sessionStatus.value = SessionStatus.RefreshFailure(RefreshFailureCause.NetworkError(e))
             delay(config.retryDelay)

Auth/src/settingsMain/kotlin/io/github/jan/supabase/auth/SettingsSessionManager.kt

@@ -5,9 +5,11 @@ import com.russhwolf.settings.Settings
 import com.russhwolf.settings.coroutines.toSuspendSettings
 import io.github.jan.supabase.auth.user.UserSession
 import io.github.jan.supabase.logging.e
+import kotlinx.coroutines.ensureActive
 import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonBuilder
+import kotlin.coroutines.coroutineContext
 
 private val settingsJson = Json {
     encodeDefaults = true
@@ -54,6 +56,7 @@ class SettingsSessionManager(
         return try {
             json.decodeFromString(session)
         } catch(e: Exception) {
+            coroutineContext.ensureActive()
             Auth.logger.e(e) { "Failed to load session" }
             null
         }

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeChannelImpl.kt

@@ -11,6 +11,7 @@ import io.github.jan.supabase.realtime.event.RealtimeEvent
 import io.ktor.client.statement.bodyAsText
 import io.ktor.http.headers
 import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -199,6 +200,7 @@ internal class RealtimeChannelImpl(
             val decodedValue = try {
                 supabaseClient.realtime.serializer.decode<T>(type, it.toString())
             } catch(e: Exception) {
+                coroutineContext.ensureActive()
                 Realtime.logger.e(e) { "Couldn't decode $it as $type. The corresponding handler wasn't called" }
                 null
             }

Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/RealtimeImpl.kt

@@ -23,6 +23,7 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -36,6 +37,7 @@ import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.jsonPrimitive
 import kotlinx.serialization.json.longOrNull
+import kotlin.coroutines.coroutineContext
 import kotlin.io.encoding.ExperimentalEncodingApi
 
 @PublishedApi internal class RealtimeImpl(override val supabaseClient: SupabaseClient, override val config: Realtime.Config) : Realtime {
@@ -83,6 +85,7 @@ import kotlin.io.encoding.ExperimentalEncodingApi
                 rejoinChannels()
             }
         } catch(e: Exception) {
+            coroutineContext.ensureActive()
             Realtime.logger.e(e) { """
                 Error while trying to connect to realtime websocket. Trying again in ${config.reconnectDelay}
                 URL: $websocketUrl
@@ -127,7 +130,7 @@ import kotlin.io.encoding.ExperimentalEncodingApi
                     }
                 }
             } catch(e: Exception) {
-                if(!isActive) return@launch
+                coroutineContext.ensureActive()
                 Realtime.logger.e(e) { "Error while listening for messages. Trying again in ${config.reconnectDelay}" }
                 reconnect()
             }
@@ -274,6 +277,7 @@ import kotlin.io.encoding.ExperimentalEncodingApi
         try {
             ws?.send(message)
         } catch(e: Exception) {
+            coroutineContext.ensureActive()
             Realtime.logger.e(e) { "Error while sending message $message. Reconnecting in ${config.reconnectDelay}" }
             reconnect()
         }

Storage/src/commonMain/kotlin/io/github/jan/supabase/storage/resumable/ResumableUpload.kt

@@ -28,6 +28,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
@@ -122,6 +123,7 @@ internal class ResumableUploadImpl(
                         dataStream.cancel() //cancel old data stream as we are start reading from a new offset
                         dataStream = createDataStream(offset) //create new data stream
                     } catch(e: Exception) {
+                        coroutineContext.ensureActive()
                         Storage.logger.e(e) { "Error while updating server offset for $path. Retrying in ${config.retryTimeout}" }
                         delay(config.retryTimeout)
                         continue
@@ -132,6 +134,7 @@ internal class ResumableUploadImpl(
                     val uploaded = uploadChunk()
                     offset += uploaded
                 } catch(e: Exception) {
+                    coroutineContext.ensureActive()
                     if(e !is IllegalStateException) {
                         Storage.logger.e(e) {"Error while uploading chunk. Retrying in ${config.retryTimeout}" }
                         delay(config.retryTimeout)

Supabase/src/commonMain/kotlin/io/github/jan/supabase/Utils.kt

@@ -61,7 +61,7 @@ suspend inline fun <reified T> HttpResponse.bodyOrNull(): T? {
     return try {
         val text = bodyAsText()
         supabaseJson.decodeFromString<T>(text)
-    } catch(e: Exception) {
+    } catch(_: Exception) {
         null
     }
 }

plugins/ComposeAuth/src/androidMain/kotlin/io/github/jan/supabase/compose/auth/composable/GoogleAuth.kt

@@ -26,6 +26,8 @@ import io.github.jan.supabase.compose.auth.hash
 import io.github.jan.supabase.compose.auth.signInWithGoogle
 import io.github.jan.supabase.logging.d
 import io.github.jan.supabase.logging.e
+import kotlinx.coroutines.ensureActive
+import kotlin.coroutines.coroutineContext
 
 private data class GoogleRequestOptions(
     val config: GoogleLoginConfig?,
@@ -100,6 +102,7 @@ internal fun ComposeAuth.signInWithCM(
                     }
                 }
             } catch (e: Exception) {
+                coroutineContext.ensureActive()
                 onResult.invoke(NativeSignInResult.Error(e.localizedMessage ?: "error", e))
                 ComposeAuth.logger.e(e) { "Error while logging into Supabase with Google ID Token Credential" }
             } finally {
@@ -135,6 +138,7 @@ private suspend fun parseCredential(
                         )
                     )
                 } catch (e: Exception) {
+                    coroutineContext.ensureActive()
                     ComposeAuth.logger.e(e) { "Error while logging into Supabase with Google ID Token Credential" }
                     onResult.invoke(
                         NativeSignInResult.Error(

plugins/ComposeAuth/src/appleMain/kotlin/io/github/jan/supabase/compose/auth/composable/AppleAuth.kt

@@ -12,6 +12,7 @@ import io.github.jan.supabase.compose.auth.hash
 import io.github.jan.supabase.compose.auth.signInWithApple
 import kotlinx.cinterop.BetaInteropApi
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.launch
 import platform.AuthenticationServices.ASAuthorization
 import platform.AuthenticationServices.ASAuthorizationAppleIDCredential
@@ -80,6 +81,7 @@ actual fun ComposeAuth.rememberSignInWithApple(
                     fallback.invoke()
                 }
             } catch (e: Exception) {
+                coroutineContext.ensureActive()
                 onResult.invoke(NativeSignInResult.Error(e.message ?: "error"))
             } finally {
                 state.reset()