Add PostgresLockProvider implementation and related tests for KrescentLock in new package

Author: helightdev

krescent-exposed/build.gradle.kts

@@ -8,9 +8,9 @@ plugins {
 
 dependencies {
     // Apply the kotlinx bundle of dependencies from the version catalog (`gradle/libs.versions.toml`).
-    implementation(libs.bundles.exposed)
     implementation(libs.bundles.kotlinxEcosystem)
     implementation(project(":krescent-core"))
+    api(libs.bundles.exposed)
 
     testImplementation(kotlin("test"))
     testImplementation(project(":krescent-test"))

krescent-postgres/build.gradle.kts

@@ -0,0 +1,21 @@
+plugins {
+    // Apply the shared build logic from a convention plugin.
+    // The shared code is located in `buildSrc/src/main/kotlin/kotlin-jvm.gradle.kts`.
+    id("buildsrc.convention.kotlin-jvm")
+    // Apply Kotlin Serialization plugin from `gradle/libs.versions.toml`.
+    alias(libs.plugins.kotlinPluginSerialization)
+}
+
+dependencies {
+    // Apply the kotlinx bundle of dependencies from the version catalog (`gradle/libs.versions.toml`).
+    implementation(libs.bundles.exposed)
+    implementation(libs.bundles.kotlinxEcosystem)
+    implementation(project(":krescent-core"))
+    implementation(project(":krescent-exposed"))
+    implementation(libs.postgresql)
+
+    testImplementation(kotlin("test"))
+    testImplementation(project(":krescent-test"))
+    testImplementation(libs.bundles.testContainers)
+
+}

krescent-postgres/src/main/kotlin/dev/helight/krescent/postgres/PostgresLock.kt

@@ -0,0 +1,87 @@
+package dev.helight.krescent.postgres
+
+import dev.helight.krescent.synchronization.KrescentLock
+import dev.helight.krescent.synchronization.KrescentLockProvider
+import kotlinx.coroutines.*
+import kotlinx.coroutines.selects.onTimeout
+import kotlinx.coroutines.selects.select
+import org.jetbrains.exposed.sql.Database
+import org.jetbrains.exposed.sql.transactions.experimental.newSuspendedTransaction
+import java.nio.ByteBuffer
+import java.security.MessageDigest
+import java.util.*
+import kotlin.time.Duration
+
+class PostgresLockProvider(
+    val database: Database,
+    val leaseTime: Duration = Duration.INFINITE,
+) : KrescentLockProvider {
+
+    override suspend fun getLock(identity: String): KrescentLock {
+        return PostgresLock(listOf(identity))
+    }
+
+    override suspend fun getMultiLock(identities: Collection<String>): KrescentLock {
+        return PostgresLock(identities)
+    }
+
+    private inner class PostgresLock(
+        val identities: Collection<String>,
+    ) : KrescentLock {
+
+        private val open = Stack<CompletableDeferred<Unit>>()
+
+        override suspend fun acquire(timeout: Duration?) {
+            val releaser = CompletableDeferred<Unit>()
+            launchPostgresLockJob(database, identities, releaser, timeout ?: Duration.INFINITE, leaseTime).await()
+            open.push(releaser)
+        }
+
+        override suspend fun release() {
+            val releaser = open.pop() ?: return
+            releaser.complete(Unit)
+        }
+    }
+}
+
+private fun hashKey(key: String): Long {
+    val bytes = key.toByteArray()
+    val digest = MessageDigest.getInstance("MD5").digest(bytes)
+    return ByteBuffer.wrap(digest.sliceArray(0..7)).long
+}
+
+@OptIn(ExperimentalCoroutinesApi::class)
+private fun launchPostgresLockJob(
+    database: Database,
+    identities: Collection<String>,
+    releaser: CompletableDeferred<Unit>,
+    timeout: Duration,
+    leaseTime: Duration,
+): Deferred<Unit> {
+    val started = CompletableDeferred<Unit>()
+    CoroutineScope(Dispatchers.IO).launch {
+        newSuspendedTransaction(db = database) {
+            try {
+                val setLockTimeout = when (timeout) {
+                    Duration.INFINITE -> "SET LOCAL lock_timeout = 0"
+                    else -> "SET LOCAL lock_timeout = ${timeout.inWholeMilliseconds}"
+                }
+                exec(
+                    identities.joinToString(
+                        prefix = "DO $$ BEGIN $setLockTimeout; ",
+                        separator = ";",
+                        postfix = "; END $$"
+                    ) { "PERFORM pg_advisory_xact_lock(${hashKey(it)})" })
+                started.complete(Unit)
+            } catch (e: Exception) {
+                started.completeExceptionally(e)
+                return@newSuspendedTransaction
+            }
+            select {
+                releaser.onAwait {}
+                onTimeout(leaseTime) {}
+            }
+        }
+    }
+    return started
+}

krescent-postgres/src/test/kotlin/dev/helight/krescent/postgres/PostgresLock.kt

@@ -0,0 +1,42 @@
+package dev.helight.krescent.postgres
+
+import dev.helight.krescent.synchronization.KrescentLockProvider
+import dev.helight.krescent.test.KrescentLockProviderContract
+import org.jetbrains.exposed.sql.Database
+import org.testcontainers.containers.GenericContainer
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.time.Duration
+
+@Testcontainers
+class PostgresLockTest : KrescentLockProviderContract {
+
+    companion object {
+
+        @Container
+        private val postgres = GenericContainer("postgres:latest")
+            .withExposedPorts(5432)
+            .withEnv("POSTGRES_USER", "root")
+            .withEnv("POSTGRES_PASSWORD", "example")
+            .withStartupTimeout(Duration.ofSeconds(60))
+
+        private val connectionString get() = "jdbc:postgresql://${postgres.host}:${postgres.getMappedPort(5432)}/root"
+
+        fun connect(): Database = Database.connect(
+            url = connectionString,
+            driver = "org.postgresql.Driver",
+            user = "root",
+            password = "example"
+        )
+    }
+
+    override val latency: Long
+        get() = 50L
+
+    override fun getProvider(): KrescentLockProvider {
+
+        return PostgresLockProvider(
+            database = connect()
+        )
+    }
+}

settings.gradle.kts

@@ -23,6 +23,7 @@ include(":krescent-core")
 include(":krescent-kurrent")
 include(":krescent-mongo")
 include(":krescent-exposed")
+include(":krescent-postgres")
 include(":krescent-redisson")
 
 include(":krescent-test")