Enhance KrescentLockProviderContract tests for lock behavior and timeout handling

Author: helightdev

krescent-redisson/src/main/kotlin/dev/helight/krescent/redisson/RedisLockProvider.kt

@@ -6,6 +6,7 @@ import kotlinx.coroutines.future.asDeferred
 import kotlinx.coroutines.withTimeout
 import org.redisson.api.RLock
 import org.redisson.api.RedissonClient
+import java.util.*
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicLong
 import kotlin.time.Duration
@@ -20,22 +21,23 @@ class RedisLockProvider(
 
     override suspend fun getLock(identity: String): KrescentLock {
         val rlock = client.getLock(prefix + identity)
-        return RedisLockImpl(rlock, lockIdCounter.getAndIncrement())
+        return RedisLockImpl(rlock)
     }
 
     override suspend fun getMultiLock(identities: Collection<String>): KrescentLock {
         val lockList = identities.map { identity ->
             client.getLock(prefix + identity)
         }.toTypedArray()
         val multiLock = client.getMultiLock(*lockList)
-        return RedisLockImpl(multiLock, lockIdCounter.getAndIncrement())
+        return RedisLockImpl(multiLock)
     }
 
     private inner class RedisLockImpl(
         val lock: RLock,
-        val lockId: Long,
     ) : KrescentLock {
 
+        private val claims: Stack<Long> = Stack()
+
         override suspend fun acquire(timeout: Duration?) {
             if (timeout != null) {
                 withTimeout(timeout) { acquire() }
@@ -45,27 +47,20 @@ class RedisLockProvider(
         }
 
         private suspend fun acquire() {
+            val claimId = lockIdCounter.getAndIncrement()
             lock.lockAsync(
                 when {
                     leaseTime.isInfinite() -> -1L
                     else -> leaseTime.inWholeMilliseconds
-                }, TimeUnit.MILLISECONDS, lockId
+                }, TimeUnit.MILLISECONDS, claimId
             ).asDeferred().await()
+            claims.push(claimId)
         }
 
         override suspend fun release() {
-            lock.unlockAsync(lockId).asDeferred().await()
-        }
-
-        override fun equals(other: Any?): Boolean {
-            if (this === other) return true
-            if (javaClass != other?.javaClass) return false
-            other as RedisLockImpl
-            return lockId == other.lockId
-        }
-
-        override fun hashCode(): Int {
-            return lockId.hashCode()
+            if (claims.isEmpty()) return
+            val claimId = claims.pop()
+            lock.unlockAsync(claimId).asDeferred().await()
         }
     }
 }

krescent-redisson/src/test/kotlin/dev/helight/krescent/redisson/RedisLockProvider.kt

@@ -28,6 +28,9 @@ class RedisLockProviderTest : KrescentLockProviderContract {
         })
     }
 
+    override val latency: Long
+        get() = 100L
+
     override fun getProvider(): KrescentLockProvider {
         val db = connect()
         val uid = UUID.randomUUID().toString()

krescent-test/src/main/kotlin/dev/helight/krescent/test/KrescentLockProviderContract.kt

@@ -2,40 +2,71 @@ package dev.helight.krescent.test
 
 import dev.helight.krescent.synchronization.KrescentLockProvider
 import dev.helight.krescent.synchronization.KrescentLockProvider.Extensions.getMultiLock
-import dev.helight.krescent.synchronization.LocalLockProvider
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.Test
 import kotlin.test.assertTrue
+import kotlin.time.DurationUnit
+import kotlin.time.toDuration
 
 interface KrescentLockProviderContract {
 
     fun getProvider(): KrescentLockProvider
+    val latency: Long
+        get() = 50L
 
     @Test
     fun `Acquired locks work as expected`() = runBlocking {
         val provider = getProvider()
-        val timeStart = System.currentTimeMillis()
+        System.currentTimeMillis()
+        val resource = LockResource()
         listOf(
             async {
                 val lock = provider.getLock("test-identity")
                 lock.runGuarded {
-                    delay(50)
+                    resource.claim()
+                    delay(latency)
+                    resource.release()
                 }
             },
             async {
                 val lock = provider.getLock("test-identity")
                 lock.runGuarded {
-                    delay(50)
+                    resource.claim()
+                    delay(latency)
+                    resource.release()
                 }
             }
         ).awaitAll()
-        val timeAfter = System.currentTimeMillis()
-        assertTrue(timeAfter - timeStart >= 95)
+        return@runBlocking
     }
 
+    @Test
+    fun `Reused lock instance still works and is not reentrant`() = runBlocking {
+        val provider = getProvider()
+        val resource = LockResource()
+        val lock = provider.getLock("test-identity")
+        listOf(
+            async {
+                lock.runGuarded {
+                    resource.claim()
+                    delay(latency)
+                    resource.release()
+                }
+            },
+            async {
+                lock.runGuarded {
+                    resource.claim()
+                    delay(latency)
+                    resource.release()
+                }
+            }
+        ).awaitAll()
+        return@runBlocking
+    }
+
+
     @Test
     fun `Locks do not conflict with each other`() = runBlocking {
         val provider = getProvider()
@@ -44,51 +75,96 @@ interface KrescentLockProviderContract {
             async {
                 val lock = provider.getLock("test-identity")
                 lock.runGuarded {
-                    delay(50)
+                    delay(latency)
                 }
             },
             async {
                 val lock = provider.getLock("another-identity")
                 lock.runGuarded {
-                    delay(50)
+                    delay(latency)
                 }
-            }
+            },
         ).awaitAll()
         val timeAfter = System.currentTimeMillis()
-        assertTrue(timeAfter - timeStart < 95)
+        assertTrue(timeAfter - timeStart < latency * 2)
+    }
+
+    @Test
+    fun `Test lock timeouts`() = runBlocking {
+        val provider = getProvider()
+        val lock = provider.getLock("timeout-identity")
+        lock.acquire()
+        var timeoutThrown = false
+        withTimeout(1000L) {
+            try {
+                lock.acquire(latency.toDuration(DurationUnit.MILLISECONDS))
+            } catch (e: Exception) {
+                timeoutThrown = true
+            }
+            println("Acquired lock after timeout period")
+        }
+        assertTrue(timeoutThrown, "Timeout was not thrown when expected")
     }
 
     @Test
     fun `Multilocks work and do not overblock`() = runBlocking {
-        val provider = LocalLockProvider()
-        val timeStart = System.currentTimeMillis()
+        val provider = getProvider()
+        val aRes = LockResource()
+        val bRes = LockResource()
+        val cRes = LockResource()
+        val dRes = LockResource()
         listOf(
             async {
                 val lock = provider.getMultiLock("a", "b")
                 lock.runGuarded {
-                    delay(50)
+                    aRes.claim()
+                    bRes.claim()
+                    delay(latency)
+                    bRes.release()
+                    aRes.release()
                 }
             },
             async {
                 val lock = provider.getMultiLock("a", "c")
                 lock.runGuarded {
-                    delay(50)
+                    aRes.claim()
+                    cRes.claim()
+                    delay(latency)
+                    cRes.release()
+                    aRes.release()
                 }
             },
             async {
                 val lock = provider.getMultiLock("d", "b")
                 lock.runGuarded {
-                    delay(50)
+                    dRes.claim()
+                    bRes.claim()
+                    delay(latency)
+                    bRes.release()
+                    dRes.release()
                 }
             },
             async {
                 val lock = provider.getMultiLock("x", "y")
                 lock.runGuarded {
-                    delay(50)
+                    delay(latency)
                 }
             },
         ).awaitAll()
-        val timeAfter = System.currentTimeMillis()
-        assertTrue(timeAfter - timeStart >= 95 && timeAfter - timeStart <= 150)
+        return@runBlocking
+    }
+
+    class LockResource {
+        private val atomic: AtomicInteger = AtomicInteger(1)
+
+        fun claim() {
+            val v = atomic.decrementAndGet()
+            assertEquals(v, 0, "Resource is already claimed!")
+        }
+
+        fun release() {
+            val v = atomic.incrementAndGet()
+            assertEquals(v, 1, "Resource was not claimed!")
+        }
     }
 }