fix(storage-ksqlite): deadlock in ConversationStore.delete/rename
ci / JVM build + tests (push) Failing after 1m16s
ci / JVM build + tests (push) Failing after 1m16s
mutex в kotlinx.coroutines НЕ reentrant — при вызове get() изнутри withLock получаем deadlock. ConversationStore.delete() и rename() использовали именно этот паттерн для проверки существования. Fixed: заменил на raw SELECT 1 FROM conversation WHERE id=? и SELECT updated_at FROM conversation WHERE id=? — те же проверки, без повторного взятия mutex. Discovered by full-test-suite run: 3 tests in KsqliteConversationStoreTest были в UncompletedCoroutinesError (UncompletedCoroutinesError после 1 минуты ожидания), хотя отдельный прогон EventStore (где нет вызовов get() внутри mutex) проходил. После фикса все 36 тестов проходят: - KsqliteEventStoreTest (11) - KsqliteConversationStoreTest (9) - KsqliteMessageStoreTest (5) - KsqliteWorkingMemoryStoreTest (5) - KsqliteReflectionStoreTest (6)
This commit is contained in:
+13
-3
@@ -70,8 +70,12 @@ class KsqliteConversationStore(
|
|||||||
|
|
||||||
override suspend fun delete(id: String): Boolean = withContext(Dispatchers.Default) {
|
override suspend fun delete(id: String): Boolean = withContext(Dispatchers.Default) {
|
||||||
mutex.withLock {
|
mutex.withLock {
|
||||||
val existed = get(id) != null
|
// Проверяем существование через raw query, НЕ через get() — get() тоже
|
||||||
if (!existed) return@withContext false
|
// берёт mutex (не реентрант), что привело бы к deadlock.
|
||||||
|
connection.prepare("SELECT 1 FROM conversation WHERE id = ?").use { check ->
|
||||||
|
check.bindText(1, id)
|
||||||
|
check.executeQuery().use { rs -> if (!rs.next()) return@withContext false }
|
||||||
|
}
|
||||||
messageStore?.clear(id)
|
messageStore?.clear(id)
|
||||||
workingMemoryStore?.clear(id)
|
workingMemoryStore?.clear(id)
|
||||||
connection.prepare("DELETE FROM conversation WHERE id = ?").use { stmt ->
|
connection.prepare("DELETE FROM conversation WHERE id = ?").use { stmt ->
|
||||||
@@ -108,7 +112,13 @@ class KsqliteConversationStore(
|
|||||||
stmt.bindText(3, id)
|
stmt.bindText(3, id)
|
||||||
stmt.executeUpdate()
|
stmt.executeUpdate()
|
||||||
}
|
}
|
||||||
get(id)?.updatedAt
|
// raw query instead of get() (deadlock — get() also takes mutex)
|
||||||
|
connection.prepare("SELECT updated_at FROM conversation WHERE id = ?").use { stmt ->
|
||||||
|
stmt.bindText(1, id)
|
||||||
|
stmt.executeQuery().use { rs ->
|
||||||
|
if (rs.next()) Instant.fromEpochMilliseconds(rs.getLong(0)!!) else null
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user