Make checkpoint during upload test more reliable

Author: simolus3

core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt

@@ -32,6 +32,8 @@ import io.kotest.matchers.shouldNotBe
 import io.kotest.matchers.string.shouldContain
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
 import kotlin.test.Test
@@ -482,10 +484,21 @@ abstract class BaseSyncIntegrationTest(
             turbineScope {
                 val turbine = database.currentStatus.asFlow().testIn(this)
                 syncLines.send(SyncLine.KeepAlive(1234))
-                turbine.waitFor { it.connected && !it.uploading }
+                turbine.waitFor { it.connected }
                 turbine.cancelAndIgnoreRemainingEvents()
             }
 
+            // Wait for the first upload task triggered when connecting to be complete.
+            withContext(Dispatchers.Default) {
+                waitFor {
+                    assertNotNull(
+                        logWriter.logs.find {
+                            it.message.contains("crud upload: notify completion")
+                        },
+                    )
+                }
+            }
+
             database.execute("INSERT INTO users (id, name, email) VALUES (uuid(), ?, ?)", listOf("local", "[email protected]"))
 
             expectUserRows(1)

core/src/commonIntegrationTest/kotlin/com/powersync/testutils/TestUtils.kt

@@ -9,6 +9,7 @@ import co.touchlab.kermit.Severity
 import co.touchlab.kermit.TestConfig
 import co.touchlab.kermit.TestLogWriter
 import com.powersync.DatabaseDriverFactory
+import com.powersync.PowerSyncTestLogWriter
 import com.powersync.TestConnector
 import com.powersync.bucket.WriteCheckpointData
 import com.powersync.bucket.WriteCheckpointResponse
@@ -73,8 +74,8 @@ internal class ActiveDatabaseTest(
     lateinit var database: PowerSyncDatabaseImpl
 
     val logWriter =
-        TestLogWriter(
-            loggable = Severity.Debug,
+        PowerSyncTestLogWriter(
+            loggable = Severity.Verbose,
         )
     val logger =
         Logger(

core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt

@@ -36,6 +36,7 @@ import io.ktor.utils.io.ByteReadChannel
 import io.ktor.utils.io.readUTF8Line
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.NonCancellable
@@ -122,7 +123,7 @@ internal class SyncStream(
     }
 
     fun triggerCrudUploadAsync(): Job =
-        uploadScope.launch {
+        uploadScope.launch(CoroutineName("triggerCrudUploadAsync")) {
             val thisIteration = PendingCrudUpload(CompletableDeferred())
             var holdingUploadLock = false
 

core/src/commonTest/kotlin/com/powersync/PowerSyncTestLogWriter.kt

@@ -0,0 +1,31 @@
+package com.powersync
+
+import co.touchlab.kermit.ExperimentalKermitApi
+import co.touchlab.kermit.LogWriter
+import co.touchlab.kermit.Severity
+import co.touchlab.kermit.TestLogWriter.LogEntry
+import kotlinx.atomicfu.locks.reentrantLock
+import kotlinx.atomicfu.locks.withLock
+
+/**
+ * A version of the `TestLogWriter` from Kermit that uses a mutex around logs instead of throwing
+ * for concurrent access.
+*/
+@OptIn(ExperimentalKermitApi::class)
+class PowerSyncTestLogWriter(private val loggable: Severity) : LogWriter() {
+    private val lock = reentrantLock()
+    private val _logs = mutableListOf<LogEntry>()
+
+    val logs: List<LogEntry>
+        get() = lock.withLock { _logs.toList() }
+
+    override fun isLoggable(tag: String, severity: Severity): Boolean {
+        return severity.ordinal >= loggable.ordinal
+    }
+
+    override fun log(severity: Severity, message: String, tag: String, throwable: Throwable?) {
+        lock.withLock {
+            _logs.add(LogEntry(severity, message, tag, throwable))
+        }
+    }
+}