Skip to content

Commit 568c3b1

Browse files
simolus3simolus3
authored
Merge pull request #170 from powersync-ja/fix-connect-on-open
Fix deadlock when connecting after opening database
2 parents c97a089 + 241f7fb commit 568c3b1

File tree

8 files changed

+116
-84
lines changed

8 files changed

+116
-84
lines changed

‎CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,10 @@
11
# Changelog
22

3+
## 1.0.0-BETA30
4+
5+
* Fix a deadlock when calling `connect()` immediately after opening a database.
6+
The issue has been introduced in version `1.0.0-BETA29`.
7+
38
## 1.0.0-BETA29
49

510
* Fix potential race condition between jobs in `connect()` and `disconnect()`.

‎core/src/commonIntegrationTest/kotlin/com/powersync/SyncIntegrationTest.kt‎

Lines changed: 27 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,26 @@ class SyncIntegrationTest {
3434
users shouldHaveSize amount
3535
}
3636

37+
@Test
38+
@OptIn(DelicateCoroutinesApi::class)
39+
fun connectImmediately() =
40+
databaseTest(createInitialDatabase = false) {
41+
// Regression test for https://github.com/powersync-ja/powersync-kotlin/issues/169
42+
val database = openDatabase()
43+
database.connect(connector)
44+
45+
turbineScope(timeout = 10.0.seconds) {
46+
val turbine = database.currentStatus.asFlow().testIn(this)
47+
turbine.waitFor { it.connected }
48+
turbine.cancel()
49+
}
50+
}
51+
3752
@Test
3853
@OptIn(DelicateCoroutinesApi::class)
3954
fun closesResponseStreamOnDatabaseClose() =
4055
databaseTest {
41-
val syncStream = database.syncStream()
42-
database.connectInternal(syncStream, 1000L)
56+
database.connect(connector)
4357

4458
turbineScope(timeout = 10.0.seconds) {
4559
val turbine = database.currentStatus.asFlow().testIn(this)
@@ -58,8 +72,7 @@ class SyncIntegrationTest {
5872
@OptIn(DelicateCoroutinesApi::class)
5973
fun cleansResourcesOnDisconnect() =
6074
databaseTest {
61-
val syncStream = database.syncStream()
62-
database.connectInternal(syncStream, 1000L)
75+
database.connect(connector)
6376

6477
turbineScope(timeout = 10.0.seconds) {
6578
val turbine = database.currentStatus.asFlow().testIn(this)
@@ -80,8 +93,7 @@ class SyncIntegrationTest {
8093
@Test
8194
fun cannotUpdateSchemaWhileConnected() =
8295
databaseTest {
83-
val syncStream = database.syncStream()
84-
database.connectInternal(syncStream, 1000L)
96+
database.connect(connector)
8597

8698
turbineScope(timeout = 10.0.seconds) {
8799
val turbine = database.currentStatus.asFlow().testIn(this)
@@ -99,8 +111,7 @@ class SyncIntegrationTest {
99111
@Test
100112
fun testPartialSync() =
101113
databaseTest {
102-
val syncStream = database.syncStream()
103-
database.connectInternal(syncStream, 1000L)
114+
database.connect(connector)
104115

105116
val checksums =
106117
buildList {
@@ -146,7 +157,7 @@ class SyncIntegrationTest {
146157
}
147158

148159
turbineScope(timeout = 10.0.seconds) {
149-
val turbine = syncStream.status.asFlow().testIn(this)
160+
val turbine = database.currentStatus.asFlow().testIn(this)
150161
turbine.waitFor { it.connected }
151162
database.expectUserCount(0)
152163

@@ -191,8 +202,7 @@ class SyncIntegrationTest {
191202
@Test
192203
fun testRemembersLastPartialSync() =
193204
databaseTest {
194-
val syncStream = database.syncStream()
195-
database.connectInternal(syncStream, 1000L)
205+
database.connect(connector)
196206

197207
syncLines.send(
198208
SyncLine.FullCheckpoint(
@@ -228,8 +238,7 @@ class SyncIntegrationTest {
228238
@Test
229239
fun setsDownloadingState() =
230240
databaseTest {
231-
val syncStream = database.syncStream()
232-
database.connectInternal(syncStream, 1000L)
241+
database.connect(connector)
233242

234243
turbineScope(timeout = 10.0.seconds) {
235244
val turbine = database.currentStatus.asFlow().testIn(this)
@@ -261,10 +270,9 @@ class SyncIntegrationTest {
261270
fun setsConnectingState() =
262271
databaseTest {
263272
turbineScope(timeout = 10.0.seconds) {
264-
val syncStream = database.syncStream()
265273
val turbine = database.currentStatus.asFlow().testIn(this)
266274

267-
database.connectInternal(syncStream, 1000L)
275+
database.connect(connector)
268276
turbine.waitFor { it.connecting }
269277

270278
database.disconnect()
@@ -277,8 +285,7 @@ class SyncIntegrationTest {
277285
@Test
278286
fun testMultipleSyncsDoNotCreateMultipleStatusEntries() =
279287
databaseTest {
280-
val syncStream = database.syncStream()
281-
database.connectInternal(syncStream, 1000L)
288+
database.connect(connector)
282289

283290
turbineScope(timeout = 10.0.seconds) {
284291
val turbine = database.currentStatus.asFlow().testIn(this)
@@ -324,7 +331,7 @@ class SyncIntegrationTest {
324331

325332
turbineScope(timeout = 10.0.seconds) {
326333
// Connect the first database
327-
database.connect(connector, 1000L)
334+
database.connect(connector)
328335
db2.connect(connector)
329336

330337
waitFor {
@@ -350,7 +357,7 @@ class SyncIntegrationTest {
350357
val turbine2 = db2.currentStatus.asFlow().testIn(this)
351358

352359
// Connect the first database
353-
database.connect(connector, 1000L)
360+
database.connect(connector)
354361

355362
turbine1.waitFor { it.connecting }
356363
db2.connect(connector)
@@ -414,7 +421,7 @@ class SyncIntegrationTest {
414421
databaseTest {
415422
val testConnector = TestConnector()
416423
connector = testConnector
417-
database.connectInternal(database.syncStream(), 1000L)
424+
database.connect(testConnector)
418425

419426
suspend fun expectUserRows(amount: Int) {
420427
val row = database.get("SELECT COUNT(*) FROM users") { it.getLong(0)!! }

‎core/src/commonIntegrationTest/kotlin/com/powersync/testutils/TestUtils.kt‎

Lines changed: 24 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -7,23 +7,23 @@ import co.touchlab.kermit.Severity
77
import co.touchlab.kermit.TestConfig
88
import co.touchlab.kermit.TestLogWriter
99
import com.powersync.DatabaseDriverFactory
10-
import com.powersync.PowerSyncDatabase
1110
import com.powersync.bucket.WriteCheckpointData
1211
import com.powersync.bucket.WriteCheckpointResponse
1312
import com.powersync.connectors.PowerSyncBackendConnector
1413
import com.powersync.connectors.PowerSyncCredentials
14+
import com.powersync.createPowerSyncDatabaseImpl
1515
import com.powersync.db.PowerSyncDatabaseImpl
1616
import com.powersync.db.schema.Schema
1717
import com.powersync.sync.SyncLine
18-
import com.powersync.sync.SyncStream
1918
import dev.mokkery.answering.returns
2019
import dev.mokkery.everySuspend
2120
import dev.mokkery.mock
21+
import io.ktor.client.HttpClient
22+
import io.ktor.client.HttpClientConfig
2223
import kotlinx.coroutines.channels.Channel
2324
import kotlinx.coroutines.test.TestScope
2425
import kotlinx.coroutines.test.runTest
2526
import kotlinx.io.files.Path
26-
import kotlinx.serialization.json.JsonObject
2727

2828
expect val factory: DatabaseDriverFactory
2929

@@ -43,18 +43,22 @@ fun generatePrintLogWriter() =
4343
}
4444
}
4545

46-
internal fun databaseTest(testBody: suspend ActiveDatabaseTest.() -> Unit) =
47-
runTest {
48-
val running = ActiveDatabaseTest(this)
46+
internal fun databaseTest(
47+
createInitialDatabase: Boolean = true,
48+
testBody: suspend ActiveDatabaseTest.() -> Unit,
49+
) = runTest {
50+
val running = ActiveDatabaseTest(this)
51+
if (createInitialDatabase) {
4952
// Make sure the database is initialized, we're using internal APIs that expect initialization.
5053
running.database = running.openDatabaseAndInitialize()
54+
}
5155

52-
try {
53-
running.testBody()
54-
} finally {
55-
running.cleanup()
56-
}
56+
try {
57+
running.testBody()
58+
} finally {
59+
running.cleanup()
5760
}
61+
}
5862

5963
@OptIn(ExperimentalKermitApi::class)
6064
internal class ActiveDatabaseTest(
@@ -104,32 +108,27 @@ internal class ActiveDatabaseTest(
104108
fun openDatabase(): PowerSyncDatabaseImpl {
105109
logger.d { "Opening database $databaseName in directory $testDirectory" }
106110
val db =
107-
PowerSyncDatabase(
111+
createPowerSyncDatabaseImpl(
108112
factory = factory,
109113
schema = Schema(UserRow.table),
110114
dbFilename = databaseName,
111115
dbDirectory = testDirectory,
112116
logger = logger,
113117
scope = scope,
118+
createClient = ::createClient,
114119
)
115120
doOnCleanup { db.close() }
116-
return db as PowerSyncDatabaseImpl
121+
return db
117122
}
118123

119124
suspend fun openDatabaseAndInitialize(): PowerSyncDatabaseImpl = openDatabase().also { it.readLock { } }
120125

121-
fun PowerSyncDatabase.syncStream(): SyncStream {
122-
val client = MockSyncService(syncLines) { checkpointResponse() }
123-
return SyncStream(
124-
bucketStorage = database.bucketStorage,
125-
connector = connector,
126-
httpEngine = client,
127-
uploadCrud = { connector.uploadData(this) },
128-
retryDelayMs = 10,
129-
logger = logger,
130-
params = JsonObject(emptyMap()),
131-
scope = scope,
132-
)
126+
private fun createClient(config: HttpClientConfig<*>.() -> Unit): HttpClient {
127+
val engine = MockSyncService(syncLines) { checkpointResponse() }
128+
129+
return HttpClient(engine) {
130+
config()
131+
}
133132
}
134133

135134
fun doOnCleanup(action: suspend () -> Unit) {

‎core/src/commonMain/kotlin/com/powersync/PowerSyncDatabaseFactory.kt‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,10 @@ import co.touchlab.kermit.Logger
44
import co.touchlab.skie.configuration.annotations.DefaultArgumentInterop
55
import com.powersync.db.PowerSyncDatabaseImpl
66
import com.powersync.db.schema.Schema
7+
import com.powersync.sync.SyncStream
78
import com.powersync.utils.generateLogger
9+
import io.ktor.client.HttpClient
10+
import io.ktor.client.HttpClientConfig
811
import kotlinx.coroutines.CoroutineScope
912
import kotlinx.coroutines.DelicateCoroutinesApi
1013
import kotlinx.coroutines.GlobalScope
@@ -30,7 +33,7 @@ public fun PowerSyncDatabase(
3033
): PowerSyncDatabase {
3134
val generatedLogger: Logger = generateLogger(logger)
3235

33-
return PowerSyncDatabaseImpl(
36+
return createPowerSyncDatabaseImpl(
3437
schema = schema,
3538
factory = factory,
3639
dbFilename = dbFilename,
@@ -39,3 +42,22 @@ public fun PowerSyncDatabase(
3942
dbDirectory = dbDirectory,
4043
)
4144
}
45+
46+
internal fun createPowerSyncDatabaseImpl(
47+
factory: DatabaseDriverFactory,
48+
schema: Schema,
49+
dbFilename: String,
50+
scope: CoroutineScope,
51+
logger: Logger,
52+
dbDirectory: String?,
53+
createClient: (HttpClientConfig<*>.() -> Unit) -> HttpClient = SyncStream::defaultHttpClient,
54+
): PowerSyncDatabaseImpl =
55+
PowerSyncDatabaseImpl(
56+
schema = schema,
57+
factory = factory,
58+
dbFilename = dbFilename,
59+
scope = scope,
60+
logger = logger,
61+
dbDirectory = dbDirectory,
62+
createClient = createClient,
63+
)

‎core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt‎

Lines changed: 21 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ import com.powersync.utils.JsonParam
2424
import com.powersync.utils.JsonUtil
2525
import com.powersync.utils.throttle
2626
import com.powersync.utils.toJsonObject
27+
import io.ktor.client.HttpClient
28+
import io.ktor.client.HttpClientConfig
2729
import kotlinx.coroutines.CancellationException
2830
import kotlinx.coroutines.CoroutineScope
2931
import kotlinx.coroutines.FlowPreview
@@ -61,6 +63,7 @@ internal class PowerSyncDatabaseImpl(
6163
private val dbFilename: String,
6264
private val dbDirectory: String? = null,
6365
val logger: Logger = Logger,
66+
private val createClient: (HttpClientConfig<*>.() -> Unit) -> HttpClient,
6467
) : PowerSyncDatabase {
6568
companion object {
6669
internal val streamConflictMessage =
@@ -148,22 +151,25 @@ internal class PowerSyncDatabaseImpl(
148151
crudThrottleMs: Long,
149152
retryDelayMs: Long,
150153
params: Map<String, JsonParam?>,
151-
) = mutex.withLock {
154+
) {
152155
waitReady()
153-
disconnectInternal()
154-
155-
connectInternal(
156-
SyncStream(
157-
bucketStorage = bucketStorage,
158-
connector = connector,
159-
uploadCrud = suspend { connector.uploadData(this) },
160-
retryDelayMs = retryDelayMs,
161-
logger = logger,
162-
params = params.toJsonObject(),
163-
scope = scope,
164-
),
165-
crudThrottleMs,
166-
)
156+
mutex.withLock {
157+
disconnectInternal()
158+
159+
connectInternal(
160+
SyncStream(
161+
bucketStorage = bucketStorage,
162+
connector = connector,
163+
uploadCrud = suspend { connector.uploadData(this) },
164+
retryDelayMs = retryDelayMs,
165+
logger = logger,
166+
params = params.toJsonObject(),
167+
scope = scope,
168+
createClient = createClient,
169+
),
170+
crudThrottleMs,
171+
)
172+
}
167173
}
168174

169175
@OptIn(FlowPreview::class)

0 commit comments

Comments
 (0)