From 266d3bc1ecf745c9f274ef67bba933851f174213 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Mon, 15 Sep 2025 16:34:17 +0200 Subject: [PATCH 1/7] Update core extension to 0.4.6 --- .github/workflows/test.yml | 3 ++- gradle/libs.versions.toml | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 230b1334..10da910f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -46,7 +46,8 @@ jobs: if: runner.os == 'macOS' uses: maxim-lobanov/setup-xcode@v1 with: - xcode-version: latest-stable + # TODO: Update to latest-stable once GH installs iOS 26 simulators + xcode-version: '^16.4.0' - name: Build and run tests with Gradle run: | diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 9224ed7f..6a0ed505 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -16,7 +16,7 @@ kotlinx-datetime = "0.7.1" kotlinx-io = "0.8.0" ktor = "3.2.3" uuid = "0.8.4" -powersync-core = "0.4.5" +powersync-core = "0.4.6" turbine = "1.2.1" kotest = "5.9.1" # we can't upgrade to 6.x because that requires Java 11 or above (we need Java 8 support) From ac5377f40ef010248495482020b0d46e11c78618 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Mon, 15 Sep 2025 16:52:15 +0200 Subject: [PATCH 2/7] Emit new connection events --- .../com/powersync/sync/SyncIntegrationTest.kt | 14 ++++++++ .../com/powersync/bucket/BucketStorage.kt | 35 ++++++++++++++++--- .../com/powersync/bucket/BucketStorageImpl.kt | 12 +------ .../kotlin/com/powersync/sync/SyncStream.kt | 5 ++- 4 files changed, 49 insertions(+), 17 deletions(-) diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt index 4d4db964..02a61f21 100644 --- a/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt +++ b/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt @@ -913,4 +913,18 @@ class NewSyncIntegrationTest : BaseSyncIntegrationTest(true) { query.cancelAndIgnoreRemainingEvents() } } + + @Test + fun `ends iteration on http close`() = databaseTest { + turbineScope(timeout = 10.0.seconds) { + val turbine = database.currentStatus.asFlow().testIn(this) + database.connect(TestConnector(), options = getOptions()) + turbine.waitFor { it.connected } + + syncLines.close() + turbine.waitFor { !it.connected } + + turbine.cancelAndIgnoreRemainingEvents() + } + } } diff --git a/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt b/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt index 0937cdac..bf4fa151 100644 --- a/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt +++ b/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt @@ -8,6 +8,7 @@ import com.powersync.sync.Instruction import com.powersync.sync.LegacySyncImplementation import com.powersync.sync.SyncDataBatch import com.powersync.sync.SyncLocalDatabaseResult +import com.powersync.utils.JsonUtil import kotlinx.serialization.Serializable import kotlinx.serialization.json.JsonObject @@ -55,25 +56,49 @@ internal interface BucketStorage { } internal sealed interface PowerSyncControlArguments { + /** + * Returns the arguments for the `powersync_control` SQL invocation. + */ + val sqlArguments: Pair + @Serializable class Start( val parameters: JsonObject, val schema: SerializableSchema, - ) : PowerSyncControlArguments + ) : PowerSyncControlArguments { + override val sqlArguments: Pair + get() = "start" to JsonUtil.json.encodeToString(this) + } - data object Stop : PowerSyncControlArguments + data object Stop : PowerSyncControlArguments { + override val sqlArguments: Pair = "stop" to null + } data class TextLine( val line: String, - ) : PowerSyncControlArguments + ) : PowerSyncControlArguments { + override val sqlArguments: Pair = "line_text" to line + } class BinaryLine( - val line: ByteArray, + line: ByteArray, ) : PowerSyncControlArguments { override fun toString(): String = "BinaryLine" + + override val sqlArguments: Pair = "line_binary" to line + } + + data object CompletedUpload : PowerSyncControlArguments { + override val sqlArguments: Pair = "completed_upload" to null } - data object CompletedUpload : PowerSyncControlArguments + data object ConnectionEstablished : PowerSyncControlArguments { + override val sqlArguments: Pair = "connection" to "established" + } + + data object ResponseStreamEnd : PowerSyncControlArguments { + override val sqlArguments: Pair = "connection" to "end" + } } @Serializable diff --git a/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt b/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt index 00331c37..b9c9e58a 100644 --- a/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt +++ b/core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt @@ -358,17 +358,7 @@ internal class BucketStorageImpl( db.writeTransaction { tx -> logger.v { "powersync_control: $args" } - val (op: String, data: Any?) = - when (args) { - is PowerSyncControlArguments.Start -> "start" to JsonUtil.json.encodeToString(args) - PowerSyncControlArguments.Stop -> "stop" to null - - PowerSyncControlArguments.CompletedUpload -> "completed_upload" to null - - is PowerSyncControlArguments.BinaryLine -> "line_binary" to args.line - is PowerSyncControlArguments.TextLine -> "line_text" to args.line - } - + val (op: String, data: Any?) = args.sqlArguments tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, data), ::handleControlResult) } } diff --git a/core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt b/core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt index 56b2c4b4..00ead40f 100644 --- a/core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt +++ b/core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt @@ -299,7 +299,6 @@ internal class SyncStream( throw RuntimeException("Received error when connecting to sync stream: ${httpResponse.bodyAsText()}") } - status.update { copy(connected = true, connecting = false) } block(isBson, httpResponse) } } @@ -307,6 +306,7 @@ internal class SyncStream( private fun receiveTextLines(req: JsonElement): Flow = flow { connectToSyncEndpoint(req, supportBson = false) { isBson, response -> + status.update { copy(connected = true, connecting = false) } check(!isBson) emitAll(response.body().lines()) @@ -316,6 +316,7 @@ internal class SyncStream( private fun receiveTextOrBinaryLines(req: JsonElement): Flow = flow { connectToSyncEndpoint(req, supportBson = false) { isBson, response -> + emit(PowerSyncControlArguments.ConnectionEstablished) val body = response.body() if (isBson) { @@ -323,6 +324,8 @@ internal class SyncStream( } else { emitAll(body.lines().map { PowerSyncControlArguments.TextLine(it) }) } + + emit(PowerSyncControlArguments.ResponseStreamEnd) } } From 26c0252a7acc8924013091806c3b9af1629f7a8d Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Tue, 16 Sep 2025 11:40:35 +0200 Subject: [PATCH 3/7] Reformat --- .../com/powersync/sync/SyncIntegrationTest.kt | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt index 02a61f21..84e69e21 100644 --- a/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt +++ b/core/src/commonIntegrationTest/kotlin/com/powersync/sync/SyncIntegrationTest.kt @@ -915,16 +915,17 @@ class NewSyncIntegrationTest : BaseSyncIntegrationTest(true) { } @Test - fun `ends iteration on http close`() = databaseTest { - turbineScope(timeout = 10.0.seconds) { - val turbine = database.currentStatus.asFlow().testIn(this) - database.connect(TestConnector(), options = getOptions()) - turbine.waitFor { it.connected } + fun `ends iteration on http close`() = + databaseTest { + turbineScope(timeout = 10.0.seconds) { + val turbine = database.currentStatus.asFlow().testIn(this) + database.connect(TestConnector(), options = getOptions()) + turbine.waitFor { it.connected } - syncLines.close() - turbine.waitFor { !it.connected } + syncLines.close() + turbine.waitFor { !it.connected } - turbine.cancelAndIgnoreRemainingEvents() + turbine.cancelAndIgnoreRemainingEvents() + } } - } } From ff28d2aee8039fafdbf94d2cc21e345474f65ba2 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Tue, 16 Sep 2025 12:13:07 +0200 Subject: [PATCH 4/7] Enable logs for attachment test --- .../kotlin/com/powersync/AttachmentsTest.kt | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt index db05a89b..9abffc98 100644 --- a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt +++ b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt @@ -93,6 +93,7 @@ class AttachmentsTest { * immediately be deleted. */ archivedCacheLimit = 0, + logger = logger, ) doOnCleanup { @@ -204,6 +205,7 @@ class AttachmentsTest { * immediately be deleted. */ archivedCacheLimit = 0, + logger = logger, ) doOnCleanup { @@ -314,6 +316,7 @@ class AttachmentsTest { */ archivedCacheLimit = 0, syncThrottleDuration = 0.seconds, + logger = logger, ) doOnCleanup { @@ -411,6 +414,7 @@ class AttachmentsTest { * Keep some items in the cache */ archivedCacheLimit = 10, + logger = logger, ) doOnCleanup { @@ -537,6 +541,7 @@ class AttachmentsTest { exception: Exception, ): Boolean = false }, + logger = logger, ) doOnCleanup { queue.stopSyncing() From 33be624e43e56bf224064e6b2110cd91c0795a2b Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Tue, 16 Sep 2025 12:31:47 +0200 Subject: [PATCH 5/7] More attachment logs --- .../powersync/attachments/AttachmentQueue.kt | 2 ++ .../implementation/AttachmentContextImpl.kt | 25 +++++++++++++------ 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/core/src/commonMain/kotlin/com/powersync/attachments/AttachmentQueue.kt b/core/src/commonMain/kotlin/com/powersync/attachments/AttachmentQueue.kt index c0dbfa04..5a6fa3ab 100644 --- a/core/src/commonMain/kotlin/com/powersync/attachments/AttachmentQueue.kt +++ b/core/src/commonMain/kotlin/com/powersync/attachments/AttachmentQueue.kt @@ -327,6 +327,8 @@ public open class AttachmentQueue( * Use a lock here to prevent conflicting state updates. */ attachmentsService.withContext { attachmentsContext -> + logger.v { "processWatchedAttachments($items)" } + /** * Need to get all the attachments which are tracked in the DB. * We might need to restore an archived attachment. diff --git a/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt b/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt index ecf52d37..1dbf5d19 100644 --- a/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt +++ b/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt @@ -1,13 +1,14 @@ package com.powersync.attachments.implementation import co.touchlab.kermit.Logger +import com.powersync.ExperimentalPowerSyncAPI import com.powersync.PowerSyncDatabase import com.powersync.attachments.Attachment import com.powersync.attachments.AttachmentContext import com.powersync.attachments.AttachmentState import com.powersync.db.getString import com.powersync.db.internal.ConnectionContext -import kotlinx.serialization.json.Json +import com.powersync.db.runWrapped import kotlin.time.Clock /** @@ -131,6 +132,7 @@ public open class AttachmentContextImpl( * @returns true if all items have been deleted. Returns false if there might be more archived * items remaining. */ + @OptIn(ExperimentalPowerSyncAPI::class) public override suspend fun deleteArchivedAttachments(callback: suspend (attachments: List) -> Unit): Boolean { // First fetch the attachments in order to allow other cleanup val limit = 1000 @@ -154,12 +156,21 @@ public open class AttachmentContextImpl( ), ) { Attachment.fromCursor(it) } callback(attachments) - db.execute( - "DELETE FROM $table WHERE id IN (SELECT value FROM json_each(?));", - listOf( - Json.encodeToString(attachments.map { it.id }), - ), - ) + + runWrapped { + db.useConnection(readOnly = false) { conn -> + conn.usePrepared("DELETE FROM $table WHERE id = ?") { stmt -> + for (attachment in attachments) { + stmt.bindText(1, attachment.id) + stmt.step() + + stmt.reset() + stmt.clearBindings() + } + } + } + } + return attachments.size < limit } From f2c0035b8e9b706562cf91811a066d8b9d7ec943 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Tue, 16 Sep 2025 12:44:55 +0200 Subject: [PATCH 6/7] Properly cancel turbine --- .../kotlin/com/powersync/AttachmentsTest.kt | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt index 9abffc98..0267575d 100644 --- a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt +++ b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt @@ -176,7 +176,7 @@ class AttachmentsTest { val exists = queue.localStorage.fileExists(localUri) exists shouldBe false - attachmentQuery.cancel() + attachmentQuery.cancelAndIgnoreRemainingEvents() } } @@ -286,7 +286,7 @@ class AttachmentsTest { // The file should have been deleted from storage queue.localStorage.fileExists(localUri) shouldBe false - attachmentQuery.cancel() + attachmentQuery.cancelAndIgnoreRemainingEvents() } } @@ -385,7 +385,7 @@ class AttachmentsTest { ) } - attachmentQuery.cancel() + attachmentQuery.cancelAndIgnoreRemainingEvents() } } @@ -497,7 +497,7 @@ class AttachmentsTest { attachmentRecord = attachmentQuery.awaitItem().first() attachmentRecord.state shouldBe AttachmentState.SYNCED - attachmentQuery.cancel() + attachmentQuery.cancelAndIgnoreRemainingEvents() } } @@ -579,7 +579,7 @@ class AttachmentsTest { attachmentRecord.state shouldBe AttachmentState.ARCHIVED - attachmentQuery.cancel() + attachmentQuery.cancelAndIgnoreRemainingEvents() } } } From 271d392da9b5738f04cb8334701b5e0bec262189 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Tue, 16 Sep 2025 13:34:55 +0200 Subject: [PATCH 7/7] Even more logs --- .../kotlin/com/powersync/AttachmentsTest.kt | 38 +++++++------------ .../implementation/AttachmentContextImpl.kt | 2 + .../attachments/sync/SyncingService.kt | 2 +- 3 files changed, 17 insertions(+), 25 deletions(-) diff --git a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt index 0267575d..4d063a2b 100644 --- a/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt +++ b/core/src/commonIntegrationTest/kotlin/com/powersync/AttachmentsTest.kt @@ -12,6 +12,7 @@ import com.powersync.attachments.createAttachmentsTable import com.powersync.db.getString import com.powersync.db.schema.Schema import com.powersync.db.schema.Table +import com.powersync.testutils.ActiveDatabaseTest import com.powersync.testutils.MockedRemoteStorage import com.powersync.testutils.UserRow import com.powersync.testutils.databaseTest @@ -27,7 +28,9 @@ import dev.mokkery.spy import dev.mokkery.verifySuspend import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.onEach import kotlinx.io.files.Path import kotlin.test.Test import kotlin.time.Duration.Companion.seconds @@ -53,6 +56,12 @@ class AttachmentsTest { ) } + private fun ActiveDatabaseTest.watchAttachmentsTable(): Flow> = + database + .watch("SELECT * FROM attachments") { + Attachment.fromCursor(it) + }.onEach { logger.i { "attachments table results: $it" } } + suspend fun updateSchema(db: PowerSyncDatabase) { db.updateSchema( Schema( @@ -76,11 +85,7 @@ class AttachmentsTest { val remote = spy(MockedRemoteStorage()) // Monitor the attachments table for testing - val attachmentQuery = - database - // language=SQL - .watch("SELECT * FROM attachments") { Attachment.fromCursor(it) } - .testIn(this) + val attachmentQuery = watchAttachmentsTable().testIn(this) val queue = AttachmentQueue( @@ -188,11 +193,7 @@ class AttachmentsTest { val remote = spy(MockedRemoteStorage()) // Monitor the attachments table for testing - val attachmentQuery = - database - // language=SQL - .watch("SELECT * FROM attachments") { Attachment.fromCursor(it) } - .testIn(this) + val attachmentQuery = watchAttachmentsTable().testIn(this) val queue = AttachmentQueue( @@ -298,11 +299,7 @@ class AttachmentsTest { val remote = spy(MockedRemoteStorage()) // Monitor the attachments table for testing - val attachmentQuery = - database - // language=SQL - .watch("SELECT * FROM attachments") { Attachment.fromCursor(it) } - .testIn(this) + val attachmentQuery = watchAttachmentsTable().testIn(this) val queue = AttachmentQueue( @@ -398,11 +395,7 @@ class AttachmentsTest { val remote = spy(MockedRemoteStorage()) // Monitor the attachments table for testing - val attachmentQuery = - database - // language=SQL - .watch("SELECT * FROM attachments") { Attachment.fromCursor(it) } - .testIn(this) + val attachmentQuery = watchAttachmentsTable().testIn(this) val queue = AttachmentQueue( @@ -513,10 +506,7 @@ class AttachmentsTest { } // Monitor the attachments table for testing - val attachmentQuery = - database - .watch("SELECT * FROM attachments") { Attachment.fromCursor(it) } - .testIn(this) + val attachmentQuery = watchAttachmentsTable().testIn(this) val queue = AttachmentQueue( diff --git a/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt b/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt index 1dbf5d19..6bf04c56 100644 --- a/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt +++ b/core/src/commonMain/kotlin/com/powersync/attachments/implementation/AttachmentContextImpl.kt @@ -66,6 +66,8 @@ public open class AttachmentContextImpl( } db.writeTransaction { tx -> + logger.v { "saveAttachments($attachments)" } + for (attachment in attachments) { upsertAttachment(attachment, tx) } diff --git a/core/src/commonMain/kotlin/com/powersync/attachments/sync/SyncingService.kt b/core/src/commonMain/kotlin/com/powersync/attachments/sync/SyncingService.kt index 5678d626..cf031027 100644 --- a/core/src/commonMain/kotlin/com/powersync/attachments/sync/SyncingService.kt +++ b/core/src/commonMain/kotlin/com/powersync/attachments/sync/SyncingService.kt @@ -78,7 +78,7 @@ public open class SyncingService( merge( // Handles manual triggers for sync events syncTriggerFlow.asSharedFlow(), - // Triggers the sync process whenever an underlaying change to the + // Triggers the sync process whenever an underlying change to the // attachments table happens attachmentsService .watchActiveAttachments(),