commit c453b695724f3a066690b2cc94f9d590f29110e9
parent 2f0903fb0c92a316e625f035dca2e3ff9c3408df
Author: Florian Dold <dold@taler.net>
Date: Fri, 11 Sep 2026 09:47:00 +0200
database: close notification connections on retry and shutdown
Diffstat:
4 files changed, 94 insertions(+), 22 deletions(-)
diff --git a/libeufin-bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt b/libeufin-bank/src/main/kotlin/tech/libeufin/bank/db/Database.kt
@@ -56,7 +56,7 @@ class Database(
// Withdrawal confirmation flow, the key is the public withdrawal UUID
private val withdrawalFlow = ConcurrentHashMap<UUID, CountedSharedFlow<WithdrawalStatus>>()
- init {
+ private val notificationWatcher =
watchNotifications(pgSource, "libeufin_bank", LoggerFactory.getLogger("libeufin-bank-db-watcher"), mapOf(
"bank_tx" to {
val (debtor, creditor, debitRow, creditRow) = it.split(' ', limit = 4).map { it.toLong() }
@@ -91,6 +91,13 @@ class Database(
}
}
))
+
+ override fun close() {
+ try {
+ notificationWatcher.close()
+ } finally {
+ super.close()
+ }
}
/** Listen for new bank transactions for [account] */
@@ -164,4 +171,4 @@ enum class AbortResult {
Success,
UnknownOperation,
AlreadyConfirmed
-}
-\ No newline at end of file
+}
diff --git a/libeufin-bank/src/test/kotlin/DatabaseTest.kt b/libeufin-bank/src/test/kotlin/DatabaseTest.kt
@@ -20,6 +20,7 @@
import io.ktor.http.*
import kotlinx.coroutines.*
import org.junit.Test
+import org.slf4j.LoggerFactory
import tech.libeufin.bank.*
import tech.libeufin.bank.db.AccountDAO.AccountCreationResult
import tech.libeufin.bank.db.TanDAO.*
@@ -36,8 +37,65 @@ import java.util.concurrent.TimeUnit
import kotlin.test.assertEquals
import kotlin.test.assertIs
import kotlin.test.assertNull
+import kotlin.test.assertNotEquals
class DatabaseTest {
+
+ @Test
+ fun notificationConnectionsClose() = setup { db, cfg ->
+ val source = pgDataSource(cfg.dbCfg.dbConnStr)
+ val applicationName = "notification-test-${UUID.randomUUID()}"
+ source.applicationName = applicationName
+ val firstNotification = CompletableDeferred<Unit>()
+ val secondNotification = CompletableDeferred<Unit>()
+ var reconnect = true
+
+ db.conn { observer ->
+ fun watcherPids(): List<Int> = observer.prepareStatement(
+ "SELECT pid FROM pg_stat_activity WHERE application_name = ? AND query = 'LISTEN watcher_test' AND state = 'idle'"
+ ).use { stmt ->
+ stmt.setString(1, applicationName)
+ stmt.executeQuery().use { rows ->
+ buildList { while (rows.next()) add(rows.getInt(1)) }
+ }
+ }
+
+ val watcher = watchNotifications(
+ source, "libeufin_bank", LoggerFactory.getLogger("notification-test"),
+ mapOf("watcher_test" to {
+ if (reconnect) {
+ reconnect = false
+ firstNotification.complete(Unit)
+ // Exercise cleanup on the retry path as well as on close().
+ throw IllegalStateException("test notification reconnect")
+ }
+ secondNotification.complete(Unit)
+ })
+ )
+ watcher.use {
+ val firstPid = withTimeout(5000) {
+ while (watcherPids().isEmpty()) delay(10)
+ watcherPids().single()
+ }
+ observer.execSQLUpdate("NOTIFY watcher_test, 'first'")
+ withTimeout(5000) { firstNotification.await() }
+ val secondPid = withTimeout(5000) {
+ while (true) {
+ val pids = watcherPids()
+ if (pids.size == 1 && pids.single() != firstPid) break
+ delay(10)
+ }
+ watcherPids().single()
+ }
+ assertNotEquals(firstPid, secondPid)
+ observer.execSQLUpdate("NOTIFY watcher_test, 'second'")
+ withTimeout(5000) { secondNotification.await() }
+ }
+ withTimeout(5000) {
+ while (watcherPids().isNotEmpty()) delay(10)
+ }
+ }
+ }
// Testing the helper that creates the admin account.
@Test
diff --git a/libeufin-common/src/main/kotlin/db/notifications.kt b/libeufin-common/src/main/kotlin/db/notifications.kt
@@ -19,14 +19,14 @@
package tech.libeufin.common.db
-import kotlinx.coroutines.delay
+import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.runBlocking
import org.postgresql.ds.PGSimpleDataSource
import org.slf4j.Logger
import tech.libeufin.common.ExpoBackoffDecorr
import tech.libeufin.common.fmtLog
+import java.io.Closeable
import java.util.concurrent.ConcurrentHashMap
// SharedFlow that are manually counted for manual garbage collection
@@ -40,14 +40,13 @@ fun watchNotifications(
schema: String,
logger: Logger,
listeners: Map<String, (suspend (String) -> Unit)>
-) {
+): Closeable {
val backoff = ExpoBackoffDecorr()
- // Run notification logic in a separated thread
- kotlin.concurrent.thread(isDaemon = true) {
- runBlocking {
- while (true) {
- try {
- val conn = pgSource.pgConnection(schema)
+ // JDBC notification reads block, so keep them off the caller's dispatcher.
+ val job = CoroutineScope(Dispatchers.IO).launch {
+ while (isActive) {
+ try {
+ pgSource.pgConnection(schema).use { conn ->
// Listen to all notifications channels
for (channel in listeners.keys) {
@@ -56,24 +55,28 @@ fun watchNotifications(
backoff.reset()
- while (true) {
- conn.getNotifications(0) // Block until we receive at least one notification
+ while (isActive) {
+ // Bound the blocking read so close() can cancel an idle watcher.
+ conn.getNotifications(1000)
.forEach {
// Dispatch
try {
listeners[it.name]!!(it.parameter)
} catch (e: Exception) {
+ ensureActive()
throw Exception("channel ${it.name} with input '${it.parameter}'", e)
}
}
}
- } catch (e: Exception) {
- e.fmtLog(logger)
- delay(backoff.next())
}
+ } catch (e: Exception) {
+ ensureActive()
+ e.fmtLog(logger)
+ delay(backoff.next())
}
}
}
+ return Closeable { runBlocking { job.cancelAndJoin() } }
}
/** Listen to flow from [map] for [key] using [lambda]*/
@@ -95,4 +98,4 @@ suspend fun <R, K, V> listen(map: ConcurrentHashMap<K, CountedSharedFlow<V>>, ke
if (v.count > 0) v else null
}
}
-}
-\ No newline at end of file
+}
diff --git a/libeufin-nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt b/libeufin-nexus/src/main/kotlin/tech/libeufin/nexus/db/Database.kt
@@ -100,7 +100,7 @@ class Database(dbConfig: DatabaseConfig, val currency: String): DbPool(dbConfig,
private val incomingTxFlows: MutableSharedFlow<Long> = MutableSharedFlow()
private val revenueTxFlows: MutableSharedFlow<Long> = MutableSharedFlow()
- init {
+ private val notificationWatcher =
watchNotifications(pgSource, "libeufin_nexus", LoggerFactory.getLogger("libeufin-nexus-db-watcher"), mapOf(
"nexus_revenue_tx" to {
val id = it.toLong()
@@ -115,6 +115,13 @@ class Database(dbConfig: DatabaseConfig, val currency: String): DbPool(dbConfig,
incomingTxFlows.emit(id)
}
))
+
+ override fun close() {
+ try {
+ notificationWatcher.close()
+ } finally {
+ super.close()
+ }
}
/** Listen for new taler outgoing transactions */
@@ -126,4 +133,4 @@ class Database(dbConfig: DatabaseConfig, val currency: String): DbPool(dbConfig,
/** Listen for new incoming transactions */
suspend fun <R> listenRevenue(lambda: suspend (Flow<Long>) -> R): R
= lambda(revenueTxFlows)
-}
-\ No newline at end of file
+}
