Apply crud throttle, better error handling

Author: simolus3

Sources/PowerSync/Implementation/sync/HttpClient.swift

@@ -9,12 +9,23 @@ import Foundation
 /// a general-purpose HTTP client. 
 protocol HttpClient: Sendable {
     /// Start streaming a `/sync/stream` response body, emitting individual lines.
+    ///
+    /// Throws an ``UnexpectedResponseError`` if the response can't be interpreted as sync lines.
     func receiveSyncLines(request: URLRequest) async throws -> (HTTPURLResponse, any SyncLineResponse)
 
     /// Read a full response body.
     func readFully(request: URLRequest) async throws -> (HTTPURLResponse, Data)
 }
 
+struct UnexpectedResponseError: Error, CustomDebugStringConvertible {
+    let response: HTTPURLResponse
+    let message: String
+    
+    var debugDescription: String {
+        message
+    }
+}
+
 protocol SyncLineResponse: Sendable, AsyncSequence where AsyncIterator: SyncLineResponseIterator {}
 
 protocol SyncLineResponseIterator: AsyncIteratorProtocol {
@@ -31,11 +42,15 @@ struct PlatformHttpClient: HttpClient {
     let session: URLSession
 
     func receiveSyncLines(request: URLRequest) async throws -> (HTTPURLResponse, any SyncLineResponse) {
-        let (bytes, response) = try await session.bytes(for: request)
+        let (bytes, originalResponse) = try await session.bytes(for: request)
+        let response = originalResponse as! HTTPURLResponse
         let jsonStreamMimeType = "application/x-ndjson"
-        
+
         if response.mimeType != jsonStreamMimeType {
-            throw PowerSyncError.operationFailed(message: "Invalid sync lines response, (expected \(jsonStreamMimeType), got \(response.mimeType, default: "")")
+            throw UnexpectedResponseError(
+                response: response,
+                message: "Invalid sync lines response, (expected \(jsonStreamMimeType), got \(response.mimeType, default: "")"
+            )
         }
 
         struct PlatformSyncLineResponse<Base>: SyncLineResponse where Base : AsyncSequence, Base.Element == UInt8, Base: Sendable {
@@ -57,7 +72,7 @@ struct PlatformHttpClient: HttpClient {
             }
         }
 
-        return (response as! HTTPURLResponse, PlatformSyncLineResponse(lines: bytes.lines))
+        return (response, PlatformSyncLineResponse(lines: bytes.lines))
     }
 
     func readFully(request: URLRequest) async throws -> (HTTPURLResponse, Data) {

Sources/PowerSync/Implementation/sync/StreamingSyncClient.swift

@@ -38,7 +38,8 @@ final class StreamingSyncClient: Sendable {
 
     private func uploadLoop(signals: SyncSignals) async throws {
         // TODO: Replace with better watch mechanism once we've dropped the Kotlin dependency and can use onChange.
-        let watch = try db.watch(sql: "SELECT 1 FROM ps_crud LIMIT 1", parameters: [], mapper: { _ in () })
+        let options = WatchOptions(sql: "SELECT 1 FROM ps_crud LIMIT 1", throttle: options.crudThrottle, mapper: { _ in ()})
+        let watch = try db.watch(options: options)
             .dropFirst() // Skip initial result, we just want to watch changes
             .map { _ in () }
         let allTriggers = AsyncAlgorithms.merge(watch, signals.signalCrudUpload.subscribe())
@@ -131,18 +132,21 @@ The next upload iteration will be delayed.
             try tx.execute(sql: "UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name = '$local'", parameters: [opId])
         }
     }
-    
+
+    private func handleCommonResponseErrors(response: HTTPURLResponse) async {
+        if response.statusCode == 401 {
+            await self.invalidateCredentials()
+        }
+    }
+
     private func getWriteCheckpoint() async throws -> String {
         let clientId = try await db.get("SELECT powersync_client_id()") { try $0.getString(index: 0) }
         let (_, request) = try await authenticatedRequest { endpoint in
             endpoint.path += "/write-checkpoint2.json"
             endpoint.queryItems = [.init(name: "client_id", value: clientId)]
         }
         let (response, data) = try await httpClient.readFully(request: request)
-        
-        if response.statusCode == 401 {
-            await self.invalidateCredentials()
-        }
+        await self.handleCommonResponseErrors(response: response)
         if response.statusCode != 200 {
             throw PowerSyncError.operationFailed(message: "Error getting write checkpoint: \(response.statusCode)")
         }
@@ -209,14 +213,23 @@ The next upload iteration will be delayed.
         httpRequest.setValue("application/x-ndjson", forHTTPHeaderField: "Accept")
         httpRequest.httpBody = try StreamingSyncClient.jsonEncoder.encode(request)
 
-        let (response, stream) = try await httpClient.receiveSyncLines(request: httpRequest)
-        if response.statusCode == 401 {
-            await invalidateCredentials()
+        let response: HTTPURLResponse
+        let stream: any SyncLineResponse
+        do {
+            (response, stream) = try await httpClient.receiveSyncLines(request: httpRequest)
+        } catch {
+            if let responseError = error as? UnexpectedResponseError {
+                await handleCommonResponseErrors(response: responseError.response)
+            }
+
+            throw error
         }
+
+        await handleCommonResponseErrors(response: response)
         if response.statusCode != 200 {
             throw PowerSyncError.operationFailed(message: "POST \(url) failed with status code \(response.statusCode)")
         }
-        
+
         return ControlInvocationsFromStream(sequence: stream)
     }
 

Tests/PowerSyncTests/SyncTests.swift

@@ -1,4 +1,5 @@
 import AsyncAlgorithms
+import Foundation
 @testable import PowerSync
 import Testing
 
@@ -226,7 +227,37 @@ class InMemorySyncIntegrationTests {
         await waitForStatus(db.currentStatus) { $0.connected }
         try #require(await connector.fetchCredentialsCalls == 2)
     }
-    
+
+    @Test func handlesThrowing401Response() async throws {
+        final actor BackendConnector: PowerSyncBackendConnectorProtocol {
+            var fetchCredentialsCalls = 0
+
+            func fetchCredentials() async throws -> PowerSyncCredentials? {
+                fetchCredentialsCalls += 1
+                return testCredentials
+            }
+
+            func uploadData(database: any PowerSyncDatabaseProtocol) async throws {}
+        }
+
+        let connector = BackendConnector()
+        let channel = AsyncThrowingChannel<PowerSync.SyncLine, any Error>()
+        let db = openDatabase(MockHttpClient { request in
+            if await connector.fetchCredentialsCalls == 1 {
+                // On a real 401 response, the platform client would throw because the body can't be interpreted as sync lines.
+                // This verifies the sync client can recognize that and reset credentials.
+                let response = HTTPURLResponse(url: request.url!, statusCode: 401, httpVersion: nil, headerFields: nil)!
+                throw UnexpectedResponseError(response: response, message: "Expected error to retry fetching credentials")
+            } else {
+                return channel
+            }
+        })
+
+        try await db.connect(connector: connector, options: ConnectOptions(retryDelay: 0))
+        await waitForStatus(db.currentStatus) { $0.connected }
+        try #require(await connector.fetchCredentialsCalls == 2)
+    }
+
     @Test func tokenThrows() async throws {
         actor BackendConnector: PowerSyncBackendConnectorProtocol {
             var isFirstFetchCall = true