Better logs for unknown requests

Author: simolus3

core/src/commonMain/kotlin/com/powersync/PowerSyncException.kt

@@ -2,5 +2,5 @@ package com.powersync
 
 public class PowerSyncException(
     message: String,
-    cause: Throwable,
+    cause: Throwable?,
 ) : Exception(message, cause)

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt

@@ -184,18 +184,7 @@ internal class BucketStorageImpl(
     override suspend fun control(op: String, payload: ByteArray): List<Instruction> {
         return db.writeTransaction { tx ->
             logger.v { "powersync_control($op, binary payload)" }
-
-            try {
-                tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
-            } catch (e: Exception) {
-                println("Got control exception, writing")
-                SystemFileSystem.sink(Path("/Users/simon/failing_line.bin")).buffered().apply {
-                    write(payload)
-                    flush()
-                    close()
-                }
-                throw e
-            }
+            tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
         }
     }
 }

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

@@ -13,6 +13,7 @@ import kotlinx.serialization.encoding.CompositeDecoder
 import kotlinx.serialization.encoding.Decoder
 import kotlinx.serialization.encoding.Encoder
 import kotlinx.serialization.encoding.decodeStructure
+import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.serializer
 
@@ -37,7 +38,7 @@ internal sealed interface Instruction {
     data object CloseSyncStream: Instruction
     data object DidCompleteSync: Instruction
 
-    data object UnknownInstruction: Instruction
+    data class UnknownInstruction(val raw: JsonElement?): Instruction
 
     class Serializer : KSerializer<Instruction> {
         private val logLine = serializer<LogLine>()
@@ -68,24 +69,25 @@ internal sealed interface Instruction {
                         2 -> decodeSerializableElement(descriptor, 2, establishSyncStream)
                         3 -> decodeSerializableElement(descriptor, 3, fetchCredentials)
                         4 -> {
-                            decodeSerializableElement(descriptor, 3, flushFileSystem)
+                            decodeSerializableElement(descriptor, 4, flushFileSystem)
                             FlushSileSystem
                         }
                         5 -> {
-                            decodeSerializableElement(descriptor, 4, closeSyncStream)
+                            decodeSerializableElement(descriptor, 5, closeSyncStream)
                             CloseSyncStream
                         }
                         6 -> {
-                            decodeSerializableElement(descriptor, 4, didCompleteSync)
+                            decodeSerializableElement(descriptor, 6, didCompleteSync)
                             DidCompleteSync
                         }
-                        CompositeDecoder.UNKNOWN_NAME, CompositeDecoder.DECODE_DONE -> UnknownInstruction
+                        CompositeDecoder.UNKNOWN_NAME,  -> UnknownInstruction(decodeSerializableElement(descriptor, index, serializer<JsonElement>()))
+                        CompositeDecoder.DECODE_DONE -> UnknownInstruction(null)
                         else -> error("Unexpected index: $index")
                     }
 
                 if (decodeElementIndex(descriptor) != CompositeDecoder.DECODE_DONE) {
                     // Sync lines are single-key objects, make sure there isn't another one.
-                    UnknownInstruction
+                    UnknownInstruction(null)
                 } else {
                     value
                 }
@@ -99,7 +101,7 @@ internal sealed interface Instruction {
 }
 
 @Serializable
-internal class CoreSyncStatus(
+internal data class CoreSyncStatus(
     val connected: Boolean,
     val connecting: Boolean,
     val downloading: CoreDownloadProgress?,
@@ -108,12 +110,12 @@ internal class CoreSyncStatus(
 )
 
 @Serializable
-internal class CoreDownloadProgress(
+internal data class CoreDownloadProgress(
     val buckets: Map<String, CoreBucketProgress>
 )
 
 @Serializable
-internal class CoreBucketProgress(
+internal data class CoreBucketProgress(
     val priority: BucketPriority,
     @SerialName("at_last")
     val atLast: Long,
@@ -124,7 +126,7 @@ internal class CoreBucketProgress(
 )
 
 @Serializable
-internal class CorePriorityStatus(
+internal data class CorePriorityStatus(
     val priority: BucketPriority,
     @SerialName("last_synced_at")
     @Serializable(with = InstantTimestampSerializer::class)

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

@@ -83,7 +83,7 @@ internal fun HttpClient.rSocketSyncStream(
 
     val rSocket = connector.connect(target)
     val syncStream = rSocket.requestStream(buildPayload {
-        data("{}")
+        data(JsonUtil.json.encodeToString(req))
         metadata(JsonUtil.json.encodeToString(RequestStreamMetadata("/sync/stream")))
     })
 

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

@@ -3,6 +3,7 @@ package com.powersync.sync
 import co.touchlab.kermit.Logger
 import co.touchlab.kermit.Severity
 import co.touchlab.stately.concurrency.AtomicBoolean
+import com.powersync.PowerSyncException
 import com.powersync.bucket.BucketStorage
 import com.powersync.bucket.WriteCheckpointResponse
 import com.powersync.connectors.PowerSyncBackendConnector
@@ -337,7 +338,9 @@ internal class SyncStream(
                 }
                 is Instruction.FetchCredentials -> TODO()
                 Instruction.DidCompleteSync -> status.update { copy(downloadError=null) }
-                Instruction.UnknownInstruction -> TODO()
+                is Instruction.UnknownInstruction -> {
+                    throw PowerSyncException("Unknown instruction received from core extension: ${instruction.raw}", null)
+                }
             }
         }
 

demos/supabase-todolist/desktopApp/src/jvmMain/kotlin/com/powersync/demos/Main.kt

@@ -7,9 +7,15 @@ import androidx.compose.ui.window.Window
 import androidx.compose.ui.window.WindowPosition
 import androidx.compose.ui.window.application
 import androidx.compose.ui.window.rememberWindowState
+import co.touchlab.kermit.Logger
+import co.touchlab.kermit.Severity
+import co.touchlab.kermit.platformLogWriter
 
 
 fun main() {
+    Logger.setLogWriters(platformLogWriter())
+    Logger.setMinSeverity(Severity.Verbose)
+
     application {
         Window(
             onCloseRequest = ::exitApplication,