Add database groups

simolus3 · simolus3 · commit dfd3ef03890b · 2026-04-27T12:27:28.000+02:00
diff --git a/Sources/PowerSync/Implementation/ActiveInstanceStore.swift b/Sources/PowerSync/Implementation/ActiveInstanceStore.swift
@@ -0,0 +1,65 @@
+final class DatabaseGroupCollection: Sendable {
+    private let groups: Mutex<[ActiveDatabaseGroupData]> = Mutex([])
+
+    fileprivate func closeGroup(identifier: String) {
+        groups.withLock { $0.removeAll { group in group.identifier == identifier } }
+    }
+
+    func referenceGroup(identifier: String, logger: LoggerProtocol) -> ActiveDatabaseGroup {
+        groups.withLock { activeDatabases in
+            let existingGroup = activeDatabases.first { $0.identifier == identifier }
+            let data: ActiveDatabaseGroupData
+            if let existingGroup {
+                logger.warning("""
+Multiple PowerSync instances for the same database have been detected.
+This can cause unexpected results.
+Please check your PowerSync client instantiation logic if this is not intentional.
+""", tag: "DatabaseGroupCollection")
+                data = existingGroup
+            } else {
+                data = ActiveDatabaseGroupData(identifier: identifier)
+                activeDatabases.append(data)
+            }
+
+            return ActiveDatabaseGroup(data: data, collection: self)
+        }
+    }
+    
+    static let shared = DatabaseGroupCollection()
+}
+
+private final class ActiveDatabaseGroupData: Sendable {
+    let identifier: String
+    let syncCoordinator = SyncCoordinator()
+
+    init(identifier: String) {
+        self.identifier = identifier
+    }
+}
+
+/// A collection  of PowerSync databases with the same path / identifier.
+///
+/// We expect that each group will only ever have one database because we encourage users to write their databases as
+/// singletons. We print a warning when two databasees are part of the same group.
+/// Additionally, we want to avoid two databases in the same group having a sync stream open at the same time to avoid
+/// duplicate resources being used. For this reason, each active database group has a single sync coordinator actor
+/// responsible for initializing the sync process for all databases in the group.
+final class ActiveDatabaseGroup: Sendable {
+    fileprivate let data: ActiveDatabaseGroupData
+    private weak let collection: DatabaseGroupCollection?
+
+    fileprivate init(data: ActiveDatabaseGroupData, collection: DatabaseGroupCollection) {
+        self.data = data
+        self.collection = collection
+    }
+
+    var syncCoordinator: SyncCoordinator {
+        data.syncCoordinator
+    }
+
+    deinit {
+        if let collection {
+            collection.closeGroup(identifier: self.data.identifier)
+        }
+    }
+}
diff --git a/Sources/PowerSync/Implementation/PowerSyncDatabaseImpl.swift b/Sources/PowerSync/Implementation/PowerSyncDatabaseImpl.swift
@@ -3,7 +3,7 @@ import Foundation
 
 final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
     let logger: any LoggerProtocol
-    let syncCoordinator = SyncCoordinator()
+    let group: ActiveDatabaseGroup
     let syncStatus = SwiftSyncStatus()
     private let dbFilename: String?
     private let httpClient: HttpClient
@@ -13,6 +13,8 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
 
     init(
         dbFilename: String? = nil,
+        identifier: String,
+        activeInstanceStore: DatabaseGroupCollection = .shared,
         logger: any LoggerProtocol,
         pool: any SQLiteConnectionPoolProtocol,
         httpClient: HttpClient,
@@ -23,14 +25,15 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
         self.schema = AsyncMutex(schema)
         self.httpClient = httpClient
         self.queries = ConnectionPoolQueries(pool: pool)
+        self.group = activeInstanceStore.referenceGroup(identifier: identifier, logger: logger)
     }
-    
+
     var currentStatus: any SyncStatus {
         syncStatus
     }
 
     func resolveOfflineSyncStatusIfNotConnected() async throws {
-        try await syncCoordinator.guardNotConnected(inner: {
+        try await group.syncCoordinator.guardNotConnected(inner: {
             try await resolveOfflineSyncStatus()
         }, ifConnected: {})
     }
@@ -56,7 +59,7 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
 
     func updateSchema(schema: any SchemaProtocol) async throws {
         try await initializer.ensureInitialized(db: self)
-        try await syncCoordinator.guardNotConnected(
+        try await group.syncCoordinator.guardNotConnected(
             inner: {
                 let schema = Schema(other: schema)
                 await self.schema.withMutex { $0 = schema }
@@ -99,7 +102,7 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
     }
     
     func disconnect() async throws {
-        await syncCoordinator.disconnect()
+        await group.syncCoordinator.disconnect()
     }
     
     func syncStream(name: String, params: JsonParam?) -> any SyncStream {
@@ -109,7 +112,7 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
     func close() async throws {
         try await initialize()
         try await initializer.close {
-            await syncCoordinator.disconnect()
+            await group.syncCoordinator.disconnect()
             try await queries.pool.close()
         }
     }
@@ -124,12 +127,12 @@ final class PowerSyncDatabaseImpl: PowerSyncDatabaseProtocol {
     }
 
     func connect(connector: any PowerSyncBackendConnectorProtocol, options: ConnectOptions?) async throws {
-        await syncCoordinator.connect(db: self, connector: connector, options: options ?? ConnectOptions(), client: httpClient)
+        await group.syncCoordinator.connect(db: self, connector: connector, options: options ?? ConnectOptions(), client: httpClient)
     }
 
     func disconnectAndClear(clearLocal: Bool, soft: Bool) async throws {
         try await initialize()
-        try await syncCoordinator.disconnectAndThen {
+        try await group.syncCoordinator.disconnectAndThen {
             var flags = 0
             if clearLocal {
                 flags |= 1
diff --git a/Sources/PowerSync/Implementation/SyncStreams.swift b/Sources/PowerSync/Implementation/SyncStreams.swift
@@ -88,11 +88,11 @@ struct PendingSyncStream: SyncStream {
     }
     
     func subscribe(ttl: TimeInterval?, priority: BucketPriority?) async throws -> any SyncStreamSubscription {
-        return try await db.syncCoordinator.streams.subscribe(db: db, stream: self, ttl: ttl, priority: priority)
+        return try await db.group.syncCoordinator.streams.subscribe(db: db, stream: self, ttl: ttl, priority: priority)
     }
     
     func unsubscribeAll() async throws {
-        let tracker = db.syncCoordinator.streams
+        let tracker = db.group.syncCoordinator.streams
         let key = self.key
         tracker.removeStreamGroup(key: key)
         try await tracker.subscriptionsCommand(db: db, request: .unsubscribe(key))
@@ -125,7 +125,7 @@ final class SyncSubscriptionImplementation: SyncStreamSubscription {
     }
     
     deinit {
-        db.syncCoordinator.streams.decrementRefCount(key: key)
+        db.group.syncCoordinator.streams.decrementRefCount(key: key)
     }
 }
 
diff --git a/Sources/PowerSync/Implementation/sqlite3/NativeConnectionPool.swift b/Sources/PowerSync/Implementation/sqlite3/NativeConnectionPool.swift
@@ -24,7 +24,6 @@ final class NativeConnectionPool: Sendable {
 
     private func dispatchWrites(lease: NativeConnectionLease) throws {
         try lease.withIterator(sql: "SELECT powersync_update_hooks('get')", parameters: []) { rows in
-            var rows = rows
             let affectedTables = try rows.next {
                 let decoder = JSONDecoder()
                 return try decoder.decode(Set<String>.self, from: try $0.getString(index: 0).data(using: .utf8)!)
diff --git a/Sources/PowerSync/Implementation/sync/StreamingSyncClient.swift b/Sources/PowerSync/Implementation/sync/StreamingSyncClient.swift
@@ -245,7 +245,7 @@ private struct ActiveSyncIteration: Sendable {
             parameters: syncClient.options.params,
             schema: await syncClient.db.schema.inner,
             includeDefaults: syncClient.options.includeDefaultStreams,
-            activeStreams: syncClient.db.syncCoordinator.streams.currentStreams,
+            activeStreams: syncClient.db.group.syncCoordinator.streams.currentStreams,
             appMetadata: syncClient.options.appMetadata,
         )))
 
@@ -357,7 +357,7 @@ private struct ActiveSyncIteration: Sendable {
     }
     
     private func watchSyncStreams() async throws {
-        let changes = syncClient.db.syncCoordinator.streams.streamsChanged.subscribe()
+        let changes = syncClient.db.group.syncCoordinator.streams.streamsChanged.subscribe()
         for await change in changes {
             self.localEvents.dispatch(event: .updateSubscriptions(streams: change))
         }
diff --git a/Sources/PowerSync/PowerSyncDatabase.swift b/Sources/PowerSync/PowerSyncDatabase.swift
@@ -16,14 +16,16 @@ public func PowerSyncDatabase(
     logger: (any LoggerProtocol) = DefaultLogger(),
     initialStatements: [String] = []
 ) -> PowerSyncDatabaseProtocol {
-    let location = if dbFilename == ":memory:" {
-        DatabaseLocation.inMemory
+    let (location, group) = if dbFilename == ":memory:" {
+        (DatabaseLocation.inMemory, DatabaseGroupCollection())
     } else {
-        DatabaseLocation.inDefaultDirectory(name: dbFilename)
+        (DatabaseLocation.inDefaultDirectory(name: dbFilename), .shared)
     }
     let pool = AsyncConnectionPool(location: location, initialStatements: initialStatements)
     return PowerSyncDatabaseImpl(
         dbFilename: dbFilename,
+        identifier: dbFilename,
+        activeInstanceStore: group,
         logger: logger,
         pool: pool,
         httpClient: PlatformHttpClient.shared,
@@ -53,6 +55,7 @@ public func OpenedPowerSyncDatabase(
     logger: (any LoggerProtocol) = DefaultLogger()
 ) -> PowerSyncDatabaseProtocol {
     return PowerSyncDatabaseImpl(
+        identifier: identifier,
         logger: logger,
         pool: pool,
         httpClient: PlatformHttpClient.shared,
diff --git a/Tests/PowerSyncTests/Kotlin/ActiveInstanceStoreTests.swift b/Tests/PowerSyncTests/Kotlin/ActiveInstanceStoreTests.swift
@@ -0,0 +1,37 @@
+@testable import PowerSync
+import Testing
+
+@Suite
+struct MultipleInstanceTest {
+    @Test func warnsAboutMultipleInstances() async throws {
+        let pool = AsyncConnectionPool(location: .inMemory)
+        let logWriter = TestLogWriterAdapter()
+        let logger = DefaultLogger(minSeverity: .warning, writers: [logWriter])
+        let schema = Schema()
+
+        let a = PowerSyncDatabaseImpl(identifier: "id", logger: logger, pool: pool, httpClient: PlatformHttpClient.shared, schema: schema)
+        try #require(logWriter.getLogs().isEmpty)
+
+        let b = PowerSyncDatabaseImpl(identifier: "id", logger: logger, pool: pool, httpClient: PlatformHttpClient.shared, schema: schema)
+        let _ = try #require(logWriter.getLogs().first { $0.contains("Multiple PowerSync instances for the same database have been detected.") })
+ 
+        // Ensure databases are kept around until the end of the test (if a gets closed before, we would't see the warning).
+        let _ = consume a
+        let _ = consume b
+    }
+    
+    @Test func doesNotWarnForClosedInstances() async throws {
+        let pool = AsyncConnectionPool(location: .inMemory)
+        let logWriter = TestLogWriterAdapter()
+        let logger = DefaultLogger(minSeverity: .warning, writers: [logWriter])
+        let schema = Schema()
+
+        do {
+            let _ = PowerSyncDatabaseImpl(identifier: "id2", logger: logger, pool: pool, httpClient: PlatformHttpClient.shared, schema: schema)
+        }
+
+        let b = PowerSyncDatabaseImpl(identifier: "id2", logger: logger, pool: pool, httpClient: PlatformHttpClient.shared, schema: schema)
+        try #require(logWriter.getLogs().isEmpty)
+        let _ = consume b
+    }
+}
diff --git a/Tests/PowerSyncTests/Kotlin/SqlCursorTests.swift b/Tests/PowerSyncTests/Kotlin/SqlCursorTests.swift
@@ -66,7 +66,8 @@ final class SqlCursorTests: XCTestCase {
         ])
 
         database = PowerSyncDatabaseImpl(
-            dbFilename: ":memory:",
+            identifier: ":memory:",
+            activeInstanceStore: DatabaseGroupCollection(),
             logger: DefaultLogger(),
             pool: AsyncConnectionPool(location: .inMemory),
             httpClient: PlatformHttpClient.shared,
diff --git a/Tests/PowerSyncTests/SyncTests.swift b/Tests/PowerSyncTests/SyncTests.swift
@@ -673,7 +673,8 @@ let defaultSchema = Schema(tables: [
 
 private func openDatabase(_ client: any HttpClient, schema: Schema = defaultSchema, logger: any LoggerProtocol = DefaultLogger()) -> PowerSyncDatabaseProtocol {
     return PowerSyncDatabaseImpl(
-        dbFilename: ":memory:",
+        identifier: ":memory:",
+        activeInstanceStore: DatabaseGroupCollection(),
         logger: logger,
         pool: AsyncConnectionPool(location: .inMemory),
         httpClient: client,

Original file line number	Diff line number	Diff line change
`@@ -88,11 +88,11 @@ struct PendingSyncStream: SyncStream {`
`88`	`88`	`}`
`89`	`89`
`90`	`90`	`func subscribe(ttl: TimeInterval?, priority: BucketPriority?) async throws -> any SyncStreamSubscription {`
`91`		`- return try await db.syncCoordinator.streams.subscribe(db: db, stream: self, ttl: ttl, priority: priority)`
	`91`	`+ return try await db.group.syncCoordinator.streams.subscribe(db: db, stream: self, ttl: ttl, priority: priority)`
`92`	`92`	`}`
`93`	`93`
`94`	`94`	`func unsubscribeAll() async throws {`
`95`		`- let tracker = db.syncCoordinator.streams`
	`95`	`+ let tracker = db.group.syncCoordinator.streams`
`96`	`96`	`let key = self.key`
`97`	`97`	`tracker.removeStreamGroup(key: key)`
`98`	`98`	`try await tracker.subscriptionsCommand(db: db, request: .unsubscribe(key))`
`@@ -125,7 +125,7 @@ final class SyncSubscriptionImplementation: SyncStreamSubscription {`
`125`	`125`	`}`
`126`	`126`
`127`	`127`	`deinit {`
`128`		`- db.syncCoordinator.streams.decrementRefCount(key: key)`
	`128`	`+ db.group.syncCoordinator.streams.decrementRefCount(key: key)`
`129`	`129`	`}`
`130`	`130`	`}`
`131`	`131`