[Legacy Sync] Fix offline upload delay

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 1.5.1 (unreleased)
+
+* Fix issue in legacy sync client where local writes made offline could have their upload delayed
+  until a keepalive event was received. This could also cause downloaded updates to be delayed even
+  further until all uploads were
+  completed.
+
 ## 1.5.0
 
 * Add `PowerSyncDatabase.getCrudTransactions()`, returning a flow of transactions. This is useful

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

@@ -612,7 +612,6 @@ abstract class BaseSyncIntegrationTest(
                     database.watch("SELECT name FROM users") { it.getString(0)!! }.testIn(scope)
                 query.awaitItem() shouldBe listOf("local write")
 
-                syncLines.send(SyncLine.KeepAlive(tokenExpiresIn = 1234))
                 syncLines.send(
                     SyncLine.FullCheckpoint(
                         Checkpoint(

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

@@ -536,9 +536,20 @@ internal class SyncStream(
             lateinit var receiveLines: Job
             receiveLines =
                 scope.launch {
+                    var hadLine = false
                     receiveTextLines(JsonUtil.json.encodeToJsonElement(req)).collect { value ->
                         val line = JsonUtil.json.decodeFromString<SyncLine>(value)
 
+                        if (!hadLine) {
+                            // Trigger a crud upload when receiving the first sync line: We could have
+                            // pending local writes made while disconnected, so in addition to listening on
+                            // updates to `ps_crud`, we also need to trigger a CRUD upload in some other
+                            // cases. We do this on the first sync line because the client is likely to be
+                            // online in that case.
+                            hadLine = true
+                            triggerCrudUploadAsync()
+                        }
+
                         state = handleInstruction(line, value, state)
 
                         if (state.abortIteration) {