CRUD auto-upload does not fire when the internal ps_crud watch is the sole tableUpdates subscriber (1.14.x)

[khaledez]

Summary

Automatic CRUD upload does not fire on local writes when the SDK's internal upload-trigger watch is the only active tableUpdates subscriber. Queued CRUD ops sit in ps_crud indefinitely until something else forces an upload (e.g. a disconnect→connect). Adding any second concurrent db.watch on ps_crud reliably revives automatic uploads.

Reproduced on 1.14.1 and 1.14.4 (powersync-sqlite-core 0.4.13, CSQLite 3.51.2), iOS 26.5 Simulator, plain PowerSyncDatabase(schema:dbFilename:) (native connection pool, not GRDB).

Symptom

1. connect(connector:) — initial download sync completes, currentStatus.connected == true.
2. Perform a local write via writeTransaction (INSERT into a synced view → INSTEAD OF trigger writes ps_crud). ps_crud now has rows.
3. The connector's uploadData is never invoked automatically. currentStatus.connected stays true; the ops just sit in the queue (observed 60s+).
4. A disconnect() + connect() drains the queue immediately (via the connect-time upload path), so the connector and the server are fine.

Root cause (as far as I can tell from the source)

StreamingSyncClient.uploadLoop triggers uploads from an internal live query:

let watch = try db.watch(options: WatchOptions(sql: "SELECT 1 FROM ps_crud LIMIT 1",
                                               throttle: options.crudThrottle, ...))
    .dropFirst()
let allTriggers = AsyncAlgorithms.merge(watch, signals.signalCrudUpload.subscribe())
for try await _ in allTriggers { try await uploadAllCrud() ... }

watchImpl resolves the watched tables ({ps_crud}) and fires when pool.tableUpdates reports a changed table in that set. When this is the sole subscriber to the pool's tableUpdates (BroadcastStream), the loop does not observe ps_crud changes from writes — so uploadAllCrud() is never triggered.

Minimal repro / evidence

In an app-side probe I subscribed a second, no-op watch on the same table:

Task {
    for try await _ in try db.watch(sql: "SELECT count(*) FROM ps_crud",
                                    parameters: nil, mapper: { try $0.getInt(name: 0) }) {
        // no-op
    }
}

• Without this probe: a confirmed write leaves ps_crud non-empty for 60s+; uploadData never called.
• With this probe: the same write triggers uploadData within the throttle window and ps_crud drains to 0 (verified 4/4 attempts). Removing it reproduces the stall.

So the presence of a second tableUpdates subscriber is what makes the SDK's own upload watch fire.

Workaround

Keep an app-owned db.watch on ps_crud alive for the connected session (e.g. one that also drives a pending-count indicator). This restores automatic uploads reliably.

Environment

• powersync-swift: 1.14.1 and 1.14.4 (same behavior)
• powersync-sqlite-core-swift: 0.4.13, CSQLite: 3.51.2
• Platform: iOS 26.5 Simulator (iPad), Swift concurrency, SwiftUI
• DB: default PowerSyncDatabase(schema:dbFilename:), native (non-GRDB) connection pool
• Writes via writeTransaction/execute on synced views (standard INSTEAD OF → ps_crud)

Happy to provide a stripped-down sample project if useful.

[simolus3]

Thank you for the report! Unfortunately, I can't reproduce it in a small demo Swift executable. I had an LLM generate this to reproduce this, the idea is that we connect to a self-hosted demo, log uploadData calls and insert whenever I hit enter:

import Foundation
import PowerSync

final class MyBackendConnector: PowerSyncBackendConnectorProtocol {
    func fetchCredentials() async throws -> PowerSync.PowerSyncCredentials? {
        let url = URL(string: "http://localhost:6060/api/auth/token")!
        let (data, _) = try await URLSession.shared.data(from: url)
        struct TokenResponse: Decodable {
            let token: String
        }
        let response = try JSONDecoder().decode(TokenResponse.self, from: data)

        return PowerSyncCredentials(
            endpoint: "http://localhost:8080",
            token: response.token,
        )
    }

    func uploadData(database: any PowerSync.PowerSyncDatabaseProtocol) async throws {
        guard let tx = try await database.getNextCrudTransaction() else {
            return
        }
        print("Upload data was called: \(tx.crud.count) items")
        try await tx.complete();
    }
}

@main
struct App {
    static func main() async throws {
        let schema = Schema(tables: [
            Table(
                name: "todos",
                columns: [
                    .text("title")
                ]
            )
        ])
        let ps = PowerSyncDatabase(
            schema: schema,
        )

        async let _ = watchSyncStatus(db: ps)
        async let _ = watchStdin(db: ps)

        try await ps.connect(connector: MyBackendConnector())
        try await ps.waitForFirstSync()
        print("First sync complete")

        print("Running, press Ctrl+C to exit")
        await waitForSigint()
    }
}

private func watchSyncStatus(db: PowerSyncDatabaseProtocol) async {
    for await status in db.currentStatus.asFlow() {
        print("Sync status: Downloading \(status.downloading), last synced: \(status.lastSyncedAt)")
    }
}

private func watchStdin(db: PowerSyncDatabaseProtocol) async {
    print("Hit enter to add a new local item")

    for await _ in stdinLines() {
        do {
            try await db.execute("INSERT INTO todos (id, title) VALUES (uuid(), 'local entry')")
            print("Inserted entry")
        } catch {
            print("Failed to insert todo: \(error)")
        }
    }
}

// readLine() blocks synchronously, so it runs on a dedicated background thread
// rather than tying up the cooperative concurrency thread pool.
private func stdinLines() -> AsyncStream<String> {
    AsyncStream { continuation in
        let thread = Thread {
            while let line = readLine() {
                continuation.yield(line)
            }
            continuation.finish()
        }
        thread.start()
    }
}

private func waitForSigint() async {
    // Ignore the default SIGINT behavior so we can intercept it ourselves below.
    signal(SIGINT, SIG_IGN)
    let source = DispatchSource.makeSignalSource(signal: SIGINT, queue: .main)
    await withCheckedContinuation { continuation in
        source.setEventHandler {
            source.cancel()
            continuation.resume()
        }
        source.resume()
    }
}

Nothing in this demo watches a query, but when I swift run it, hitting enter adds a new local item and immediately logs Upload data was called.